ats-system / backend /api /auth.py
Aju360's picture
feat: Add ATS Resume Scorer components and views
10288a2
Raw
History Blame Contribute Delete
3.65 kB
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