File size: 3,278 Bytes
6405808
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
faeab5c
 
 
 
 
c29587d
 
 
 
faeab5c
c29587d
 
faeab5c
 
c29587d
faeab5c
 
 
6405808
 
 
 
 
 
 
 
c29587d
 
6405808
 
 
c29587d
6405808
 
 
 
 
 
 
b17b103
c29587d
6405808
 
fbbd126
 
b17b103
fbbd126
 
 
b17b103
fbbd126
6405808
c29587d
 
b17b103
c29587d
b17b103
c29587d
6405808
b17b103
 
c29587d
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""
Firebase initialization and helper functions.
"""

import firebase_admin
from firebase_admin import credentials, firestore, auth
from pathlib import Path
from src.utils.config import settings
from src.utils.logger import setup_logger

logger = setup_logger(__name__)

_db = None

def get_firebase_db():
    """
    Initialize Firebase Admin SDK and return Firestore client.
    """
    global _db
    if _db is not None:
        return _db

    try:
        # Priority 1: Check if credentials string is provided (e.g., from HF Secrets)
        if settings.firebase_service_account_json:
            import json
            service_account_info = json.loads(settings.firebase_service_account_json)
            cred = credentials.Certificate(service_account_info)
            
            # Explicitly extract project_id from JSON if possible
            project_id = service_account_info.get('project_id') or settings.firebase_project_id
            
            firebase_admin.initialize_app(cred, {
                'storageBucket': settings.firebase_storage_bucket,
                'projectId': project_id
            })
            _db = firestore.client()
            logger.info(f"Firebase initialized successfully for project: {project_id} (using JSON)")
            return _db

        # Priority 2: Check for service account file
        service_account_path = Path(settings.firebase_service_account_path)
        
        if not service_account_path.exists():
            logger.warning(f"Firebase service account file not found at {service_account_path}. Using fallback/mock behavior if applicable.")
            return None

        cred = credentials.Certificate(str(service_account_path))
        firebase_admin.initialize_app(cred, {
            'storageBucket': settings.firebase_storage_bucket,
            'projectId': settings.firebase_project_id
        })
        
        _db = firestore.client()
        logger.info(f"Firebase initialized successfully for project: {settings.firebase_project_id} (using file)")
        return _db
    except Exception as e:
        logger.error(f"Failed to initialize Firebase: {e}")
        return None

def verify_token(id_token: str):
    """
    Verify a Firebase ID token and return the payload or error.
    Returns: {"payload": decoded_token, "error": error_message}
    """
    try:
        # Check if any app is initialized, if not try to initialize
        if not firebase_admin._apps:
            logger.warning("Firebase app not initialized. Attempting default initialization...")
            get_firebase_db()
            
        if not firebase_admin._apps:
            return {"payload": None, "error": "Firebase app could not be initialized."}

        decoded_token = auth.verify_id_token(id_token)
        
        if decoded_token is None:
             return {"payload": None, "error": "Firebase SDK returned None for token verification."}
             
        logger.info(f"Firebase token successfully verified for UID: {decoded_token.get('uid')}")
        return {"payload": decoded_token, "error": None}
    except Exception as e:
        error_msg = str(e) or repr(e)
        logger.error(f"Firebase token verification failed: {error_msg}")
        return {"payload": None, "error": error_msg}