Spaces:
Sleeping
Sleeping
File size: 3,228 Bytes
432e043 | 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 87 88 89 90 | """
ChillTrack API — Firebase auth helpers.
Auth pattern follows RentNow-Admin-API/app.py and NeoFix-API/main.py:
- Firebase service account JSON lives in the `FIREBASE` env var (json.loads it).
- Firestore is the datastore (unlike the Realtime-DB repos), so init mirrors
RentNow's firestore.client(database_id=os.environ.get("FIRESTORE_DB_NAME", "(default)")).
- Firebase ID token in `Authorization: Bearer <token>`.
- 401 = missing/invalid token, 403 = signed in but wrong role.
"""
import os
import json
from functools import wraps
from flask import request, jsonify
import firebase_admin
from firebase_admin import credentials, auth, firestore
AUDIT_COLLECTION = "ctk_audit"
def init_firebase():
"""Initializes the Firebase Admin app exactly once and returns the Firestore client."""
if firebase_admin._apps:
return firestore.client(database_id=os.environ.get("FIRESTORE_DB_NAME", "(default)"))
raw = os.environ.get("FIREBASE")
if not raw:
print("FATAL: The 'FIREBASE' environment variable is not set.")
exit(1)
try:
cred_json = json.loads(raw)
cred = credentials.Certificate(cred_json)
storage_bucket = os.environ.get("FIREBASE_STORAGE_BUCKET", "")
firebase_admin.initialize_app(cred, {"storageBucket": storage_bucket} if storage_bucket else {})
print("Firebase Admin SDK initialized successfully.")
except Exception as e:
print(f"FATAL: Error initializing Firebase: {e}")
exit(1)
return firestore.client(database_id=os.environ.get("FIRESTORE_DB_NAME", "(default)"))
def require_auth(fn):
"""Verifies the Firebase ID token and attaches the decoded token to request.user."""
@wraps(fn)
def wrapper(*args, **kwargs):
header = request.headers.get("Authorization", "")
token = header[7:].strip() if header.startswith("Bearer ") else ""
if not token:
return jsonify({"error": "Missing bearer token."}), 401
try:
decoded_token = auth.verify_id_token(token)
except Exception as e:
print(f"Token verification failed: {e}")
return jsonify({"error": "Invalid or expired token."}), 401
request.user = decoded_token
return fn(*args, **kwargs)
return wrapper
def require_role(role):
"""Decorator factory: require_auth, plus request.user['role'] must equal `role`."""
def decorator(fn):
@wraps(fn)
@require_auth
def wrapper(*args, **kwargs):
if request.user.get("role") != role:
return jsonify({"error": f"'{role}' access required."}), 403
return fn(*args, **kwargs)
return wrapper
return decorator
def audit_log(db, actor, action, entity, entity_id, metadata=None):
"""Writes an audit trail doc. actor is {uid, email}. Never raises — logs and swallows."""
try:
db.collection(AUDIT_COLLECTION).add({
"actor": actor,
"action": action,
"entity": entity,
"entity_id": entity_id,
"metadata": metadata or {},
"ts": firestore.SERVER_TIMESTAMP,
})
except Exception as e:
print(f"audit_log failed: {e}")
|