BackendJunsen / main.py
2nzi's picture
correction main
6d8700d verified
raw
history blame
3.36 kB
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"}