Spaces:
Sleeping
Sleeping
File size: 4,772 Bytes
d3295cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """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
|