""" Firebase Authentication Middleware for OptiQ API. Verifies Firebase JWT tokens and extracts user info. """ from __future__ import annotations import os from typing import Optional from functools import wraps from fastapi import Request, HTTPException, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials # Mock Firebase verification for development # In production, use firebase_admin to verify tokens USE_MOCK_AUTH = os.environ.get("OPTIQ_MOCK_AUTH", "true").lower() == "true" security = HTTPBearer(auto_error=False) class FirebaseUser: """Represents an authenticated Firebase user.""" def __init__(self, uid: str, email: str, name: Optional[str] = None): self.uid = uid self.email = email self.name = name or email.split("@")[0] async def verify_firebase_token( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> Optional[FirebaseUser]: """Verify Firebase JWT token and return user info. In mock mode (development), accepts any token and extracts user info from it. In production mode, verifies the token with Firebase Admin SDK. """ if credentials is None: return None token = credentials.credentials if USE_MOCK_AUTH: # Development mode: parse mock token or accept any token try: # Try to parse as "uid:email:name" format for testing parts = token.split(":") if len(parts) >= 2: return FirebaseUser(uid=parts[0], email=parts[1], name=parts[2] if len(parts) > 2 else None) else: # Accept any token as demo user return FirebaseUser( uid="demo_user_" + token[:8] if len(token) >= 8 else "demo_user", email="demo@optiq.ai", name="Demo User" ) except Exception: return FirebaseUser(uid="demo_user", email="demo@optiq.ai", name="Demo User") else: # Production mode: verify with Firebase Admin SDK try: import firebase_admin from firebase_admin import auth, credentials as fb_credentials # Initialize Firebase Admin if not already done if not firebase_admin._apps: cred_path = os.environ.get("FIREBASE_CREDENTIALS") if cred_path: cred = fb_credentials.Certificate(cred_path) firebase_admin.initialize_app(cred) else: firebase_admin.initialize_app() decoded_token = auth.verify_id_token(token) return FirebaseUser( uid=decoded_token["uid"], email=decoded_token.get("email", ""), name=decoded_token.get("name"), ) except Exception as e: raise HTTPException(status_code=401, detail=f"Invalid authentication token: {str(e)}") async def require_auth( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> FirebaseUser: """Require authentication - raises 401 if not authenticated.""" user = await verify_firebase_token(credentials) if user is None: raise HTTPException(status_code=401, detail="Authentication required") return user async def optional_auth( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> Optional[FirebaseUser]: """Optional authentication - returns None if not authenticated.""" return await verify_firebase_token(credentials)