Spaces:
Sleeping
Sleeping
File size: 2,584 Bytes
923cd30 |
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 |
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")
@app.get("/")
async def root():
return {"message": "API is running"} |