import json import os from typing import Optional from fastapi import HTTPException, Request try: import firebase_admin from firebase_admin import app_check, credentials except ImportError: # pragma: no cover - handled via requirements firebase_admin = None # type: ignore app_check = None # type: ignore credentials = None # type: ignore FIREBASE_APP_CHECK_ENABLED = os.getenv("FIREBASE_APP_CHECK_ENABLED", "0") == "1" ALLOWED_APP_IDS = [ value.strip() for value in os.getenv("FIREBASE_ALLOWED_APP_IDS", "").split(",") if value.strip() ] firebase_app: Optional["firebase_admin.App"] = None def _initialize_firebase_app() -> None: global firebase_app, FIREBASE_APP_CHECK_ENABLED if not FIREBASE_APP_CHECK_ENABLED: return if firebase_admin is None or credentials is None: print("[Firebase] firebase_admin package not available; disabling App Check verification.") FIREBASE_APP_CHECK_ENABLED = False return if firebase_admin._apps: # type: ignore[attr-defined] firebase_app = firebase_admin.get_app() return try: service_account_json = os.getenv("FIREBASE_SERVICE_ACCOUNT_JSON") if service_account_json: cred_info = json.loads(service_account_json) cred = credentials.Certificate(cred_info) else: service_account_path = os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH") if not service_account_path: raise ValueError( "FIREBASE_SERVICE_ACCOUNT_JSON or FIREBASE_SERVICE_ACCOUNT_PATH must be provided when App Check is enabled." ) cred = credentials.Certificate(service_account_path) firebase_app = firebase_admin.initialize_app(cred) print("[Firebase] Firebase App initialized for App Check verification.") except Exception as exc: print(f"[Firebase] Failed to initialize App Check verification: {exc}") FIREBASE_APP_CHECK_ENABLED = False _initialize_firebase_app() async def verify_app_check_token(request: Request) -> Optional[dict]: """ FastAPI dependency to verify Firebase App Check tokens. Returns decoded claims on success or raises HTTPException on failure. """ if not FIREBASE_APP_CHECK_ENABLED: return None if firebase_app is None or app_check is None: raise HTTPException(status_code=500, detail="Firebase App Check misconfigured") token = request.headers.get("X-Firebase-AppCheck") if not token: raise HTTPException(status_code=401, detail="Missing Firebase App Check token") try: claims = app_check.verify_token(token, app=firebase_app) except Exception as exc: # pragma: no cover - SDK handles errors raise HTTPException(status_code=401, detail=f"Invalid Firebase App Check token: {exc}") from exc app_id = claims.get("app_id") if ALLOWED_APP_IDS and app_id not in ALLOWED_APP_IDS: raise HTTPException(status_code=403, detail="Firebase App ID not allowed") # Store claims on request state for downstream handlers if needed request.state.firebase_app_check = claims return claims