Spaces:
Running
Running
| import logging | |
| import jwt | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer | |
| from backend.core.config import SUPABASE_JWT_SECRET, SUPABASE_URL | |
| logger = logging.getLogger('ats_resume_scorer') | |
| _bearer_scheme = HTTPBearer(auto_error=False) | |
| _ASYMMETRIC_ALGS = ['ES256', 'RS256'] | |
| _jwks_client: jwt.PyJWKClient | None = None | |
| def _get_jwks_client() -> jwt.PyJWKClient | None: | |
| global _jwks_client | |
| if _jwks_client is not None: | |
| return _jwks_client | |
| if not SUPABASE_URL: | |
| return None | |
| jwks_url = f"{SUPABASE_URL.rstrip('/')}/auth/v1/.well-known/jwks.json" | |
| _jwks_client = jwt.PyJWKClient(jwks_url, cache_keys=True, lifespan=3600) | |
| return _jwks_client | |
| def _verify_token(token: str) -> dict: | |
| header = jwt.get_unverified_header(token) | |
| alg = header.get('alg') | |
| if alg in _ASYMMETRIC_ALGS: | |
| jwks_client = _get_jwks_client() | |
| if jwks_client is None: | |
| raise jwt.InvalidTokenError( | |
| 'SUPABASE_URL not configured β cannot fetch JWKS to verify token' | |
| ) | |
| signing_key = jwks_client.get_signing_key_from_jwt(token).key | |
| return jwt.decode( | |
| token, | |
| signing_key, | |
| algorithms=_ASYMMETRIC_ALGS, | |
| audience='authenticated', | |
| ) | |
| if alg == 'HS256': | |
| if not SUPABASE_JWT_SECRET: | |
| raise jwt.InvalidTokenError( | |
| 'HS256 token received but SUPABASE_JWT_SECRET is not configured' | |
| ) | |
| return jwt.decode( | |
| token, | |
| SUPABASE_JWT_SECRET, | |
| algorithms=['HS256'], | |
| audience='authenticated', | |
| ) | |
| raise jwt.InvalidTokenError(f'Unsupported JWT algorithm: {alg}') | |
| def get_current_user( | |
| creds: HTTPAuthorizationCredentials | None = Depends(_bearer_scheme), | |
| ) -> str: | |
| if creds is None or not creds.credentials: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail='Missing Authorization: Bearer <token> header', | |
| headers={'WWW-Authenticate': 'Bearer'}, | |
| ) | |
| if not SUPABASE_URL and not SUPABASE_JWT_SECRET: | |
| logger.error('Neither SUPABASE_URL (for JWKS) nor SUPABASE_JWT_SECRET configured β cannot verify tokens') | |
| raise HTTPException( | |
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, | |
| detail='Auth not configured on the server', | |
| ) | |
| try: | |
| payload = _verify_token(creds.credentials) | |
| except jwt.ExpiredSignatureError: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail='Token expired β sign in again', | |
| headers={'WWW-Authenticate': 'Bearer'}, | |
| ) | |
| except jwt.InvalidTokenError as exc: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail=f'Invalid token: {exc}', | |
| headers={'WWW-Authenticate': 'Bearer'}, | |
| ) | |
| except Exception as exc: | |
| # PyJWKClient can raise network errors fetching JWKS; surface them as 401 | |
| # so a misconfigured backend doesn't look like a 500 to the user. | |
| logger.warning(f'JWT verification failed: {exc}') | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail=f'Token verification failed: {exc}', | |
| headers={'WWW-Authenticate': 'Bearer'}, | |
| ) | |
| user_id = payload.get('sub') | |
| if not user_id: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail='Token missing subject claim', | |
| ) | |
| return user_id | |