Repository navigation
chore: Resolves Issue #92, updates to use transitive dependency explicit #134
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
ff4c63f
202832c
2337e5a
d13849d
501d1d2
e67688e
0e9d19b
b32f14f
42e209b
7782e9e
529128c
b4ed67e
b80c5c9
02e40bb
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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 | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. With Is this intended? If so, it needs a mention in the docs.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Should these be listed in the changelog or docs so 3.10+ users aren't surprised? |
||
| # 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 | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. On 3.10+ the decoded proof is used as a dict straight away, but Shall we check the result is a dict right after decode and raise
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Is the change in return type intended? If so it's worth a note in the changelog. |
||
| 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"]) | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
joserfcis imported directly here but isn't listed inpyproject.tomlorrequirements.txt. It only arrives throughauthlib1.7 and above, andauthlib = "^1.0"still allows older versions that don't bring it. On Python 3.10+ with one of those, importing the package would fail.Should we add
joserfcas a dependency for 3.10+, or raise theauthlibminimum? WDYT?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
3.9 is well past EOL. I'd raise the minimum?