90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
|
"""Operator token auth for activity-core ops console (ACTIVITY-WP-0024).
|
||
|
|
|
||
|
|
Mutations under ``/ops`` are fail-closed:
|
||
|
|
- If ``ACTIVITY_CORE_OPERATOR_TOKEN`` is set, requests must send matching
|
||
|
|
``X-Operator-Token`` (or ``Authorization: Bearer <token>``).
|
||
|
|
- If the token is **unset**, mutations are refused unless
|
||
|
|
``ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS`` is truthy (local dev only).
|
||
|
|
|
||
|
|
Read endpoints do not require the token (ClusterIP / port-forward posture).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hmac
|
||
|
|
import os
|
||
|
|
from typing import Annotated
|
||
|
|
|
||
|
|
from fastapi import Header, HTTPException, Request
|
||
|
|
|
||
|
|
OPERATOR_TOKEN_ENV = "ACTIVITY_CORE_OPERATOR_TOKEN"
|
||
|
|
ALLOW_UNAUTH_ENV = "ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS"
|
||
|
|
HEADER_NAME = "X-Operator-Token"
|
||
|
|
|
||
|
|
|
||
|
|
def operator_token_configured() -> bool:
|
||
|
|
return bool((os.environ.get(OPERATOR_TOKEN_ENV) or "").strip())
|
||
|
|
|
||
|
|
|
||
|
|
def allow_unauth_mutations() -> bool:
|
||
|
|
return (os.environ.get(ALLOW_UNAUTH_ENV) or "").strip().lower() in {
|
||
|
|
"1",
|
||
|
|
"true",
|
||
|
|
"yes",
|
||
|
|
"on",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def extract_operator_token(
|
||
|
|
*,
|
||
|
|
x_operator_token: str | None = None,
|
||
|
|
authorization: str | None = None,
|
||
|
|
) -> str | None:
|
||
|
|
if x_operator_token and x_operator_token.strip():
|
||
|
|
return x_operator_token.strip()
|
||
|
|
if authorization and authorization.lower().startswith("bearer "):
|
||
|
|
return authorization[7:].strip() or None
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def verify_operator_token(provided: str | None) -> str:
|
||
|
|
"""Return operator principal label or raise HTTPException."""
|
||
|
|
expected = (os.environ.get(OPERATOR_TOKEN_ENV) or "").strip()
|
||
|
|
if not expected:
|
||
|
|
if allow_unauth_mutations():
|
||
|
|
return "anonymous-dev"
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=403,
|
||
|
|
detail=(
|
||
|
|
"operator auth not configured; set "
|
||
|
|
f"{OPERATOR_TOKEN_ENV} or enable {ALLOW_UNAUTH_ENV} for local dev"
|
||
|
|
),
|
||
|
|
)
|
||
|
|
if not provided:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=401,
|
||
|
|
detail=f"missing operator token ({HEADER_NAME} or Authorization Bearer)",
|
||
|
|
)
|
||
|
|
if not hmac.compare_digest(provided, expected):
|
||
|
|
raise HTTPException(status_code=401, detail="invalid operator token")
|
||
|
|
return "operator"
|
||
|
|
|
||
|
|
|
||
|
|
async def require_operator(
|
||
|
|
request: Request,
|
||
|
|
x_operator_token: Annotated[str | None, Header(alias=HEADER_NAME)] = None,
|
||
|
|
authorization: Annotated[str | None, Header()] = None,
|
||
|
|
) -> str:
|
||
|
|
"""FastAPI dependency: require valid operator token for mutations."""
|
||
|
|
# Prefer dependency headers; fall back to raw request (HTML form headers rare).
|
||
|
|
provided = extract_operator_token(
|
||
|
|
x_operator_token=x_operator_token,
|
||
|
|
authorization=authorization,
|
||
|
|
)
|
||
|
|
if provided is None:
|
||
|
|
provided = extract_operator_token(
|
||
|
|
x_operator_token=request.headers.get(HEADER_NAME),
|
||
|
|
authorization=request.headers.get("Authorization"),
|
||
|
|
)
|
||
|
|
return verify_operator_token(provided)
|