app/security.py

"""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)