diff --git a/packages/auth0_api_python/src/auth0_api_python/api_client.py b/packages/auth0_api_python/src/auth0_api_python/api_client.py index b38409e..cf10da9 100644 --- a/packages/auth0_api_python/src/auth0_api_python/api_client.py +++ b/packages/auth0_api_python/src/auth0_api_python/api_client.py @@ -1,11 +1,13 @@ import time -from typing import Optional, List, Dict, Any +import hashlib +from typing import Optional, List, Dict, Any, Tuple from authlib.jose import JsonWebToken, JsonWebKey from .config import ApiClientOptions -from .errors import MissingRequiredArgumentError, VerifyAccessTokenError -from .utils import fetch_oidc_metadata, fetch_jwks, get_unverified_header +from .errors import MissingRequiredArgumentError, VerifyAccessTokenError, InvalidAuthSchemeError, InvalidDpopProofError +from .utils import fetch_oidc_metadata, fetch_jwks, get_unverified_header, normalize_url_for_htu, sha256_base64url, calculate_jwk_thumbprint +from .types import RequestData class ApiClient: """ @@ -26,6 +28,42 @@ def __init__(self, options: ApiClientOptions): self._jwt = JsonWebToken(["RS256"]) + self._dpop_algorithms = ["ES256"] + self._dpop_jwt = JsonWebToken(self._dpop_algorithms) + + + def _build_www_authenticate( + self, + *, + dpop_error: Optional[str] = None, + dpop_error_description: Optional[str] = None + ) -> List[Tuple[str, str]]: + """ + Returns one or two ('WWW-Authenticate', ...) tuples. + If dpop_required mode → single DPoP challenge (with optional error params). + Otherwise → Bearer realm=…, then DPoP algs=… (with error params). + """ + realm = getattr(self.options, "realm", None) or f'https://{self.options.domain}' + algs = " ".join(self._dpop_algorithms) + + # build the DPoP piece + dpop_parts = [f'algs="{algs}"'] + if dpop_error: + dpop_parts.append(f'error="{dpop_error}"') + if dpop_error_description: + dpop_parts.append(f'error_description="{dpop_error_description}"') + dpop_header = "DPoP " + ", ".join(dpop_parts) + + if getattr(self.options, "dpop_required", False): + return [("WWW-Authenticate", dpop_header)] + + # allowed mode: first Bearer, then DPoP + bearer_header = f'Bearer realm="{realm}"' + return [ + ("WWW-Authenticate", bearer_header), + ("WWW-Authenticate", dpop_header), + ] + async def _discover(self) -> Dict[str, Any]: """Lazy-load OIDC discovery metadata.""" if self._metadata is None: @@ -125,4 +163,193 @@ async def verify_access_token( if rc not in claims: raise VerifyAccessTokenError(f"Missing required claim: {rc}") - return claims \ No newline at end of file + return claims + + async def verify_dpop_proof( + self, + access_token: str, + proof: str, + http_method: str, + http_url: str + ) -> Dict[str, Any]: + """ + 1. Single well-formed compact JWS + 2. typ="dpop+jwt", alg∈allowed, alg≠none + 3. jwk header present & public only + 4. Signature verifies with jwk + 5. iat within leeway + 6. htm == http_method + 7. htu == http_url (normalized) + 8. ath == SHA256(access_token) + Raises InvalidDpopProofError on any failure. + """ + if not proof: + raise MissingRequiredArgumentError("dpop_proof") + if not access_token: + raise MissingRequiredArgumentError("access_token") + if not http_method or not http_url: + raise MissingRequiredArgumentError("http_method/http_url") + + header = await get_unverified_header(proof) + + if header.get("typ") != "dpop+jwt": + raise InvalidDpopProofError("Invalid typ header") + + alg = header.get("alg") + if alg not in self._dpop_algorithms: + raise InvalidDpopProofError(f"Unsupported alg: {alg}") + + self._validate_claims_presence(claims, ["iat", "ath", "htm", "htu", "jti"]) + + jwk_dict = header.get("jwk") + + if "d" in jwk_dict: + raise InvalidDpopProofError("Private key material found in jwk header") + + if jwk_dict.get("kty") != "EC": + raise InvalidDpopProofError("Only EC keys are supported for DPoP") + + if jwk_dict.get("crv") != "P-256": + raise InvalidDpopProofError("Only P-256 curve is supported") + + + public_key = JsonWebKey.import_key(jwk_dict) + try: + claims = self._dpop_jwt.decode(proof, public_key) + except Exception as e: + raise InvalidDpopProofError(f"Signature verification failed: {e}") + + + + jti = claims["jti"] + + if not isinstance(jti, str): + raise InvalidDpopProofError("jti claim must be a string") + + if not jti.strip(): + raise InvalidDpopProofError("jti claim must not be empty") + + now = int(time.time()) + iat = claims.get("iat") + + offset = getattr(self.options, "dpop_iat_offset", 300) # default 5 minutes + leeway = getattr(self.options, "dpop_iat_leeway", 30) # default 30 seconds + + if not isinstance(iat, int): + raise InvalidDpopProofError("Missing or invalid iat claim") + + if iat < now - offset or iat > now + leeway: + raise InvalidDpopProofError("DPoP proof iat outside allowed window") + + if claims.get("htm") != http_method: + raise InvalidDpopProofError("htm claim mismatch") + + if normalize_url_for_htu(claims.get("htu","")) != normalize_url_for_htu(http_url): + raise InvalidDpopProofError("htu claim mismatch") + + if claims.get("ath") != sha256_base64url(access_token): + raise InvalidDpopProofError("ath claim mismatch") + + return claims + + async def verify_request( + self, + request: RequestData, # More specific than Mapping[str, Any] + ) -> Dict[str, Any]: + """ + Dispatch based on Authorization scheme: + • If scheme is 'DPoP', verifies both access token and DPoP proof + • If scheme is 'Bearer', verifies only the access token + + Args: + request: A mapping containing: + - authorization_header: The Authorization header value (required) + - dpop_proof: The DPoP proof header value (required for DPoP) + - http_method: The HTTP method (required for DPoP) + - http_url: The HTTP URL (required for DPoP) + + Returns: + The decoded access token claims + + Raises: + MissingRequiredArgumentError: If required args are missing + InvalidAuthSchemeError: If an unsupported scheme is provided + InvalidDpopProofError: If DPoP verification fails + VerifyAccessTokenError: If access token verification fails + """ + + + authorization_header = request.get("authorization_header", "") + dpop_proof = request.get("dpop_proof") + http_method = request.get("http_method") + http_url = request.get("http_url") + + if not authorization_header: + raise MissingRequiredArgumentError("authorization_header") + if authorization_header is not None: + if isinstance(authorization_header, list) and len(authorization_header) > 1: + raise InvalidAuthSchemeError("Multiple Authorization headers are not allowed") + try: + scheme, token = authorization_header.split(" ", 1) + except ValueError: + raise InvalidAuthSchemeError("Malformed Authorization header (expected ' ')") + + scheme = scheme.strip().lower() + + if scheme == "dpop": + if not self.options.dpop_enabled: + raise InvalidAuthSchemeError("DPoP is disabled") + if not dpop_proof: + raise MissingRequiredArgumentError("dpop_proof") + if not http_method or not http_url: + raise MissingRequiredArgumentError( + "http_method and http_url are required for DPoP" + ) + access_token_claims = await self.verify_access_token(token) + await self.verify_dpop_proof( + access_token=token, + proof=dpop_proof, + http_method=http_method, + http_url=http_url + ) + jwk_dict = (await get_unverified_header(dpop_proof))["jwk"] + actual_jkt = calculate_jwk_thumbprint(jwk_dict) + expected_jkt = access_token_claims.get("cnf", {}).get("jkt") + if expected_jkt != actual_jkt: + raise InvalidDpopProofError("cnf.jkt thumbprint mismatch") + + return access_token_claims + + if scheme == "bearer": + return await self.verify_access_token(token) + + raise InvalidAuthSchemeError(f"Unsupported auth scheme: {scheme}") + + def _validate_claims_presence( + self, + claims: Dict[str, Any], + required_claims: List[str] + ) -> None: + """ + Validates that all required claims are present in the claims dict. + + Args: + claims: The claims dictionary to validate + required_claims: List of claim names that must be present + + Raises: + InvalidDpopProofError: If any required claim is missing + """ + missing_claims = [] + + for claim in required_claims: + if claim not in claims: + missing_claims.append(claim) + + if missing_claims: + if len(missing_claims) == 1: + error_message = f"Missing required claim: {missing_claims[0]}" + else: + error_message = f"Missing required claims: {', '.join(missing_claims)}" + + raise InvalidDpopProofError(error_message) diff --git a/packages/auth0_api_python/src/auth0_api_python/config.py b/packages/auth0_api_python/src/auth0_api_python/config.py index de2f4f8..8418542 100644 --- a/packages/auth0_api_python/src/auth0_api_python/config.py +++ b/packages/auth0_api_python/src/auth0_api_python/config.py @@ -17,8 +17,14 @@ def __init__( self, domain: str, audience: str, - custom_fetch: Optional[Callable[..., object]] = None + custom_fetch: Optional[Callable[..., object]] = None, + realm: Optional[str] = None ): self.domain = domain self.audience = audience self.custom_fetch = custom_fetch + self.realm = realm + self.dpop_enabled = True + self.dpop_required = False + self.dpop_iat_leeway = 30 + self.dpop_iat_offset = 300 diff --git a/packages/auth0_api_python/src/auth0_api_python/errors.py b/packages/auth0_api_python/src/auth0_api_python/errors.py index e450059..33f6fdb 100644 --- a/packages/auth0_api_python/src/auth0_api_python/errors.py +++ b/packages/auth0_api_python/src/auth0_api_python/errors.py @@ -19,3 +19,20 @@ class VerifyAccessTokenError(Exception): def __init__(self, message: str): super().__init__(message) self.name = self.__class__.__name__ + +class InvalidAuthSchemeError(Exception): + """Error raised when the provided authentication scheme is unsupported.""" + code = "invalid_auth_scheme" + + def __init__(self, scheme: str): + super().__init__(f"Unsupported authentication scheme: '{scheme}'") + self.scheme = scheme + self.name = self.__class__.__name__ + +class InvalidDpopProofError(Exception): + """Error raised when validating a DPoP proof fails.""" + code = "invalid_dpop_proof" + + def __init__(self, message: str): + super().__init__(message) + self.name = self.__class__.__name__ diff --git a/packages/auth0_api_python/src/auth0_api_python/types.py b/packages/auth0_api_python/src/auth0_api_python/types.py new file mode 100644 index 0000000..a5b0c57 --- /dev/null +++ b/packages/auth0_api_python/src/auth0_api_python/types.py @@ -0,0 +1,7 @@ +from typing import TypedDict, Optional + +class RequestData(TypedDict): + authorization_header: str + dpop_proof: Optional[str] + http_method: str + http_url: str \ No newline at end of file diff --git a/packages/auth0_api_python/src/auth0_api_python/utils.py b/packages/auth0_api_python/src/auth0_api_python/utils.py index 2d66ecb..e3835b8 100644 --- a/packages/auth0_api_python/src/auth0_api_python/utils.py +++ b/packages/auth0_api_python/src/auth0_api_python/utils.py @@ -6,8 +6,11 @@ import httpx import base64 import json +import hashlib from typing import Any, Dict, Optional, Callable, Union +from urllib.parse import urlparse, urlunparse + async def fetch_oidc_metadata( domain: str, custom_fetch: Optional[Callable[..., Any]] = None @@ -85,4 +88,62 @@ def remove_bytes_prefix(s: str) -> str: """If the string looks like b'eyJh...', remove the leading b' and trailing '.""" if s.startswith("b'"): return s[2:] # cut off the leading b' - return s \ No newline at end of file + return s + +def normalize_url_for_htu(raw_url: str) -> str: + """ + Strip query and fragment from the URL so it can be compared to the + DPoP proof's htu claim (RFC 3986 §6.2.2/6.2.3). + """ + p = urlparse(raw_url) + return urlunparse((p.scheme, p.netloc, p.path, "", "", "")) + + +def sha256_base64url(input_str: str) -> str: + """ + Compute SHA-256 digest of the input string and return a + Base64URL-encoded string *without* padding. + """ + digest = hashlib.sha256(input_str.encode("utf-8")).digest() + b64 = base64.urlsafe_b64encode(digest).decode("utf-8") + return b64.rstrip("=") + +def calculate_jwk_thumbprint(jwk: Dict[str, str]) -> str: + """ + Compute the RFC 7638 JWK thumbprint for a public JWK. + + - For EC keys, includes only: crv, kty, x, y + - For RSA keys, includes only: e, kty, n + - For OKP keys, includes only: crv, kty, x + - For oct keys, includes only: k, kty + - Serializes with no whitespace, keys sorted lexicographically + - Hashes with SHA-256 and returns base64url-encoded string without padding + """ + kty = jwk.get("kty") + + if kty == "EC": + if not all(k in jwk for k in ["crv", "x", "y"]): + raise ValueError("EC key missing required parameters") + members = ("crv", "kty", "x", "y") + elif kty == "RSA": + if not all(k in jwk for k in ["e", "n"]): + raise ValueError("RSA key missing required parameters") + members = ("e", "kty", "n") + elif kty == "OKP": + if not all(k in jwk for k in ["crv", "x"]): + raise ValueError("OKP key missing required parameters") + members = ("crv", "kty", "x") + elif kty == "oct": + if "k" not in jwk: + raise ValueError("oct key missing required parameter") + members = ("k", "kty") + else: + raise ValueError(f"Unsupported key type: {kty}") + + ordered = {k: jwk[k] for k in members if k in jwk} + + thumbprint_json = json.dumps(ordered, separators=(",", ":"), sort_keys=True) + + digest = hashlib.sha256(thumbprint_json.encode("utf-8")).digest() + + return base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=") \ No newline at end of file