diff --git a/CHANGELOG.md b/CHANGELOG.md index 306d4c9..8610101 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,20 @@ # Change Log +## Unreleased + +**Changed** +- Migrated JOSE operations to `joserfc` on Python 3.10+ while retaining Authlib + JOSE on Python 3.9. The `joserfc` dependency is now declared explicitly for + Python 3.10+. +- Added configurable `jwt_algorithms` for access-token signatures (default + `RS256`); only recognized asymmetric algorithms are accepted. JWKS `alg` + restrictions are enforced consistently across Python versions. +- Both JOSE backends now return plain dictionary claims. On Python 3.10+, + `joserfc` validates JOSE header names/types, header and payload sizes, and + key `use`, `key_ops`, and `alg` metadata more strictly than legacy Authlib. + Incompatible tokens or keys may require migration. +- Malformed DPoP proof payloads and unusable JWKS keys now raise SDK-specific + verification errors instead of exposing lower-level exceptions. + ## [1.0.0b10](https://github.com/auth0/auth0-api-python/tree/1.0.0b10) (2026-08-03) [Full Changelog](https://github.com/auth0/auth0-api-python/compare/1.0.0b9...1.0.0b10) diff --git a/README.md b/README.md index 4e48f65..ffafcee 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,36 @@ This SDK provides comprehensive support for securing APIs with Auth0-issued acce - **Framework Agnostic** - Works with FastAPI, Django, Flask, or any Python web framework ## Documentation +## JWT algorithms and Python-version compatibility + +Access-token verification accepts `RS256` by default. To allow another asymmetric +signature algorithm, supply `jwt_algorithms` when constructing `ApiClientOptions`: + +```python +ApiClientOptions( + domain="example.us.auth0.com", + audience="my-api", + jwt_algorithms=["RS256", "RS384"], +) +``` + +The list must be non-empty and contain supported `RS*`, `PS*`, or `ES*` algorithms. +Symmetric (`HS*`), unsigned (`none`), and unrecognized algorithms are rejected at +construction. The configuration is copied when the client is constructed. The JWT's +declared algorithm must also match any `alg` restriction set on the selected JWKS +key; configuring `RS384` does not override a key declared for `RS256`. A tenant +that issues RS384 tokens must publish an RS384-capable key. + +Python 3.9 uses Authlib's legacy JOSE implementation; Python 3.10+ uses +`joserfc` directly. Both return decoded access-token and DPoP claims as plain +dictionaries, and invalid key material produces SDK verification errors. +`joserfc` imposes additional strict validation of JOSE headers (including +registered header names and their types), encoded header/payload sizes, and +JWK usage restrictions (`use`, `key_ops`, and `alg`). Tokens accepted by +the legacy implementation may be rejected on newer runtimes if they violate +these requirements. DPoP remains restricted to `ES256` regardless of +`jwt_algorithms`. + - [Docs Site](https://auth0.com/docs) - explore our docs site and learn more about Auth0. diff --git a/poetry.lock b/poetry.lock index c874bc7..f96a209 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1377,4 +1377,4 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""] [metadata] lock-version = "2.1" python-versions = ">=3.9.2" -content-hash = "f78b7a13854c1a3aedd7ae7d4c4d0685dbeb549a5d847abd08bb98561f043b9d" +content-hash = "60c437f68a0e6647bbcef9663008838a970ab7855b20e27285756d68a1855960" diff --git a/pyproject.toml b/pyproject.toml index 75b7fb8..ab670a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,8 @@ packages = [ [tool.poetry.dependencies] python = ">=3.9.2" -authlib = "^1.0" # For JWT/OIDC features +authlib = "^1.0" # For JWT/OIDC features and Python 3.9 legacy JOSE +joserfc = {version = ">=1.6.8,<2.0", python = ">=3.10"} # Used directly on Python 3.10+ requests = "^2.31.0" # If you use requests for HTTP calls (e.g., discovery) httpx = "^0.28.1" cryptography = ">=43.0.1" diff --git a/requirements.txt b/requirements.txt index 19089c2..47692c1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ # Source of truth for dependencies: pyproject.toml + poetry.lock authlib>=1.6.12 +joserfc>=1.6.8,<2.0; python_version >= "3.10" httpx>=0.28.1 ada-url>=1.32.0; python_version >= "3.10" ada-url>=1.27.0; python_version < "3.10" diff --git a/src/auth0_api_python/_internal/jose_compat.py b/src/auth0_api_python/_internal/jose_compat.py new file mode 100644 index 0000000..e41cfbc --- /dev/null +++ b/src/auth0_api_python/_internal/jose_compat.py @@ -0,0 +1,6 @@ +"""Select the supported JOSE implementation for the current Python runtime.""" + +import sys + +# Authlib JOSE is retained for Python 3.9; newer runtimes use joserfc. +USE_LEGACY_JOSE = sys.version_info < (3, 10) diff --git a/src/auth0_api_python/api_client.py b/src/auth0_api_python/api_client.py index f200996..50229ed 100644 --- a/src/auth0_api_python/api_client.py +++ b/src/auth0_api_python/api_client.py @@ -6,9 +6,9 @@ from typing import Any, Optional, Union import httpx -from authlib.jose import JsonWebKey, JsonWebToken from ._internal.cache_expiry import is_fresh +from ._internal.jose_compat import USE_LEGACY_JOSE from ._internal.cache_keys import m2m_cache_key from ._internal.obo_cache import OboCache from ._internal.token_vault_cache import TokenVaultCache @@ -48,8 +48,23 @@ sha256_base64url, ) +# Authlib 1.6 remains the supported JOSE implementation on Python 3.9. +# joserfc 1.7+ (and Authlib's dependency on it) requires Python 3.10+. +if USE_LEGACY_JOSE: + from authlib.jose import JsonWebKey, JsonWebToken +else: + from joserfc import jwk, jwt + logger = logging.getLogger(__name__) +# Access tokens must use a supported asymmetric signature algorithm. +# Never allow "none" (unsigned JWTs) or symmetric algorithms with public JWKS keys. +SUPPORTED_JWT_ALGORITHMS = frozenset({ + "RS256", "RS384", "RS512", + "PS256", "PS384", "PS512", + "ES256", "ES384", "ES512", +}) + # Token Exchange constants TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" # noqa: S105 OBO_ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" # noqa: S105 @@ -77,6 +92,17 @@ def __init__(self, options: ApiClientOptions): if not options.audience: raise MissingRequiredArgumentError("audience") + if not isinstance(options.jwt_algorithms, list) or not options.jwt_algorithms: + raise ConfigurationError("jwt_algorithms must be a non-empty list") + for algorithm in options.jwt_algorithms: + if not isinstance(algorithm, str) or algorithm not in SUPPORTED_JWT_ALGORITHMS: + raise ConfigurationError( + f"Unsupported JWT signing algorithm: {algorithm!r}. " + "Only RS*, PS*, and ES* asymmetric algorithms are supported." + ) + # Do not allow mutation of the caller's list to change accepted algorithms. + self._jwt_algorithms = list(options.jwt_algorithms) + # Validate domains parameter if provided if options.domains is not None: if isinstance(options.domains, list): @@ -182,10 +208,10 @@ def __init__(self, options: ApiClientOptions): self._cache_ttl = options.cache_ttl_seconds - self._jwt = JsonWebToken(["RS256"]) - self._dpop_algorithms = ["ES256"] - self._dpop_jwt = JsonWebToken(self._dpop_algorithms) + if USE_LEGACY_JOSE: + self._jwt = JsonWebToken(self._jwt_algorithms) + self._dpop_jwt = JsonWebToken(self._dpop_algorithms) def is_dpop_required(self) -> bool: """Check if DPoP authentication is required.""" @@ -473,7 +499,7 @@ async def verify_access_token( Asynchronously verifies the provided JWT access token. - Fetches OIDC metadata and JWKS if not already cached. - - Decodes and validates signature (RS256) with the correct key. + - Decodes and validates the signature with the configured jwt_algorithms (default RS256). - Checks standard claims: 'iss', 'aud', 'exp', 'nbf', 'iat' - Checks extra required claims if 'required_claims' is provided. - Enforces organization_policy: requires 'org_id' when set to "required", @@ -516,6 +542,8 @@ async def verify_access_token( f"Symmetric algorithm '{alg}' is not supported. " "Only asymmetric algorithms (e.g., RS256) are allowed." ) + if alg not in self._jwt_algorithms: + raise VerifyAccessTokenError(f"JWT signing algorithm '{alg}' is not allowed") # Extract and validate issuer claim (before network calls) try: @@ -602,13 +630,25 @@ async def verify_access_token( if not matching_key_dict: raise VerifyAccessTokenError("No matching key found in JWKS") - # Import public key and verify signature - public_key = JsonWebKey.import_key(matching_key_dict) + # Respect the key's declared algorithm on both JOSE backends. + # A JWKS key for RS256 must not silently verify an RS384 token. + key_alg = matching_key_dict.get("alg") + if key_alg is not None and key_alg != alg: + raise VerifyAccessTokenError("JWT algorithm does not match JWKS key algorithm") if isinstance(access_token, str) and access_token.startswith("b'"): access_token = access_token[2:-1] try: - claims = self._jwt.decode(access_token, public_key) + if USE_LEGACY_JOSE: + public_key = JsonWebKey.import_key(matching_key_dict) + claims = self._jwt.decode(access_token, public_key) + else: + public_key = jwk.import_key(matching_key_dict) + claims = jwt.decode(access_token, public_key, algorithms=self._jwt_algorithms).claims + # Return the same plain-dict contract on Python 3.9 and 3.10+. + if not isinstance(claims, dict): + raise ValueError("JWT payload is not a JSON object") + claims = dict(claims) except Exception as e: raise VerifyAccessTokenError(f"Signature verification failed: {str(e)}") from e @@ -704,15 +744,24 @@ async def verify_dpop_proof( raise InvalidDpopProofError("Only P-256 curve is supported") try: - public_key = JsonWebKey.import_key(jwk_dict) + if USE_LEGACY_JOSE: + public_key = JsonWebKey.import_key(jwk_dict) + else: + public_key = jwk.import_key(jwk_dict) except Exception: raise InvalidDpopProofError("Missing or invalid jwk in header") try: - claims = self._dpop_jwt.decode(proof, public_key) + if USE_LEGACY_JOSE: + claims = self._dpop_jwt.decode(proof, public_key) + else: + claims = jwt.decode(proof, public_key, algorithms=self._dpop_algorithms).claims except Exception as e: raise InvalidDpopProofError(f"JWT signature verification failed: {e}") + if not isinstance(claims, dict): + raise InvalidDpopProofError("DPoP proof payload must be a JSON object") + # Checks all required claims are present self._validate_claims_presence(claims, ["iat", "ath", "htm", "htu", "jti"]) diff --git a/src/auth0_api_python/config.py b/src/auth0_api_python/config.py index 7275c21..964d565 100644 --- a/src/auth0_api_python/config.py +++ b/src/auth0_api_python/config.py @@ -47,6 +47,9 @@ class ApiClientOptions: match. "non_strict" reuses any cached token whose granted scopes cover the request, and requires a token_store subclassing IndexedTokenStore (raises ConfigurationError at construction without one). + jwt_algorithms: Non-empty list of supported asymmetric JWT signing algorithms. + Defaults to ["RS256"]; unknown, symmetric, and unsigned algorithms + raise ConfigurationError at ApiClient construction. """ def __init__( self, @@ -68,10 +71,12 @@ def __init__( organization_policy: Literal["required", "allow"] = "allow", organization_id: Optional[Union[str, list[str]]] = None, scope_matching: str = "strict", + jwt_algorithms: Optional[list[str]] = None, ): self.domain = domain self.domains = domains self.audience = audience + self.jwt_algorithms = ["RS256"] if jwt_algorithms is None else jwt_algorithms self.custom_fetch = custom_fetch self.cache_adapter = cache_adapter self.token_store = token_store diff --git a/src/auth0_api_python/token_utils.py b/src/auth0_api_python/token_utils.py index c234681..27a2ef1 100644 --- a/src/auth0_api_python/token_utils.py +++ b/src/auth0_api_python/token_utils.py @@ -1,11 +1,19 @@ +import base64 +import json import time import uuid from typing import Any, Optional, Union -from authlib.jose import JsonWebKey, jwt - +from ._internal.jose_compat import USE_LEGACY_JOSE from .utils import calculate_jwk_thumbprint, normalize_url_for_htu, sha256_base64url +# Python 3.9 resolves to Authlib 1.6 without joserfc. +# Use the legacy Authlib JOSE API on that runtime. +if USE_LEGACY_JOSE: + from authlib.jose import JsonWebKey, jwt +else: + from joserfc import jwk, jwt + # A private RSA JWK for test usage. PRIVATE_JWK = { @@ -33,9 +41,10 @@ async def generate_token( exp: bool = True, claims: Optional[dict[str, Any]] = None, expiration_time: int = 3600, + algorithm: str = "RS256", ) -> str: """ - Generates a real RS256-signed JWT using the private key above. + Generates a signed JWT using the private key above. Args: domain: The Auth0 domain (used if issuer is not False). @@ -49,9 +58,10 @@ async def generate_token( exp: Whether to set the 'exp' claim. If False, skip it. claims: Additional custom claims to merge into the token. expiration_time: If exp is True, how many seconds from now until expiration. + algorithm: Signing algorithm (default RS256); tests may use RS384, for example. Returns: - A RS256-signed JWT string. + A signed JWT string. Example usage: token = generate_token( @@ -80,9 +90,13 @@ async def generate_token( token_claims["aud"] = audience - key = JsonWebKey.import_key(PRIVATE_JWK) + key_data = {**PRIVATE_JWK, "alg": algorithm} + if USE_LEGACY_JOSE: + key = JsonWebKey.import_key(key_data) + else: + key = jwk.import_key(key_data) - header = {"alg": "RS256", "kid": PRIVATE_JWK["kid"]} + header = {"alg": algorithm, "kid": PRIVATE_JWK["kid"]} token = jwt.encode(header, token_claims, key) # Ensure we return a string, not bytes return token.decode('utf-8') if isinstance(token, bytes) else token @@ -163,13 +177,25 @@ async def generate_dpop_proof( } - if header_overrides: - header.update(header_overrides) + if USE_LEGACY_JOSE: + key = JsonWebKey.import_key(PRIVATE_EC_JWK) + else: + key = jwk.import_key(PRIVATE_EC_JWK) - key = JsonWebKey.import_key(PRIVATE_EC_JWK) + # Sign a valid proof first. Malformed-header fixtures are intentionally + # tampered *after* signing so no production registry overrides are needed. token = jwt.encode(header, proof_claims, key) - # Ensure we return a string, not bytes - return token.decode('utf-8') if isinstance(token, bytes) else token + token = token.decode("utf-8") if isinstance(token, bytes) else token + if header_overrides: + protected, payload, signature = token.split(".") + padded = protected + "=" * (-len(protected) % 4) + modified_header = json.loads(base64.urlsafe_b64decode(padded)) + modified_header.update(header_overrides) + protected = base64.urlsafe_b64encode( + json.dumps(modified_header, separators=(",", ":")).encode("utf-8") + ).decode("ascii").rstrip("=") + token = f"{protected}.{payload}.{signature}" + return token async def generate_token_with_cnf( diff --git a/tests/test_api_client.py b/tests/test_api_client.py index d5d7a36..2615232 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -3,6 +3,7 @@ import json import logging import time +from types import SimpleNamespace import httpx import pytest @@ -40,6 +41,7 @@ token_vault_cache_key, ) from auth0_api_python.api_client import MAX_ARRAY_VALUES_PER_KEY, ApiClient +from auth0_api_python._internal.jose_compat import USE_LEGACY_JOSE from auth0_api_python.config import ApiClientOptions from auth0_api_python.errors import ( ApiError, @@ -86,6 +88,132 @@ async def test_init_missing_args(): _ = ApiClient(ApiClientOptions(domain="example.us.auth0.com", audience="")) +def test_custom_jwt_algorithms_are_copied(): + algorithms = ["RS384", "RS512"] + client = ApiClient(ApiClientOptions( + domain="example.us.auth0.com", + audience="my-audience", + jwt_algorithms=algorithms, + )) + assert client._jwt_algorithms == ["RS384", "RS512"] + algorithms.append("RS256") + assert client._jwt_algorithms == ["RS384", "RS512"] + + +@pytest.mark.parametrize("algorithms", [ + [], (), "RS256", [None], [""], [" "], ["FOO"], + ["none"], ["RS256", "none"], ["HS256"], ["RS256", 7], +]) +def test_reject_invalid_jwt_algorithms(algorithms): + with pytest.raises(ConfigurationError): + ApiClient(ApiClientOptions( + domain="example.us.auth0.com", + audience="my-audience", + jwt_algorithms=algorithms, + )) + + +@pytest.mark.asyncio +async def test_verify_access_token_rs384_requires_explicit_configuration(httpx_mock: HTTPXMock): + """A genuine RS384 signature is rejected by default, accepted when enabled.""" + token = await generate_token( + domain="auth0.local", user_id="user_123", + audience="my-audience", algorithm="RS384", + ) + httpx_mock.add_response( + method="GET", url=DISCOVERY_URL, + json={"issuer": "https://auth0.local/", "jwks_uri": JWKS_URL}, + ) + # A key without an 'alg' restriction may verify another RS* algorithm. + jwks_key = {key: value for key, value in PUBLIC_RSA_JWK.items() if key != "alg"} + httpx_mock.add_response(method="GET", url=JWKS_URL, json={"keys": [jwks_key]}) + default_client = ApiClient(ApiClientOptions(domain="auth0.local", audience="my-audience")) + with pytest.raises(VerifyAccessTokenError, match="not allowed"): + await default_client.verify_access_token(token) + + configured = ApiClient(ApiClientOptions( + domain="auth0.local", audience="my-audience", jwt_algorithms=["RS384"], + )) + claims = await configured.verify_access_token(token) + assert isinstance(claims, dict) + assert claims["sub"] == "user_123" + + +@pytest.mark.asyncio +async def test_verify_access_token_rejects_conflicting_jwks_key_algorithm(httpx_mock: HTTPXMock): + """Both Python JOSE backends enforce JWK algorithm restrictions.""" + token = await generate_token( + domain="auth0.local", user_id="user_123", + audience="my-audience", algorithm="RS384", + ) + httpx_mock.add_response( + method="GET", url=DISCOVERY_URL, + json={"issuer": "https://auth0.local/", "jwks_uri": JWKS_URL}, + ) + httpx_mock.add_response(method="GET", url=JWKS_URL, json={"keys": [PUBLIC_RSA_JWK]}) + client = ApiClient(ApiClientOptions( + domain="auth0.local", audience="my-audience", jwt_algorithms=["RS384"], + )) + with pytest.raises(VerifyAccessTokenError, match="JWKS key algorithm"): + await client.verify_access_token(token) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("key_changes", [ + {"n": None}, + {"key_ops": ["encrypt"]}, +]) +async def test_verify_access_token_bad_jwks_key_raises_sdk_error(httpx_mock: HTTPXMock, key_changes): + """Malformed or unusable JWKS material never leaks a JOSE exception.""" + token = await generate_token(domain="auth0.local", user_id="user_123", audience="my-audience") + httpx_mock.add_response( + method="GET", url=DISCOVERY_URL, + json={"issuer": "https://auth0.local/", "jwks_uri": JWKS_URL}, + ) + bad_key = {**PUBLIC_RSA_JWK, **key_changes} + if bad_key.get("n") is None: + bad_key.pop("n", None) + httpx_mock.add_response(method="GET", url=JWKS_URL, json={"keys": [bad_key]}) + client = ApiClient(ApiClientOptions(domain="auth0.local", audience="my-audience")) + with pytest.raises(VerifyAccessTokenError): + await client.verify_access_token(token) + + +@pytest.mark.asyncio +async def test_unsigned_jwt_is_never_supported(): + """The none algorithm must be rejected even before a JWKS lookup.""" + client = ApiClient(ApiClientOptions(domain="auth0.local", audience="my-audience")) + unsigned = unsigned_jwt( + {"alg": "none", "kid": "TEST_KEY"}, + {"iss": "https://auth0.local/", "aud": "my-audience"}, + ) + with pytest.raises(VerifyAccessTokenError, match="not allowed"): + await client.verify_access_token(unsigned) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("payload", [5, None, True, "not-an-object"]) +async def test_verify_dpop_proof_rejects_non_object_decoded_payload(payload, monkeypatch): + """A decoded non-object must be reported as InvalidDpopProofError, not TypeError.""" + token = "test_token" + proof = await generate_dpop_proof( + access_token=token, http_method="GET", http_url="https://api.example.com/resource", + ) + client = ApiClient(ApiClientOptions(domain="auth0.local", audience="my-audience")) + if USE_LEGACY_JOSE: + monkeypatch.setattr(client._dpop_jwt, "decode", lambda *_args, **_kwargs: payload) + else: + monkeypatch.setattr( + "auth0_api_python.api_client.jwt.decode", + lambda *_args, **_kwargs: SimpleNamespace(claims=payload), + ) + with pytest.raises(InvalidDpopProofError, match="JSON object"): + await client.verify_dpop_proof( + access_token=token, proof=proof, + http_method="GET", http_url="https://api.example.com/resource", + ) + + @pytest.mark.asyncio async def test_verify_access_token_successfully(httpx_mock: HTTPXMock): """