47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from app.core.supabase_client import supabase, supabase_admin
|
|
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
) -> dict:
|
|
"""Valida el JWT de Supabase y devuelve {user_id, email, role}."""
|
|
token = credentials.credentials
|
|
try:
|
|
resp = supabase.auth.get_user(token)
|
|
auth_user = resp.user
|
|
if auth_user is None:
|
|
raise ValueError("usuario nulo")
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token inválido o expirado",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
result = (
|
|
supabase_admin.table("users")
|
|
.select("role")
|
|
.eq("id", str(auth_user.id))
|
|
.maybe_single()
|
|
.execute()
|
|
)
|
|
role = result.data["role"] if result.data else "citizen"
|
|
|
|
return {"user_id": str(auth_user.id), "email": auth_user.email, "role": role}
|
|
|
|
|
|
def require_role(*roles: str):
|
|
"""Factory: devuelve una dependencia que exige que el usuario tenga uno de los roles dados."""
|
|
async def checker(current_user: dict = Depends(get_current_user)) -> dict:
|
|
if current_user["role"] not in roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Rol requerido: {' o '.join(roles)}",
|
|
)
|
|
return current_user
|
|
return checker
|