feat: add Firebase Auth ID token verification + /api/me endpoint; support static password + Firebase ID tokens

api_server.py

@@ -15,7 +15,7 @@ from pathlib import Path
 from fastapi.responses import FileResponse, StreamingResponse, JSONResponse
 import logging
 from logging.handlers import BufferingHandler
-from firebase_app_check import verify_app_check_token
+from firebase_app_check import verify_app_check_token, verify_firebase_id_token
 
 # Import face swap functionality
 import sys
@@ -31,14 +31,29 @@ API_PASSWORD = os.getenv("API_PASSWORD", "logicgo_videoswap@153")
 security = HTTPBearer()
 
 def verify_api_key(credentials: HTTPAuthorizationCredentials = Security(security)):
-    """Verify API key from Bearer token"""
-    if credentials.credentials != API_PASSWORD:
-        raise HTTPException(
-            status_code=401,
-            detail="Invalid authentication credentials",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    return credentials.credentials
+    """Verify API key from Bearer token.
+    
+    Supports two authentication methods:
+    1. Static password (API_PASSWORD)
+    2. Firebase ID token (from Firebase Auth)
+    """
+    token = credentials.credentials
+    
+    # Try static password first
+    if token == API_PASSWORD:
+        return {"auth_mode": "static", "token": token}
+    
+    # Try Firebase ID token
+    firebase_claims = verify_firebase_id_token(token)
+    if firebase_claims:
+        return {"auth_mode": "firebase", "claims": firebase_claims}
+    
+    # If neither works, raise error
+    raise HTTPException(
+        status_code=401,
+        detail="Invalid authentication credentials. Use static password or Firebase ID token.",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
 
 # CORS middleware
 app.add_middleware(
@@ -340,7 +355,7 @@ async def process_face_swap(job_id: str, source_image_path: str, target_video_pa
 @app.post("/api/source-image", response_model=SourceImageResponse)
 async def upload_source_image(
     file: UploadFile = File(...),
-    api_key: str = Depends(verify_api_key),
+    api_key: dict = Depends(verify_api_key),
     _app_check_ok: bool = Depends(verify_app_check)
 ):
     """Upload and store source image in MongoDB"""
@@ -377,7 +392,7 @@ async def upload_source_image(
 @app.post("/api/target-video", response_model=TargetVideoResponse)
 async def upload_target_video(
     file: UploadFile = File(...),
-    api_key: str = Depends(verify_api_key),
+    api_key: dict = Depends(verify_api_key),
     _app_check_ok: bool = Depends(verify_app_check)
 ):
     """Upload and store target video in MongoDB"""
@@ -415,7 +430,7 @@ async def upload_target_video(
 async def start_face_swap(
     request: FaceSwapRequest,
     background_tasks: BackgroundTasks,
-    api_key: str = Depends(verify_api_key),
+    api_key: dict = Depends(verify_api_key),
     _app_check_ok: bool = Depends(verify_app_check)
 ):
     """Start face swap processing"""
@@ -460,7 +475,7 @@ async def start_face_swap(
         raise HTTPException(status_code=500, detail=f"Error starting face swap: {str(e)}")
 
 @app.get("/api/job/{job_id}", response_model=JobStatus)
-async def get_job_status(job_id: str, api_key: str = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
+async def get_job_status(job_id: str, api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """Get job status"""
     job = await jobs_collection.find_one({"job_id": job_id})
     if not job:
@@ -480,7 +495,7 @@ async def get_job_status(job_id: str, api_key: str = Depends(verify_api_key), _a
     )
 
 @app.get("/api/result-video/{result_video_id}")
-async def get_result_video(result_video_id: str, api_key: str = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
+async def get_result_video(result_video_id: str, api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """Get result video file"""
     result = await result_videos_collection.find_one({"_id": ObjectId(result_video_id)})
     if not result:
@@ -496,7 +511,7 @@ async def get_result_video(result_video_id: str, api_key: str = Depends(verify_a
     )
 
 @app.get("/api/source-images", response_model=List[SourceImageResponse])
-async def list_source_images(api_key: str = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
+async def list_source_images(api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """List all source images"""
     cursor = source_images_collection.find().sort("uploaded_at", -1)
     images = []
@@ -511,7 +526,7 @@ async def list_source_images(api_key: str = Depends(verify_api_key), _app_check_
     return images
 
 @app.get("/api/target-videos", response_model=List[TargetVideoResponse])
-async def list_target_videos(api_key: str = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
+async def list_target_videos(api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """List all target videos"""
     cursor = target_videos_collection.find().sort("uploaded_at", -1)
     videos = []
@@ -526,7 +541,7 @@ async def list_target_videos(api_key: str = Depends(verify_api_key), _app_check_
     return videos
 
 @app.get("/api/result-videos", response_model=List[ResultVideoResponse])
-async def list_result_videos(api_key: str = Depends(verify_app_check), _app_check_ok: bool = Depends(verify_app_check)):
+async def list_result_videos(api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """List all result videos"""
     cursor = result_videos_collection.find().sort("created_at", -1)
     results = []
@@ -543,7 +558,7 @@ async def list_result_videos(api_key: str = Depends(verify_app_check), _app_chec
     return results
 
 @app.get("/api/health")
-async def api_health(api_key: str = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
+async def api_health(api_key: dict = Depends(verify_api_key), _app_check_ok: bool = Depends(verify_app_check)):
     """Health check endpoint with GPU status (requires authentication)"""
     import onnxruntime
     available_providers = onnxruntime.get_available_providers()
@@ -589,7 +604,7 @@ async def get_api_logs(
     limit: int = 100,
     level: Optional[str] = None,
     endpoint: Optional[str] = None,
-    api_key: str = Depends(verify_api_key)
+    api_key: dict = Depends(verify_api_key)
 ):
     """Get API logs from MongoDB"""
     if api_logs_collection is None:
@@ -623,6 +638,35 @@ async def get_api_logs(
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Error fetching logs: {str(e)}")
 
+# User info endpoint
+@app.get("/api/me")
+async def get_current_user(api_key: dict = Depends(verify_api_key)):
+    """Get current authenticated user info"""
+    auth_mode = api_key.get("auth_mode")
+    
+    if auth_mode == "firebase":
+        claims = api_key.get("claims", {})
+        return {
+            "auth_mode": "firebase",
+            "user_id": claims.get("uid"),
+            "email": claims.get("email"),
+            "email_verified": claims.get("email_verified", False),
+            "name": claims.get("name"),
+            "picture": claims.get("picture"),
+            "firebase": {
+                "sign_in_provider": claims.get("firebase", {}).get("sign_in_provider"),
+                "iss": claims.get("iss"),
+                "aud": claims.get("aud"),
+                "auth_time": claims.get("auth_time"),
+                "exp": claims.get("exp")
+            }
+        }
+    else:
+        return {
+            "auth_mode": "static",
+            "message": "Using static API password authentication"
+        }
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)

firebase_app_check.py

@@ -4,7 +4,7 @@ from typing import Optional
 
 try:
     import firebase_admin
-    from firebase_admin import credentials
+    from firebase_admin import credentials, auth
     # App Check verification is available in newer firebase_admin versions
     try:
         from firebase_admin import app_check
@@ -13,6 +13,7 @@ try:
 except ImportError:
     firebase_admin = None
     credentials = None
+    auth = None
     app_check = None
 
 _initialized = False
@@ -94,4 +95,27 @@ def verify_app_check_token(token: Optional[str]) -> bool:
         print(f"Warning: App Check token verification failed: {e}")
         return False
 
+def verify_firebase_id_token(id_token: Optional[str]) -> Optional[dict]:
+    """Verify Firebase ID token and return decoded claims.
+    
+    Returns:
+        dict: Decoded token claims (user info) if valid, None if invalid/not available
+    """
+    if not id_token:
+        return None
+    
+    if auth is None:
+        return None
+    
+    if not initialize_firebase():
+        return None
+    
+    try:
+        # Verify the ID token
+        decoded_token = auth.verify_id_token(id_token)
+        return decoded_token
+    except Exception as e:
+        print(f"Warning: Firebase ID token verification failed: {e}")
+        return None
+