""" Server-side authentication & authorization for admin/protected routes. The single source of truth for who may call /api/admin/* lives here, not in the frontend. localStorage and client-side email filters are advisory only; these dependencies are the actual gate. Configuration (set as Hugging Face Space secrets): FIREBASE_SERVICE_ACCOUNT -> full JSON contents of the Firebase service account private key (NOT a VITE_ value) ADMIN_EMAILS -> comma-separated admin emails SHEET_REFRESH_TOKEN -> (optional) static token the Apps Script onEdit trigger uses to call refresh-sheet """ import os import json import logging import firebase_admin from firebase_admin import credentials, auth as fb_auth from fastapi import Depends, HTTPException, status, Header from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from typing import Optional _log = logging.getLogger("auth") # --------------------------------------------------------------------------- # Firebase Admin init — verifies ID tokens against Google's signing keys. # --------------------------------------------------------------------------- _FIREBASE_READY = False if not firebase_admin._apps: _sa_raw = os.environ.get("FIREBASE_SERVICE_ACCOUNT") if _sa_raw: try: firebase_admin.initialize_app( credentials.Certificate(json.loads(_sa_raw)) ) _FIREBASE_READY = True except Exception as e: # malformed secret — fail closed, log loudly _log.error("[auth] FIREBASE_SERVICE_ACCOUNT invalid: %s", e) else: _log.error( "[auth] FIREBASE_SERVICE_ACCOUNT not set — admin routes will 503" ) else: _FIREBASE_READY = True ADMIN_EMAILS = { e.strip().lower() for e in os.environ.get("ADMIN_EMAILS", "").split(",") if e.strip() } ALLOWED_DOMAINS = ("@alustudent.com", "@alueducation.com") # auto_error=False so we can return clean 503/401 instead of FastAPI's default bearer = HTTPBearer(auto_error=False) def _verify(cred: Optional[HTTPAuthorizationCredentials]) -> dict: """Verify a Firebase ID token and return its decoded claims.""" if not _FIREBASE_READY: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, "Auth not configured on the server", ) if cred is None or not cred.credentials: raise HTTPException( status.HTTP_401_UNAUTHORIZED, "Missing bearer token" ) try: return fb_auth.verify_id_token(cred.credentials) except Exception: raise HTTPException( status.HTTP_401_UNAUTHORIZED, "Invalid or expired token" ) async def require_user( cred: Optional[HTTPAuthorizationCredentials] = Depends(bearer), ) -> dict: """Any authenticated ALU user. Enforces the signup allowlist server-side.""" user = _verify(cred) email = (user.get("email") or "").lower() if not email.endswith(ALLOWED_DOMAINS): raise HTTPException(status.HTTP_403_FORBIDDEN, "ALU email required") return user async def require_admin(user: dict = Depends(require_user)) -> dict: """An ALU user whose email is in ADMIN_EMAILS.""" if (user.get("email") or "").lower() not in ADMIN_EMAILS: raise HTTPException( status.HTTP_403_FORBIDDEN, "Admin access required" ) return user # --------------------------------------------------------------------------- # Dual gate for routes that must also be callable by non-interactive automation # (the Google Sheet onEdit trigger, which cannot perform a Firebase login). # Accepts EITHER a valid Firebase admin token OR the static refresh token. # --------------------------------------------------------------------------- _SHEET_REFRESH_TOKEN = os.environ.get("SHEET_REFRESH_TOKEN", "") async def require_admin_or_token( cred: Optional[HTTPAuthorizationCredentials] = Depends(bearer), x_refresh_token: Optional[str] = Header(default=None, alias="X-Refresh-Token"), ) -> dict: # Path 1: the Apps Script automation presents the static token. if _SHEET_REFRESH_TOKEN and (x_refresh_token or "").strip() == _SHEET_REFRESH_TOKEN: return {"email": "apps-script@automation", "via": "refresh_token"} # Path 2: a logged-in human admin presents a Firebase token. if cred is not None and cred.credentials: user = _verify(cred) email = (user.get("email") or "").lower() if email.endswith(ALLOWED_DOMAINS) and email in ADMIN_EMAILS: return user raise HTTPException(status.HTTP_403_FORBIDDEN, "Admin access required") raise HTTPException( status.HTTP_401_UNAUTHORIZED, "Provide a valid admin token or X-Refresh-Token", )