Spaces:
Running on Zero
Running on Zero
| """App-layer API key for the THOX Space fleet. | |
| Why app-layer and not Hugging Face's edge gate | |
| ---------------------------------------------- | |
| Flipping a Space to private makes HF answer **404** to an unauthenticated caller, which | |
| reads as "the app is broken" rather than "you are not signed in", and the edge gate eats | |
| the `Authorization` header of anything proxying through it. Checking the key in the app | |
| keeps the Space public — so `/health`, the Gradio UI and the model card stay reachable — | |
| while `/v1/*` behaves the way every OpenAI client already expects: `Authorization: Bearer | |
| <key>`, and a real `401` with a `WWW-Authenticate` header when it is wrong. | |
| Fail-closed | |
| ----------- | |
| If `THOX_API_KEY` is unset the endpoint refuses everything with 503 rather than serving | |
| openly. An unset secret is a deployment mistake, and the failure mode of guessing wrong | |
| here is a public, uncapped inference endpoint — so the safe direction is down. | |
| `/health` deliberately stays open: liveness has to be checkable without a credential. | |
| It must therefore never echo the key back — see `redact()`. | |
| """ | |
| from __future__ import annotations | |
| import hmac | |
| import os | |
| from fastapi.responses import JSONResponse | |
| ENV_VAR = "THOX_API_KEY" | |
| def configured_key() -> str: | |
| return (os.environ.get(ENV_VAR) or "").strip() | |
| def is_configured() -> bool: | |
| return bool(configured_key()) | |
| def redact(text: str) -> str: | |
| """Strip the key out of anything that might reach a log or an open endpoint.""" | |
| key = configured_key() | |
| if key and key in text: | |
| return text.replace(key, "***redacted***") | |
| return text | |
| def _deny(message: str, code: str, status: int = 401) -> JSONResponse: | |
| return JSONResponse( | |
| status_code=status, | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| content={"error": {"message": message, "type": "invalid_request_error", | |
| "param": None, "code": code}}, | |
| ) | |
| def presented_key(request) -> str: | |
| """Read the key the caller sent. `Authorization: Bearer` first, `X-API-Key` second.""" | |
| auth = request.headers.get("authorization") or "" | |
| if auth[:7].lower() == "bearer ": | |
| return auth[7:].strip() | |
| return (request.headers.get("x-api-key") or "").strip() | |
| def check(request): | |
| """Return None when the caller may proceed, or the JSONResponse to send back.""" | |
| expected = configured_key() | |
| if not expected: | |
| return _deny( | |
| f"This endpoint is not configured: {ENV_VAR} is unset on the Space. " | |
| "Set it as a Space secret.", | |
| "no_api_key_configured", | |
| 503, | |
| ) | |
| presented = presented_key(request) | |
| if not presented: | |
| return _deny( | |
| "Missing API key. Send 'Authorization: Bearer <key>'.", | |
| "missing_api_key", | |
| ) | |
| # Constant time: a short-circuiting == leaks the length of the matching prefix. | |
| if not hmac.compare_digest(presented, expected): | |
| return _deny("Incorrect API key provided.", "invalid_api_key") | |
| return None | |
| # -------------------------------------------------------------------------------------- | |
| # OpenAI-shaped validation errors | |
| # -------------------------------------------------------------------------------------- | |
| def openai_validation_handler(request, exc): | |
| """FastAPI answers a bad body with `422 {"detail": [...]}`; OpenAI uses | |
| `400 {"error": {...}}`. A client that reads `error.message` sees nothing in the | |
| former, so translate.""" | |
| try: | |
| first = exc.errors()[0] | |
| where = ".".join(str(p) for p in first.get("loc", ()) if p != "body") | |
| message = f"{first.get('msg', 'Invalid request')}" + (f" (at '{where}')" if where else "") | |
| param = where or None | |
| except Exception: # noqa: BLE001 | |
| message, param = "Invalid request body.", None | |
| return JSONResponse( | |
| status_code=400, | |
| content={"error": {"message": message, "type": "invalid_request_error", | |
| "param": param, "code": None}}, | |
| ) | |
| def install(app) -> None: | |
| """Attach the handler to the app that will actually serve the request. | |
| This has to be Gradio's app, not the FastAPI instance the routes were declared on: | |
| Starlette resolves exception handlers against the app whose middleware stack is | |
| running, and the routes are grafted onto Gradio's router. Clearing | |
| `middleware_stack` forces a rebuild so a handler added after `launch()` is picked up | |
| -- adding to the dict alone would be silently ignored, because the stack that closed | |
| over the old handler map is already built. | |
| """ | |
| from fastapi.exceptions import RequestValidationError | |
| app.add_exception_handler(RequestValidationError, openai_validation_handler) | |
| app.middleware_stack = None | |
| print("[thox] installed OpenAI-shaped validation error handler", flush=True) | |