Spaces:
Sleeping
Sleeping
| import base64 | |
| import json | |
| import os | |
| from pathlib import Path | |
| import firebase_admin | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer | |
| from firebase_admin import auth, credentials | |
| from logger import logger | |
| _BASE_DIR = Path(__file__).parent.parent.parent | |
| def _get_sa_key() -> str | None: | |
| sa_key = os.getenv("FIREBASE_SA_KEY") | |
| if sa_key: | |
| return sa_key | |
| env_local = _BASE_DIR / ".env.local" | |
| if env_local.exists(): | |
| for line in env_local.read_text().splitlines(): | |
| if line.startswith("FIREBASE_SA_KEY="): | |
| return line.partition("=")[2].strip() | |
| return None | |
| def initialize_firebase(): | |
| try: | |
| sa_key = _get_sa_key() | |
| if sa_key: | |
| cred = credentials.Certificate(json.loads(base64.b64decode(sa_key))) | |
| firebase_admin.initialize_app(cred) | |
| else: | |
| firebase_admin.initialize_app() | |
| print("Firebase app initialized successfully.") | |
| except Exception as e: | |
| print(f"Failed to initialize Firebase: {e}") | |
| def get_user_token( | |
| credential: HTTPAuthorizationCredentials = Depends( | |
| HTTPBearer(auto_error=False)), | |
| ): | |
| if credential is None: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Authorization credentials not provided", | |
| headers={"WWW-Authenticate": 'Bearer error="invalid_token"'}, | |
| ) | |
| try: | |
| decoded_token = auth.verify_id_token( | |
| credential.credentials, clock_skew_seconds=60) | |
| logger.info("Token verified successfully.") | |
| user_id = decoded_token.get("user_id") | |
| return user_id | |
| except Exception as err: | |
| logger.error(f"Token verification failed: {err}") | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail=f"Invalid User. {err}", | |
| headers={"WWW-Authenticate": 'Bearer error="invalid_token"'}, | |
| ) | |