feat: simple shared-token api auth
Adds a shared-token auth dependency for the viewing API (Task 2.9). Trust model: when API_TOKEN is empty (the default) auth is disabled — we trust the network boundary (local/intranet tool). When set, every request must present the token via `Authorization: Bearer <token>` or `?token=<token>`, else 401. `/api/health` is exempt so liveness probes work without the token. - app/config.py: add `api_token: str = ""` to Settings. - app/api/auth.py: `require_auth` dependency (secrets.compare_digest, reads get_settings() per-request so token rotation via cache_clear() works). - app/api/main.py: apply the dependency globally; document the trust model. - .env.example: document API_TOKEN. - tests/test_api_auth.py: 8 tests covering disabled/correct/missing/wrong/ malformed-header/query-param/health-exempt paths. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
"""Simple shared-token auth for the yunke data-mirror API (Task 2.9).
|
||||
|
||||
Trust model (per design doc §8 + plan Task 2.9):
|
||||
|
||||
* **No token configured** (``Settings.api_token`` empty, the default) -> auth
|
||||
is DISABLED. This is a local/intranet tool; when no token is set we trust the
|
||||
network boundary and every request passes through. This keeps local dev and
|
||||
single-box deployments zero-config.
|
||||
* **Token configured** -> every request must present the same token via either
|
||||
the ``Authorization: Bearer <token>`` header or the ``?token=<token>`` query
|
||||
parameter. Missing / mismatched tokens get ``401 Unauthorized``.
|
||||
|
||||
The dependency is applied globally on the FastAPI app (see
|
||||
:mod:`app.api.main`). ``GET /api/health`` is exempt so liveness probes can
|
||||
reach it without knowing the token — that exemption is encoded here in
|
||||
:data:`_PUBLIC_PATHS` so the rule lives with the auth logic itself.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
|
||||
from fastapi import HTTPException, Request, status
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
# Request paths that bypass auth even when a token is configured. Keep this
|
||||
# minimal so the protected surface is easy to reason about — currently just the
|
||||
# liveness probe.
|
||||
_PUBLIC_PATHS = frozenset({"/api/health"})
|
||||
|
||||
|
||||
async def require_auth(request: Request) -> None:
|
||||
"""No-op when ``api_token`` is unset; otherwise enforce a shared bearer token.
|
||||
|
||||
Settings are read on every call (``get_settings`` is ``lru_cache``-ed, so
|
||||
this is cheap) so rotating the token — env change plus
|
||||
``get_settings.cache_clear()`` — takes effect without a process restart.
|
||||
The comparison uses :func:`secrets.compare_digest` on UTF-8 bytes to avoid
|
||||
a timing side channel and to tolerate non-ASCII token values.
|
||||
"""
|
||||
settings = get_settings()
|
||||
expected = settings.api_token
|
||||
if not expected:
|
||||
# No token configured -> trust the network boundary.
|
||||
return
|
||||
|
||||
if request.url.path in _PUBLIC_PATHS:
|
||||
return
|
||||
|
||||
presented = _extract_token(request)
|
||||
if presented is None or not secrets.compare_digest(
|
||||
presented.encode("utf-8"), expected.encode("utf-8")
|
||||
):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="invalid or missing API token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
|
||||
def _extract_token(request: Request) -> str | None:
|
||||
"""Pull the presented token from the ``Authorization`` header or ``?token=``.
|
||||
|
||||
Header wins when present and well-formed (``Bearer <value>``); otherwise we
|
||||
fall back to the ``token`` query parameter, which is convenient for
|
||||
browser-visible endpoints like media streaming.
|
||||
"""
|
||||
auth_header = request.headers.get("Authorization")
|
||||
if auth_header:
|
||||
scheme, _, value = auth_header.partition(" ")
|
||||
if scheme.lower() == "bearer" and value:
|
||||
return value
|
||||
return request.query_params.get("token")
|
||||
Reference in New Issue
Block a user