implements firebase ID tokens

app/web/security.py

@@ -4,6 +4,9 @@ from fastapi import HTTPException, status, Depends
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from sqlalchemy.orm import Session
 
+import firebase_admin
+from firebase_admin import credentials, auth, exceptions
+
 from app.web.config import ALLOW_ANY_EMAIL
 from app.shared.settings import get_settings
 from app.shared.db.database import get_db_dependency
@@ -12,6 +15,11 @@ from app.web.db.user_state import UserState
 settings = get_settings()
 bearer_security = HTTPBearer()
 
+FIREBASE_OAUTH_ENABLED = settings.FIREBASE_SERVICE_ACCOUNT_JSON != ""
+if FIREBASE_OAUTH_ENABLED:
+    logger.debug("Firebase OAUTH enabled, initializing...")
+    firebase_admin.initialize_app(credentials.Certificate(settings.FIREBASE_SERVICE_ACCOUNT_JSON))
+
 
 def secure_compare(token, api_key):
     return secrets.compare_digest(token.encode("utf8"), api_key.encode("utf8"))
@@ -59,6 +67,19 @@ async def get_user_auth(credentials: HTTPAuthorizationCredentials = Depends(bear
 
 
 def authenticate_user(access_token):
+    if FIREBASE_OAUTH_ENABLED:
+        try:
+            j = auth.verify_id_token(access_token)
+            email = j.get('email', None)
+            logger.debug(f"Successfully verified the ID token for {email}")
+            if email is None:
+                return False, "email not found in token"
+            if email in settings.BLOCKED_EMAILS:
+                return False, f"email '{email}' not allowed"
+            return True, email
+        except exceptions.FirebaseError as e:
+            logger.warning(f"Error verifying ID token: {str(e)}")
+
     # https://cloud.google.com/docs/authentication/token-types#access
     if type(access_token) != str or len(access_token) < 10: return False, "invalid access_token"
     r = requests.get("https://oauth2.googleapis.com/tokeninfo", {"access_token": access_token})
@@ -79,5 +100,5 @@ def authenticate_user(access_token):
         return False, "exception occurred"
 
 
-def get_user_state(email:str=Depends(get_user_auth), db:Session=Depends(get_db_dependency)):
-    return UserState(db, email)
+def get_user_state(email: str = Depends(get_user_auth), db: Session = Depends(get_db_dependency)):
+    return UserState(db, email)