Spaces:
Running
Running
| """ | |
| 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) | |