Spaces:
Sleeping
Sleeping
File size: 2,311 Bytes
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 |
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",
"https://junsen-ai.vercel.app",
# "https://huggingface.co"
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 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)
# Middleware pour vérifier le token Firebase
async def verify_token(authorization: str = Header(...)):
try:
token = authorization.split("Bearer ")[1]
decoded_token = auth.verify_id_token(token)
return decoded_token
except Exception:
raise HTTPException(status_code=401, detail="Token invalide")
# 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):
if token.get("role") not in roles:
raise HTTPException(status_code=403, detail="Accès non autorisé")
return await func(*args, **kwargs)
return wrapper
return decorator
# Routes sécurisées
@app.get("/api/protected/admin")
@require_role(["admin"])
async def admin_only():
return {"message": "Route admin"}
@app.get("/api/protected/user")
@require_role(["user", "admin"])
async def user_route():
return {"message": "Route utilisateur"}
# Route de test pour vérifier que le serveur fonctionne
@app.get("/")
async def root():
return {"message": "Backend Junsen AI opérationnel"} |