Spaces:
Sleeping
Sleeping
File size: 3,364 Bytes
6d8700d 53e151f |
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 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
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"} |