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): @wraps(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) } ) @app.get("/api/protected/user") @require_role(["user_extern", "admin"]) 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 @app.get("/api/protected/admin") @require_role(["admin"]) async def admin_only(): return {"message": "Route admin"} # Route de test pour vérifier que le serveur fonctionne @app.get("/") async def root(): return {"message": "Backend Junsen AI opérationnel"}