Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| """Signed passes from the EchoDepth app, required on every request except health checks. | |
| WHY. The Space has to stay reachable from the browser (a video upload is too big to route through | |
| the app's serverless functions), so the address alone must not be enough to use it. The app mints | |
| a short-lived pass for signed-in users and for its own server calls (src/lib/video-service-token.ts); | |
| this checks it. | |
| FORMAT. base64url(JSON {exp, t}) + "." + base64url(HMAC-SHA256 of the first part), keyed by the | |
| VIDEO_SERVICE_SECRET Space secret, which must match the app's. | |
| ROLLOUT. With no VIDEO_SERVICE_SECRET set, requests are let through and /health says | |
| "auth": "off", so the code can be deployed before the secret is added without breaking uploads. | |
| Setting the secret turns enforcement on at the next restart. | |
| """ | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import json | |
| import os | |
| import time | |
| from fastapi import Request | |
| from fastapi.responses import JSONResponse | |
| HEADER = "x-service-token" | |
| SECRET = os.environ.get("VIDEO_SERVICE_SECRET", "").encode() | |
| OPEN_PATHS = {"/", "/health"} | |
| # Browsers may only call from the app itself; server-to-server calls send no Origin. | |
| ALLOWED_ORIGINS = [ | |
| o.strip() | |
| for o in os.environ.get( | |
| "ALLOWED_ORIGINS", | |
| "https://www.cavefishlabs.co.uk,https://cavefishlabs.co.uk,http://localhost:3000", | |
| ).split(",") | |
| if o.strip() | |
| ] | |
| def enforcing() -> bool: | |
| return bool(SECRET) | |
| def _b64decode(s: str) -> bytes: | |
| return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) | |
| def valid(token: str) -> bool: | |
| try: | |
| payload, sig = token.split(".", 1) | |
| want = hmac.new(SECRET, payload.encode(), hashlib.sha256).digest() | |
| if not hmac.compare_digest(want, _b64decode(sig)): | |
| return False | |
| return int(json.loads(_b64decode(payload))["exp"]) >= time.time() | |
| except Exception: | |
| return False | |
| async def require_pass(request: Request, call_next): | |
| """FastAPI HTTP middleware. CORS preflights pass through to the CORS middleware.""" | |
| if not enforcing() or request.method == "OPTIONS" or request.url.path in OPEN_PATHS: | |
| return await call_next(request) | |
| if not valid(request.headers.get(HEADER, "")): | |
| return JSONResponse({"detail": "Not authorised"}, status_code=401) | |
| return await call_next(request) | |