Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, Depends, HTTPException, Header | |
from fastapi.middleware.cors import CORSMiddleware | |
from firebase_admin import auth, credentials, initialize_app | |
from functools import wraps | |
import firebase_admin | |
import os | |
import json | |
from dotenv import load_dotenv | |
# Charger les variables d'environnement | |
load_dotenv() | |
app = FastAPI() | |
# Configuration CORS pour autoriser les requêtes depuis le frontend | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=[ | |
"http://localhost:5173", | |
"http://localhost:8000", | |
"https://junsen-ai.vercel.app", | |
"https://2nzi-backendjunsen.hf.space" | |
], | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
# Décorateur pour restreindre l'accès en fonction des rôles | |
def require_role(roles): | |
def decorator(func): | |
async def wrapper(*args, token=Depends(verify_token), **kwargs): | |
print("Token dans require_role:", token) | |
print("Rôle requis:", roles) | |
print("Rôle trouvé:", token.get("role")) | |
if token.get("role") not in roles: | |
raise HTTPException( | |
status_code=403, | |
detail={"message": "Accès non autorisé", "role": token.get("role")} | |
) | |
return await func(*args, **kwargs) | |
return wrapper | |
return decorator | |
# Initialiser Firebase avec les credentials depuis les variables d'environnement | |
if os.getenv('FIREBASE_CREDENTIALS'): | |
cred_dict = json.loads(os.getenv('FIREBASE_CREDENTIALS')) | |
cred = credentials.Certificate(cred_dict) | |
else: | |
# Fallback pour le développement local | |
cred = credentials.Certificate("serviceAccountKey.json") | |
firebase_admin.initialize_app(cred) | |
async def verify_token(authorization: str = Header(...)): | |
try: | |
print("Authorization reçu:", authorization[:50] + "...") | |
token = authorization.split("Bearer ")[1] | |
print("Token extrait:", token[:50] + "...") | |
decoded_token = auth.verify_id_token(token) | |
# Récupérer les custom claims ou le rôle depuis Firestore | |
try: | |
user = auth.get_user(decoded_token['uid']) | |
print("User data:", user._data) | |
except Exception as e: | |
print("Erreur lors de la récupération des données utilisateur:", str(e)) | |
return decoded_token | |
except Exception as e: | |
print("Erreur de vérification:", str(e)) | |
raise HTTPException( | |
status_code=401, | |
detail={ | |
"message": "Token invalide", | |
"error": str(e) | |
} | |
) | |
async def user_route(): | |
try: | |
return {"message": "Route utilisateur"} | |
except Exception as e: | |
print("Erreur dans user_route:", str(e)) | |
raise HTTPException( | |
status_code=500, | |
detail={ | |
"message": "Erreur serveur", | |
"error": str(e) | |
} | |
) | |
# Routes sécurisées | |
async def admin_only(): | |
return {"message": "Route admin"} | |
# Route de test pour vérifier que le serveur fonctionne | |
async def root(): | |
return {"message": "Backend Junsen AI opérationnel"} |