Coverage for src/lumen/auth/token.py: 100%

1""" 

2This file validates OIDC/SMART bearer tokens and resolves the associated tenant. 

3It decodes JWTs, verifies signatures via JWKS, and extracts tenant identifiers. 

4""" 

5 

6from __future__ import annotations 

7 

8from uuid import UUID 

9 

10import jwt 

11import structlog 

12from fastapi import HTTPException, Request, status 

13 

14from ..config import SmartSettings 

15from ..core.context import current_claims 

16from .jwks import discover_jwks_uri, get_jwks_cache 

17 

18log = structlog.get_logger() 

19 

20_SUPPORTED_ALGORITHMS = ["RS256", "RS384", "RS512", "ES256", "ES384", "ES512"] 

21 

22# WWW-Authenticate header returned on 401 (RFC 6750 §3). 

23_WWW_AUTH = 'Bearer realm="lumen"' 

24 

25 

26# TODO: Move errors classes to a separate file, e.g., lumen/core/errors.py 

27class AuthError(HTTPException): 

28 """401 or 403 with a FHIR OperationOutcome body (built by the exception handler).""" 

29 

30 def __init__(self, status_code: int, fhir_code: str, diagnostics: str) -> None: 

31 headers = ( 

32 {"WWW-Authenticate": _WWW_AUTH} if status_code == status.HTTP_401_UNAUTHORIZED else {} 

33 ) 

34 super().__init__( 

35 status_code=status_code, 

36 detail={"fhir_code": fhir_code, "diagnostics": diagnostics}, 

37 headers=headers, 

38 ) 

39 

40 

41def _extract_bearer(request: Request) -> str: 

42 auth = request.headers.get("Authorization", "") 

43 if not auth.startswith("Bearer "): 

44 raise AuthError( 

45 status.HTTP_401_UNAUTHORIZED, 

46 "login", 

47 "Missing or malformed Authorization header — expected 'Bearer <token>'.", 

48 ) 

49 return auth[len("Bearer ") :] 

50 

51 

52async def validate_token(request: Request, settings: SmartSettings) -> UUID: 

53 """Validate the Bearer token and return the resolved tenant_id.""" 

54 raw = _extract_bearer(request) 

55 

56 # Peek at the header to obtain `kid` and `iss` without full verification. 

57 try: 

58 unverified_header = jwt.get_unverified_header(raw) 

59 unverified_claims = jwt.decode(raw, options={"verify_signature": False}) 

60 except jwt.DecodeError as exc: 

61 log.warning("auth.token_invalid", error=f"Token decode error: {exc}") 

62 raise AuthError( 

63 status.HTTP_401_UNAUTHORIZED, "security", f"Token decode error: {exc}" 

64 ) from exc 

65 

66 alg = unverified_header.get("alg", "") 

67 kid = unverified_header.get("kid", "") 

68 iss: str = unverified_claims.get("iss", "") 

69 

70 if alg not in _SUPPORTED_ALGORITHMS: 

71 log.warning("auth.token_invalid", error=f"Unsupported algorithm: {alg!r}") 

72 raise AuthError( 

73 status.HTTP_401_UNAUTHORIZED, "security", f"Unsupported algorithm: {alg!r}." 

74 ) 

75 

76 if not iss: 

77 log.warning("auth.token_invalid", error="Token missing 'iss' claim.") 

78 raise AuthError(status.HTTP_401_UNAUTHORIZED, "security", "Token missing 'iss' claim.") 

79 

80 # Resolve JWKS URI from the issuer's OIDC discovery document. 

81 # Use internal_issuer_map to translate external → internal base URL when 

82 # Lumen runs in Docker and cannot reach the external issuer URL directly. 

83 fetch_base = settings.internal_issuer_map.get(iss) 

84 cache = get_jwks_cache(ttl=settings.jwks_cache_ttl) 

85 jwks_uri: str = "" 

86 try: 

87 jwks_uri = await discover_jwks_uri(iss, fetch_base=fetch_base) 

88 public_key = await cache.get_public_key(jwks_uri, kid) 

89 except KeyError as exc: 

90 # `kid` not in JWKS — try once after invalidating cache (key rotation). 

91 cache.invalidate(jwks_uri) 

92 try: 

93 public_key = await cache.get_public_key(jwks_uri, kid) 

94 except KeyError: 

95 log.warning("auth.token_invalid", error=str(exc)) 

96 raise AuthError(status.HTTP_401_UNAUTHORIZED, "security", str(exc)) from exc 

97 except Exception as exc: 

98 log.error("auth.jwks_fetch_failed", error=str(exc), iss=iss) 

99 raise AuthError( 

100 status.HTTP_401_UNAUTHORIZED, "security", f"Failed to fetch JWKS: {exc}" 

101 ) from exc 

102 

103 # Full verification: signature + iss + aud + exp. 

104 try: 

105 claims: dict[str, object] = jwt.decode( 

106 raw, 

107 public_key, 

108 algorithms=[alg], 

109 audience=settings.audience, 

110 issuer=iss, 

111 ) 

112 except jwt.ExpiredSignatureError as exc: 

113 log.warning("auth.token_invalid", error="Token has expired") 

114 raise AuthError(status.HTTP_401_UNAUTHORIZED, "security", "Token has expired.") from exc 

115 except jwt.InvalidAudienceError as exc: 

116 log.warning("auth.token_invalid", error=f"Invalid audience: {exc}") 

117 raise AuthError( 

118 status.HTTP_401_UNAUTHORIZED, "security", f"Invalid audience: {exc}" 

119 ) from exc 

120 except jwt.InvalidIssuerError as exc: 

121 log.warning("auth.token_invalid", error=f"Invalid issuer: {exc}") 

122 raise AuthError(status.HTTP_401_UNAUTHORIZED, "security", f"Invalid issuer: {exc}") from exc 

123 except jwt.PyJWTError as exc: 

124 log.warning("auth.token_invalid", error=f"Token invalid: {exc}") 

125 raise AuthError(status.HTTP_401_UNAUTHORIZED, "security", f"Token invalid: {exc}") from exc 

126 

127 # Map issuer -> tenant_id. 

128 tenant_id = settings.tenant_map.get(iss) 

129 if tenant_id is None: 

130 log.warning("auth.unknown_issuer", iss=iss) 

131 raise AuthError( 

132 status.HTTP_403_FORBIDDEN, 

133 "forbidden", 

134 f"Issuer '{iss}' is not registered with this server.", 

135 ) 

136 

137 log.debug("auth.token_valid", iss=iss, tenant_id=str(tenant_id)) 

138 

139 # Store full claims so the scope enforcer can read them without re-validating. 

140 request.state.smart_claims = claims 

141 current_claims.set(claims) 

142 return tenant_id