src/monitor_app/auth0.py

0001 """Auth0 JWT validation utilities for MCP OAuth 2.1 authentication."""
0002
0003 import logging
0004 from functools import lru_cache
0005
0006 import requests
0007 from django.conf import settings
0008 from jose import jwt, JWTError
0009
0010 logger = logging.getLogger(__name__)
0011
0012 # Cache JWKS for 1 hour (3600 seconds)
0013 _jwks_cache = {"keys": None, "expires": 0}
0014
0015
0016 def get_jwks():
0017     """Fetch and cache JWKS from Auth0."""
0018     import time
0019
0020     now = time.time()
0021     if _jwks_cache["keys"] and now < _jwks_cache["expires"]:
0022         return _jwks_cache["keys"]
0023
0024     if not settings.AUTH0_DOMAIN:
0025         logger.warning("AUTH0_DOMAIN not configured")
0026         return None
0027
0028     jwks_url = f"https://{settings.AUTH0_DOMAIN}/.well-known/jwks.json"
0029     try:
0030         response = requests.get(jwks_url, timeout=10)
0031         response.raise_for_status()
0032         jwks = response.json()
0033         _jwks_cache["keys"] = jwks
0034         _jwks_cache["expires"] = now + 3600  # Cache for 1 hour
0035         return jwks
0036     except Exception as e:
0037         logger.error(f"Failed to fetch JWKS from Auth0: {e}")
0038         return None
0039
0040
0041 def validate_token(token: str) -> dict | None:
0042     """
0043     Validate a JWT token from Auth0.
0044
0045     Returns the decoded token payload if valid, None otherwise.
0046     """
0047     if not settings.AUTH0_DOMAIN or not settings.AUTH0_API_IDENTIFIER:
0048         logger.warning("Auth0 not configured, skipping token validation")
0049         return None
0050
0051     jwks = get_jwks()
0052     if not jwks:
0053         logger.warning("Failed to get JWKS")
0054         return None
0055
0056     try:
0057         # Get the key ID from the token header
0058         unverified_header = jwt.get_unverified_header(token)
0059         kid = unverified_header.get("kid")
0060
0061         # Find the matching key
0062         rsa_key = None
0063         for key in jwks.get("keys", []):
0064             if key.get("kid") == kid:
0065                 rsa_key = {
0066                     "kty": key["kty"],
0067                     "kid": key["kid"],
0068                     "use": key["use"],
0069                     "n": key["n"],
0070                     "e": key["e"],
0071                 }
0072                 break
0073
0074         if not rsa_key:
0075             logger.warning(f"No matching key found for kid: {kid}")
0076             return None
0077
0078         # Validate the token
0079         payload = jwt.decode(
0080             token,
0081             rsa_key,
0082             algorithms=settings.AUTH0_ALGORITHMS,
0083             audience=settings.AUTH0_API_IDENTIFIER,
0084             issuer=f"https://{settings.AUTH0_DOMAIN}/",
0085         )
0086         return payload
0087
0088     except JWTError as e:
0089         logger.warning(f"JWT validation failed: {e}")
0090         return None
0091     except Exception as e:
0092         logger.error(f"Unexpected error validating token: {e}")
0093         return None
0094
0095
0096 def get_bearer_token(request) -> str | None:
0097     """Extract Bearer token from Authorization header."""
0098     auth_header = request.META.get("HTTP_AUTHORIZATION", "")
0099     if auth_header.startswith("Bearer "):
0100         return auth_header[7:]
0101     return None