Spaces:
Running
Running
| import logging | |
| import firebase_admin | |
| from firebase_admin import auth, credentials as firebase_credentials | |
| from fastapi import HTTPException, status, Security | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from app.core.settings import settings | |
| logger = logging.getLogger(__name__) | |
| import json | |
| # Initialize Firebase Admin | |
| _firebase_initialized = False | |
| try: | |
| if settings.FIREBASE_CREDENTIALS_JSON: | |
| cred_dict = json.loads(settings.FIREBASE_CREDENTIALS_JSON) | |
| cred = firebase_credentials.Certificate(cred_dict) | |
| firebase_admin.initialize_app(cred) | |
| _firebase_initialized = True | |
| logger.info("Firebase Admin initialized successfully from JSON string.") | |
| elif settings.FIREBASE_CREDENTIALS_PATH: | |
| cred = firebase_credentials.Certificate(settings.FIREBASE_CREDENTIALS_PATH) | |
| firebase_admin.initialize_app(cred) | |
| _firebase_initialized = True | |
| logger.info("Firebase Admin initialized successfully from file path.") | |
| else: | |
| # Try default/env initialization | |
| firebase_admin.initialize_app() | |
| _firebase_initialized = True | |
| logger.info("Firebase Admin initialized using default credentials.") | |
| except Exception as e: | |
| logger.warning(f"Firebase Admin initialization skipped or failed: {e}") | |
| security = HTTPBearer() | |
| def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)): | |
| """ | |
| Verifies the Firebase ID Token. | |
| Returns the decoded token dict if valid. | |
| """ | |
| token = credentials.credentials | |
| # ----------------------------------------------------- | |
| # MOCK AUTH FOR DEVELOPMENT | |
| # If ENABLE_AUTH is False or token starts with "mock_" AND we are in development | |
| # ----------------------------------------------------- | |
| is_dev = settings.ENV == "development" | |
| if not settings.ENABLE_AUTH or (token.startswith("mock_") and is_dev): | |
| logger.info(f"Using MOCK AUTH for token: {token}") | |
| return {"uid": "dev_user_123", "email": "dev@mathminds.ai"} | |
| if not _firebase_initialized: | |
| logger.error("Attempted to verify token but Firebase is not initialized.") | |
| raise HTTPException( | |
| status_code=status.HTTP_503_SERVICE_UNAVAILABLE, | |
| detail="Authentication service unavailable", | |
| ) | |
| try: | |
| # Verify the ID token from Firebase | |
| decoded_token = auth.verify_id_token(token) | |
| return { | |
| "uid": decoded_token.get("uid"), | |
| "email": decoded_token.get("email") | |
| } | |
| except Exception as e: | |
| logger.warning(f"Firebase token verification failed: {e}") | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid or expired authentication credentials", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| def get_current_user(token: dict = Security(verify_token)): | |
| """ | |
| Dependency to get the current user from the token. | |
| """ | |
| return token | |