Spaces:
Running
Running
| from typing import Optional | |
| from jose import jwt, JWTError | |
| from fastapi import HTTPException, Depends, status | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from app.core.config import settings | |
| # Utilitário para o Swagger/Bearer Token | |
| security = HTTPBearer() | |
| def decode_access_token(token: str) -> dict: | |
| """ | |
| Decodifica o token JWT do Supabase usando a Secret Key do projeto. | |
| """ | |
| # Remove aspas acidentais caso o token seja enviado como: "eyJ..." | |
| token = token.strip('"').strip("'") | |
| try: | |
| payload = jwt.decode( | |
| token, | |
| settings.SUPABASE_JWT_SECRET.get_secret_value(), | |
| algorithms=["HS256", "ES256"], | |
| audience="authenticated", | |
| options={"verify_signature": False}, | |
| ) | |
| return payload | |
| except JWTError as e: | |
| print(f"JWT Decode Error: {e}") | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Token inválido ou expirado", | |
| ) | |
| def get_current_user_id(res: HTTPAuthorizationCredentials = Depends(security)) -> str: | |
| """ | |
| Dependência FastAPI que extrai o ID do usuário (sub) do token. | |
| """ | |
| token = res.credentials | |
| payload = decode_access_token(token) | |
| user_id = payload.get("sub") | |
| if not user_id: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Token não contém identificador do usuário (sub)", | |
| ) | |
| return user_id |