Spaces:
Running on Zero
Running on Zero
| """DecisionLab request security: body-size limit and response headers. | |
| A plain ASGI middleware with no web-framework imports, so every rule is testable anywhere. | |
| - No token and no login anywhere (operator ruling 2026-09-28, for the Hugging Face Space): the page and /api/* are | |
| open to whoever can reach the app. The remaining guards are request limits and response headers. | |
| - /api/* request bodies over max_body bytes get 413, whether the size is declared or streamed. | |
| - The lab's own responses (/, /static/*, /api/*) carry SECURITY_HEADERS. The page may be framed only by | |
| huggingface.co (a Space is shown inside a frame there), so there is no X-Frame-Options header. | |
| - /gradio/* belongs to the mounted Gradio app, which sets its own headers and handles its own uploads. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import threading | |
| GRADIO_PREFIX = "/gradio" | |
| SECURITY_HEADERS = { | |
| "content-security-policy": ( | |
| "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; " | |
| "font-src 'self'; object-src 'none'; base-uri 'none'; form-action 'none'; " | |
| "frame-ancestors 'self' https://huggingface.co https://*.hf.space" | |
| ), | |
| "x-content-type-options": "nosniff", | |
| "referrer-policy": "no-referrer", | |
| "cross-origin-resource-policy": "same-origin", | |
| "permissions-policy": "camera=(), microphone=(), geolocation=(), payment=(), usb=()", | |
| } | |
| class Gate: | |
| """A non-blocking counter: try_enter() admits up to `size` holders at once, then refuses.""" | |
| def __init__(self, size: int): | |
| self._sem = threading.BoundedSemaphore(size) | |
| def try_enter(self) -> bool: | |
| return self._sem.acquire(blocking=False) | |
| def leave(self) -> None: | |
| self._sem.release() | |
| def _header(scope, name: bytes) -> str: | |
| for k, v in scope.get("headers") or []: | |
| if k.lower() == name: | |
| return v.decode("latin-1") | |
| return "" | |
| class SecurityMiddleware: | |
| def __init__(self, app, max_body: int): | |
| self.app, self.max_body = app, max_body | |
| async def __call__(self, scope, receive, send): | |
| if scope["type"] != "http": | |
| return await self.app(scope, receive, send) | |
| path = scope["path"] | |
| if path == GRADIO_PREFIX or path.startswith(GRADIO_PREFIX + "/"): | |
| return await self.app(scope, receive, send) | |
| is_api = path.startswith("/api/") | |
| async def send_secured(message): | |
| if message["type"] == "http.response.start": | |
| drop = set(SECURITY_HEADERS) | {"cache-control", "x-frame-options"} | |
| headers = [(k, v) for k, v in message.get("headers", []) if k.decode("latin-1").lower() not in drop] | |
| headers += [(k.encode(), v.encode()) for k, v in SECURITY_HEADERS.items()] | |
| # API answers are never stored; the page and its files are revalidated on every load. | |
| headers.append((b"cache-control", b"no-store" if is_api else b"no-cache")) | |
| message = dict(message, headers=headers) | |
| await send(message) | |
| async def reply(status: int, detail: str): | |
| body = json.dumps({"detail": detail}).encode() | |
| await send_secured({"type": "http.response.start", "status": status, | |
| "headers": [(b"content-type", b"application/json")]}) | |
| await send_secured({"type": "http.response.body", "body": body}) | |
| if is_api and scope.get("method") in ("POST", "PUT", "PATCH"): | |
| too_big = f"Request body is over the {self.max_body}-byte limit." | |
| declared = _header(scope, b"content-length") | |
| if declared.isdigit() and int(declared) > self.max_body: | |
| return await reply(413, too_big) | |
| chunks, size = [], 0 | |
| while True: | |
| msg = await receive() | |
| if msg["type"] != "http.request": | |
| break | |
| size += len(msg.get("body", b"")) | |
| if size > self.max_body: | |
| return await reply(413, too_big) | |
| chunks.append(msg.get("body", b"")) | |
| if not msg.get("more_body"): | |
| break | |
| replay = [{"type": "http.request", "body": b"".join(chunks), "more_body": False}] | |
| async def receive_replay(): | |
| return replay.pop(0) if replay else await receive() | |
| return await self.app(scope, receive_replay, send_secured) | |
| return await self.app(scope, receive, send_secured) | |