"""FastAPI dependencies: tenant authentication and admin gating.""" from __future__ import annotations from fastapi import Depends, Header, HTTPException, status from backend.api import security from backend.config import settings def get_current_tenant(authorization: str = Header(default="")) -> str: """Resolve the tenant id from a Bearer JWT.""" if not authorization.lower().startswith("bearer "): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) token = authorization.split(" ", 1)[1].strip() try: payload = security.decode_token(token) except Exception as exc: # noqa: BLE001 raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=f"Invalid token: {exc}", headers={"WWW-Authenticate": "Bearer"}, ) from exc tenant_id = payload.get("sub") if not tenant_id: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Token missing subject" ) from backend.storage import tenant_store tenant_store.ensure_tenant_schema(tenant_id) return tenant_id def require_admin( tenant_id: str = Depends(get_current_tenant), x_admin_token: str = Header(default=""), ) -> str: """Require both a valid JWT and the configured admin token, with the master upload feature flag enabled.""" if not settings.master_template_upload_enabled: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Admin master-template override is disabled (master_template_upload_enabled=false).", ) if not settings.admin_token or x_admin_token != settings.admin_token: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid admin token" ) return tenant_id