Spaces:
Sleeping
Sleeping
File size: 3,748 Bytes
dc95236 | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | 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, SUPABASE_ANON_KEY
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,
headers={"apikey": SUPABASE_ANON_KEY},
)
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
|