"""OIDC bearer-token validation for native, customer, and admin routes.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import json from typing import Mapping, Optional, Protocol from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen import jwt from jwt import PyJWKClient from licensing_shared.canonical_json import parse_bounded_json from licensing_shared.errors import LicensingError from .config import ServerSettings from .constants import ( MAX_BEARER_TOKEN_CHARACTERS, MAX_EXTERNAL_SUBJECT_CHARACTERS, MAX_OIDC_JWKS_KEYS, MAX_OIDC_JWKS_RESPONSE_BYTES, MAX_OIDC_SCOPE_CHARACTERS, MAX_OIDC_SCOPE_COUNT, MAX_VERIFIED_EMAIL_CHARACTERS, OIDC_JWKS_CACHE_SECONDS, OIDC_JWKS_HTTP_TIMEOUT_SECONDS, ) from .errors import ServerErrorCode, ServerLicensingError ALLOWED_OIDC_ALGORITHMS = ("RS256", "ES256") class _BoundedPyJWKClient(PyJWKClient): """Fetch JWKS documents through the project's bounded JSON boundary.""" def fetch_data(self) -> object: jwk_set: object try: request = Request(url=self.uri, headers=self.headers) with urlopen( request, timeout=self.timeout, context=self.ssl_context, ) as response: document = response.read(MAX_OIDC_JWKS_RESPONSE_BYTES + 1) except (OSError, TimeoutError, URLError) as exc: if isinstance(exc, HTTPError): exc.close() raise jwt.PyJWKClientConnectionError( "failed to fetch the OIDC signing-key set" ) from exc try: jwk_set = parse_bounded_json( document, maximum_bytes=MAX_OIDC_JWKS_RESPONSE_BYTES, ) if not isinstance(jwk_set, dict): raise ValueError("the OIDC signing-key set must be an object") keys = jwk_set.get("keys") if ( not isinstance(keys, list) or not keys or len(keys) > MAX_OIDC_JWKS_KEYS or any(not isinstance(key, Mapping) for key in keys) ): raise ValueError("the OIDC signing-key set has invalid keys") except (LicensingError, TypeError, ValueError) as exc: raise jwt.PyJWKClientError( "the OIDC provider returned an invalid bounded signing-key set" ) from exc if self.jwk_set_cache is not None: self.jwk_set_cache.put(jwk_set) return jwk_set @dataclass(frozen=True) class AuthenticatedPrincipal: issuer: str subject: str scopes: frozenset[str] verified_email: Optional[str] = None authentication_time: Optional[datetime] = None def __post_init__(self) -> None: for field_name in ("issuer", "subject"): value = getattr(self, field_name) if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") if len(value.strip()) > MAX_EXTERNAL_SUBJECT_CHARACTERS: raise ValueError(f"{field_name} is too long") object.__setattr__(self, field_name, value.strip()) if not isinstance(self.scopes, frozenset): raise TypeError("scopes must be a frozenset of non-empty strings") if len(self.scopes) > MAX_OIDC_SCOPE_COUNT or any( not isinstance(value, str) or not value.strip() or len(value.strip()) > MAX_OIDC_SCOPE_CHARACTERS for value in self.scopes ): raise ValueError("scopes contain an invalid or oversized value") object.__setattr__( self, "scopes", frozenset(value.strip() for value in self.scopes), ) if self.verified_email is not None: if not isinstance(self.verified_email, str) or not self.verified_email.strip(): raise ValueError("verified_email must be a non-empty string or None") normalized_email = self.verified_email.strip() if ( len(normalized_email) > MAX_VERIFIED_EMAIL_CHARACTERS or "\r" in normalized_email or "\n" in normalized_email ): raise ValueError("verified_email is invalid or too long") object.__setattr__(self, "verified_email", normalized_email) if self.authentication_time is not None: if ( not isinstance(self.authentication_time, datetime) or self.authentication_time.tzinfo is None or self.authentication_time.utcoffset() is None ): raise ValueError("authentication_time must be timezone-aware or None") object.__setattr__( self, "authentication_time", self.authentication_time.astimezone(timezone.utc), ) def require_scope(self, scope: str) -> None: if not isinstance(scope, str) or not scope.strip(): raise ValueError("scope must be a non-empty string") if scope.strip() not in self.scopes: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "the authenticated account lacks the required authorization", status_code=403, ) class TokenValidator(Protocol): def validate(self, token: str) -> AuthenticatedPrincipal: ... class OIDCTokenValidator: def __init__(self, settings: ServerSettings) -> None: if not isinstance(settings, ServerSettings): raise TypeError("settings must be ServerSettings") self._settings = settings self._jwk_client = _BoundedPyJWKClient( settings.oidc_jwks_url, cache_keys=False, lifespan=OIDC_JWKS_CACHE_SECONDS, timeout=OIDC_JWKS_HTTP_TIMEOUT_SECONDS, ) def validate(self, token: str) -> AuthenticatedPrincipal: if not isinstance(token, str) or not token.strip(): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "a bearer token is required", status_code=401, ) normalized_token = token.strip() if len(normalized_token) > MAX_BEARER_TOKEN_CHARACTERS: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token is invalid or expired", status_code=401, ) try: signing_key = self._signing_key(normalized_token) claims = jwt.decode( normalized_token, signing_key.key, algorithms=list(ALLOWED_OIDC_ALGORITHMS), audience=self._settings.oidc_audience, issuer=self._settings.oidc_issuer, options={"require": ["exp", "iat", "iss", "sub", "aud"]}, ) except jwt.PyJWTError as exc: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token is invalid or expired", status_code=401, ) from exc return self._principal_from_claims(claims) def _signing_key(self, token: str) -> jwt.PyJWK: if not isinstance(token, str) or not token.strip(): raise ValueError("token must be a non-empty string") if len(token.strip()) > MAX_BEARER_TOKEN_CHARACTERS: raise ValueError("token is too long") header = jwt.get_unverified_header(token.strip()) key_id = header.get("kid") if not isinstance(key_id, str) or not key_id.strip(): raise jwt.InvalidTokenError("bearer token has no signing key identifier") try: signing_keys = self._jwk_client.get_signing_keys() signing_key = self._jwk_client.match_kid(signing_keys, key_id.strip()) if signing_key is None: signing_keys = self._jwk_client.get_signing_keys(refresh=True) signing_key = self._jwk_client.match_kid( signing_keys, key_id.strip(), ) except ( json.JSONDecodeError, UnicodeError, jwt.PyJWKClientConnectionError, jwt.PyJWKClientError, jwt.PyJWKSetError, ) as exc: raise ServerLicensingError( ServerErrorCode.IDENTITY_UNAVAILABLE, "account identity verification is temporarily unavailable", status_code=503, retryable=True, ) from exc if signing_key is None: raise jwt.InvalidTokenError("bearer token signing key is not trusted") return signing_key def _principal_from_claims(self, claims: Mapping[str, object]) -> AuthenticatedPrincipal: if not isinstance(claims, Mapping): raise TypeError("claims must be a mapping") raw_scope = claims.get("scope", "") if isinstance(raw_scope, str): raw_scopes = tuple(value for value in raw_scope.split() if value) elif isinstance(raw_scope, list) and all(isinstance(value, str) for value in raw_scope): raw_scopes = tuple(raw_scope) else: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token scope claim is invalid", status_code=401, ) if len(raw_scopes) > MAX_OIDC_SCOPE_COUNT or any( not value.strip() or len(value.strip()) > MAX_OIDC_SCOPE_CHARACTERS for value in raw_scopes ): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token scope claim is invalid", status_code=401, ) scopes = frozenset(value.strip() for value in raw_scopes) verified_email = None if claims.get("email_verified") is True: raw_email = claims.get("email") if not isinstance(raw_email, str): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token verified email claim is invalid", status_code=401, ) verified_email = raw_email issued_at = claims.get("iat") if isinstance(issued_at, bool) or not isinstance(issued_at, int): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token issue-time claim is invalid", status_code=401, ) authentication_time = None auth_time = claims.get("auth_time") if auth_time is not None: if ( isinstance(auth_time, bool) or not isinstance(auth_time, int) or auth_time > issued_at ): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token authentication-time claim is invalid", status_code=401, ) try: authentication_time = datetime.fromtimestamp( auth_time, tz=timezone.utc, ) except (OSError, OverflowError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token authentication-time claim is invalid", status_code=401, ) from exc try: return AuthenticatedPrincipal( issuer=claims.get("iss"), subject=claims.get("sub"), scopes=scopes, verified_email=verified_email, authentication_time=authentication_time, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "bearer token identity claims are invalid", status_code=401, ) from exc __all__ = [ "AuthenticatedPrincipal", "OIDCTokenValidator", "TokenValidator", ]