Spaces:
Sleeping
Sleeping
| """Admin session cookies, CSRF tokens, rate limiting and password checks.""" | |
| from __future__ import annotations | |
| import hmac | |
| import ipaddress | |
| import secrets | |
| import threading | |
| import time | |
| from collections import defaultdict, deque | |
| from typing import Optional | |
| from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer | |
| SESSION_COOKIE = "curas_admin_session" | |
| class SessionManager: | |
| def __init__(self, secret: str, max_age_seconds: int): | |
| self._s = URLSafeTimedSerializer(secret, salt="curas-admin-session") | |
| self.max_age = max_age_seconds | |
| def create(self) -> tuple[str, str]: | |
| """Return (cookie_value, csrf_token) for a fresh admin session.""" | |
| csrf = secrets.token_urlsafe(32) | |
| return self._s.dumps({"csrf": csrf, "v": 1}), csrf | |
| def read(self, cookie_value: Optional[str]) -> Optional[dict]: | |
| if not cookie_value: | |
| return None | |
| try: | |
| data = self._s.loads(cookie_value, max_age=self.max_age) | |
| except (BadSignature, SignatureExpired): | |
| return None | |
| if not isinstance(data, dict) or "csrf" not in data: | |
| return None | |
| return data | |
| class FormTokens: | |
| """Signed, time-limited tokens for the anonymous public form. | |
| The token is carried only in a hidden input (no cookie), so the form also | |
| works when the site is shown inside another site's frame, such as the | |
| Space page on huggingface.co. The form is anonymous, so there is no | |
| session for a cross-site request to ride on; the token's job is to | |
| accept posts only from a form this server rendered recently. | |
| """ | |
| def __init__(self, secret: str, max_age_seconds: int = 6 * 3600): | |
| self._s = URLSafeTimedSerializer(secret, salt="curas-form-token") | |
| self.max_age = max_age_seconds | |
| def issue(self) -> str: | |
| return self._s.dumps({"n": secrets.token_urlsafe(16)}) | |
| def verify(self, form_value: Optional[str]) -> bool: | |
| if not form_value: | |
| return False | |
| try: | |
| data = self._s.loads(str(form_value), max_age=self.max_age) | |
| except (BadSignature, SignatureExpired): | |
| return False | |
| return isinstance(data, dict) and "n" in data | |
| class RateLimiter: | |
| """In-memory sliding-window limiter. Adequate for a single-instance Space.""" | |
| def __init__(self): | |
| self._events: dict[tuple[str, str], deque] = defaultdict(deque) | |
| self._lock = threading.Lock() | |
| def hit(self, bucket: str, key: str, limit: int, window_seconds: int) -> bool: | |
| """Record an event; return True if still within the limit.""" | |
| now = time.monotonic() | |
| with self._lock: | |
| q = self._events[(bucket, key)] | |
| while q and now - q[0] > window_seconds: | |
| q.popleft() | |
| if len(q) >= limit: | |
| return False | |
| q.append(now) | |
| if len(self._events) > 10000: | |
| # Drop stale buckets so memory cannot grow without bound. | |
| for k in [k for k, v in self._events.items() if not v or now - v[-1] > window_seconds]: | |
| del self._events[k] | |
| return True | |
| def is_limited(self, bucket: str, key: str, limit: int, window_seconds: int) -> bool: | |
| now = time.monotonic() | |
| with self._lock: | |
| q = self._events.get((bucket, key)) | |
| if not q: | |
| return False | |
| while q and now - q[0] > window_seconds: | |
| q.popleft() | |
| return len(q) >= limit | |
| def reset(self, bucket: str, key: str) -> None: | |
| with self._lock: | |
| self._events.pop((bucket, key), None) | |
| def password_matches(expected: str, supplied: str) -> bool: | |
| if not expected or supplied is None: | |
| return False | |
| return hmac.compare_digest(expected.encode("utf-8"), supplied.encode("utf-8")) | |
| def client_ip(headers, fallback: str) -> str: | |
| """Client address as seen by the proxy in front of the app. | |
| Proxies append the real peer address to whatever X-Forwarded-For the | |
| client sent (Hugging Face Spaces does this, one hop), so the LAST entry is | |
| the trustworthy one; the first entry can be chosen by the client. | |
| """ | |
| xff = headers.get("x-forwarded-for") | |
| host = fallback or "unknown" | |
| if xff: | |
| hosts = [h.strip() for h in xff.split(",") if h.strip()] | |
| if hosts: | |
| host = hosts[-1] | |
| return rate_limit_key(host) | |
| def rate_limit_key(host: str) -> str: | |
| """IPv6 clients are keyed by their /64 so one network cannot mint endless addresses.""" | |
| try: | |
| ip = ipaddress.ip_address(host) | |
| except ValueError: | |
| return host | |
| if ip.version == 6: | |
| return str(ipaddress.ip_network((ip, 64), strict=False)) | |
| return host | |