Spaces:
Running
Running
| """What every page of both apps (the explorer and its admin) gets: security headers, a localhost-only rule when run | |
| locally, same-origin state changes, and pages that revalidate so a deploy is never half-cached.""" | |
| from __future__ import annotations | |
| import os | |
| import threading | |
| import time | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import JSONResponse | |
| from . import config | |
| LOCAL_HOSTS = {"localhost", "127.0.0.1", "[::1]", "::1"} | {h.strip() for h in os.environ.get("RLX_ALLOWED_HOSTS", "").split(",") if h.strip()} | |
| CSP = ("default-src 'self'; script-src 'self' https://cdn.jsdelivr.net; style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdn.jsdelivr.net; " | |
| "font-src https://fonts.gstatic.com https://cdn.jsdelivr.net; img-src 'self' data: blob: https:; media-src 'self' data: blob: https:; connect-src 'self'; frame-src 'self' https://*.hf.space; object-src 'none'; " | |
| "base-uri 'none'; form-action 'self'; frame-ancestors 'self' https://huggingface.co https://*.hf.space") | |
| def client_ip(request: Request) -> str: | |
| """The visitor's address: the last X-Forwarded-For entry is the one the Space's proxy added (earlier ones are | |
| whatever the client claimed).""" | |
| fwd = [p.strip() for p in request.headers.get("x-forwarded-for", "").split(",") if p.strip()] | |
| return fwd[-1] if fwd else (request.client.host if request.client else "?") | |
| class Limiter: | |
| """At most `n` hits per `window` seconds for one key (an address, an account): a sliding window, in memory, bounded | |
| so a flood of distinct keys can't grow it without end. Per replica, which is what a Space runs.""" | |
| def __init__(self, n: int, window: float, keys: int = 20000): | |
| self.n, self.window, self.keys = n, window, keys | |
| self._hits: dict[str, list[float]] = {} | |
| self._lock = threading.Lock() | |
| def hit(self, key: str) -> bool: | |
| """Count one hit; False when `key` is over its limit.""" | |
| now = time.time() | |
| with self._lock: | |
| hits = [t for t in self._hits.get(key, ()) if now - t < self.window] | |
| if len(hits) >= self.n: | |
| self._hits[key] = hits | |
| return False | |
| hits.append(now) | |
| self._hits[key] = hits | |
| if len(self._hits) > self.keys: | |
| for k in [k for k, v in self._hits.items() if not v or now - v[-1] >= self.window] or list(self._hits)[: self.keys // 5]: | |
| self._hits.pop(k, None) | |
| return True | |
| def same_origin(request: Request) -> bool: | |
| origin = request.headers.get("origin") | |
| host = request.headers.get("x-forwarded-host") or request.headers.get("host") | |
| return not origin or origin.split("://", 1)[-1] == host | |
| def install(app: FastAPI) -> None: | |
| async def _revalidate(request: Request, call_next): | |
| """Code and styles revalidate on every load (cheap: ETag -> 304), so a deploy is never half-cached.""" | |
| resp = await call_next(request) | |
| if not request.url.path.startswith("/api/"): | |
| resp.headers.setdefault("Cache-Control", "no-cache") | |
| return resp | |
| async def _security(request: Request, call_next): | |
| """Locally, only answer to localhost: a web page could otherwise DNS-rebind its own name to 127.0.0.1 and drive | |
| this server (and the machine's HF token) from the browser. Everywhere: the usual security headers.""" | |
| if config.LOCAL_MODE: | |
| raw = request.headers.get("host") or "" | |
| host = raw.split("]")[0] + "]" if raw.startswith("[") else raw.rsplit(":", 1)[0] | |
| if host not in LOCAL_HOSTS: | |
| return JSONResponse({"detail": "host not allowed (set RLX_ALLOWED_HOSTS to serve under another name)"}, 403) | |
| resp = await call_next(request) | |
| resp.headers.setdefault("X-Content-Type-Options", "nosniff") | |
| resp.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin") | |
| resp.headers.setdefault("Permissions-Policy", "camera=(), microphone=(), geolocation=(), payment=()") | |
| if not request.url.path.startswith("/api/"): | |
| resp.headers.setdefault("Content-Security-Policy", CSP) | |
| return resp | |
| async def _same_origin_posts(request: Request, call_next): | |
| """The session cookie is SameSite=None (it has to work in the huggingface.co iframe), so a state-changing | |
| request must prove it came from this page: another site could otherwise start rollouts billed to you.""" | |
| if request.method not in ("GET", "HEAD", "OPTIONS") and request.url.path.startswith("/api/"): | |
| origin = request.headers.get("origin") | |
| host = request.headers.get("x-forwarded-host") or request.headers.get("host") | |
| if origin and origin.split("://", 1)[-1] != host: | |
| return JSONResponse({"detail": "cross-site request refused"}, 403) | |
| if not origin and request.headers.get("sec-fetch-site") not in (None, "same-origin", "none"): | |
| return JSONResponse({"detail": "cross-site request refused"}, 403) | |
| return await call_next(request) | |