File size: 1,527 Bytes
6cc9547 | 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 | import os
import jwt
from jwt import PyJWKClient
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
bearer = HTTPBearer()
CLERK_JWKS_URL = os.getenv("CLERK_JWKS_URL")
CLERK_ISSUER_URL = os.getenv("CLERK_ISSUER_URL", "") # e.g. https://pretty-bird-74.clerk.accounts.dev
_jwks_client: PyJWKClient | None = None
def get_jwks_client() -> PyJWKClient:
global _jwks_client
if _jwks_client is None:
_jwks_client = PyJWKClient(CLERK_JWKS_URL, cache_keys=True)
return _jwks_client
async def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer)) -> dict:
token = credentials.credentials
try:
signing_key = get_jwks_client().get_signing_key_from_jwt(token)
decode_kwargs: dict = {
"algorithms": ["RS256"],
"options": {"verify_aud": False},
}
if CLERK_ISSUER_URL:
decode_kwargs["issuer"] = CLERK_ISSUER_URL
payload = jwt.decode(token, signing_key.key, **decode_kwargs)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError as e:
raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
def get_user_id(payload: dict = Depends(verify_token)) -> str:
user_id = payload.get("sub")
if not user_id:
raise HTTPException(status_code=401, detail="No user ID in token")
return user_id
|