Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, Depends, HTTPException, status, Response | |
from fastapi.middleware.cors import CORSMiddleware | |
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
from firebase_admin import auth, credentials, firestore | |
from firebase_admin import auth | |
from app.core.firebase import db, get_firebase_app # Modifier cette ligne | |
from app.api.endpoints.videos import router as videos_router | |
get_firebase_app() | |
app = FastAPI() | |
# Configuration CORS | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["POST", "GET"], | |
allow_headers=["*"] | |
) | |
# Configuration CORS | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["POST", "GET"], | |
allow_headers=["*"] | |
) | |
def get_user(res: Response, | |
cred: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False))): | |
if cred is None: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="Bearer authentication required", | |
headers={'WWW-Authenticate': 'Bearer realm="auth_required"'}, | |
) | |
try: | |
decoded_token = auth.verify_id_token( | |
cred.credentials, | |
check_revoked=True, | |
clock_skew_seconds=1800 | |
) | |
user_id = decoded_token['uid'] | |
user_doc = db.collection('users').document(user_id).get() | |
if not user_doc.exists: | |
raise HTTPException(status_code=401, detail="Utilisateur non trouvé dans Firestore") | |
user_data = user_doc.to_dict() | |
user_role = user_data.get('role', 'user_extern') | |
decoded_token['role'] = user_role | |
res.headers['WWW-Authenticate'] = 'Bearer realm="auth_required"' | |
return decoded_token | |
except Exception as err: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail=f"Invalid authentication credentials. {err}", | |
headers={'WWW-Authenticate': 'Bearer error="invalid_token"'}, | |
) | |
def require_role(allowed_roles): | |
def role_checker(user_info=Depends(get_user)): | |
if user_info['role'] not in allowed_roles: | |
raise HTTPException(status_code=403, detail="Accès non autorisé") | |
return user_info | |
return role_checker | |
# Inclure le router videos | |
app.include_router(videos_router, prefix="/api") | |
async def root(): | |
return {"message": "API is running"} |