From 68cce691ee81a35e746d9d4751a6c0a7a6a6af37 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Tue, 1 Jul 2025 22:22:05 +0530 Subject: [PATCH 1/6] POC for DPoP support in Auth0 API Python SDK --- .../src/auth0_api_python/api_client.py | 150 +++++++++++++++++- .../src/auth0_api_python/config.py | 7 +- .../src/auth0_api_python/errors.py | 17 ++ .../src/auth0_api_python/utils.py | 23 ++- 4 files changed, 191 insertions(+), 6 deletions(-) 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..f00bff9 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,12 @@ 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 class ApiClient: """ @@ -26,6 +27,34 @@ 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) -> List[Tuple[str,str]]: + """ + Build one or two WWW-Authenticate headers: + - Required mode: single DPoP challenge + - Allowed mode: Bearer + DPoP challenges + """ + realm = self.options.realm \ + or f'https://{self.options.domain}' + algs = " ".join(self._dpop_algorithms) + + # 1) If DPoP *required*, only send a DPoP header + if self.options.dpop_required: + return [ + ("WWW-Authenticate", f'DPoP algs="{algs}"') + ] + + # 2) Otherwise, send Bearer then DPoP + bearer = f'Bearer realm="{realm}"' + dpop = f'DPoP algs="{algs}"' + return [ + ("WWW-Authenticate", bearer), + ("WWW-Authenticate", dpop) + ] + async def _discover(self) -> Dict[str, Any]: """Lazy-load OIDC discovery metadata.""" if self._metadata is None: @@ -125,4 +154,117 @@ 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.options.dpop_algorithms: + raise InvalidDpopProofError(f"Unsupported alg: {alg}") + + + jwk_dict = header.get("jwk") + if not jwk_dict or "d" in jwk_dict: + raise InvalidDpopProofError("Missing or private jwk in header") + + + 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}") + + now = int(time.time()) + iat = claims.get("iat") + + if not isinstance(iat, int): + raise InvalidDpopProofError("Missing or invalid iat claim") + leeway = getattr(self.options, "dpop_iat_leeway", 30) + if abs(now - iat) > leeway: + raise InvalidDpopProofError("iat timestamp check failed") + + 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, + authorization_header: str, + dpop_proof: Optional[str], + http_method: Optional[str], + http_url: Optional[str] + ) -> Dict[str, Any]: + """ + Dispatch based on Authorization scheme: + • If scheme is 'DPoP', calls verify_dpop_request() + • Else treats as Bearer and calls verify_access_token() + + Raises: + MissingRequiredArgumentError if required args are missing + InvalidDpopSchemeError if an unsupported scheme is provided + """ + + if not authorization_header: + raise MissingRequiredArgumentError("authorization_header") + 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" + ) + return await self.verify_dpop_proof( + access_token=token, + dpop_proof=dpop_proof, + http_method=http_method, + http_url=http_url + ) + + if scheme == "bearer": + return await self.verify_access_token(token) + + raise InvalidAuthSchemeError(f"Unsupported auth scheme: {scheme}") \ No newline at end of file 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..fcb745b 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,13 @@ 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 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/utils.py b/packages/auth0_api_python/src/auth0_api_python/utils.py index 2d66ecb..c2bbfd3 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,22 @@ 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("=") \ No newline at end of file From 2d76d3006cdeb0bcd7df06b70e8b6dea403cd006 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Wed, 2 Jul 2025 00:59:10 +0530 Subject: [PATCH 2/6] Fixing the dpop_algorithms variable --- packages/auth0_api_python/src/auth0_api_python/api_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 f00bff9..0f4b2ea 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 @@ -187,7 +187,7 @@ async def verify_dpop_proof( raise InvalidDpopProofError("Invalid typ header") alg = header.get("alg") - if alg not in self.options.dpop_algorithms: + if alg not in self._dpop_algorithms: raise InvalidDpopProofError(f"Unsupported alg: {alg}") From fbcf744fb43ec0be21f1b6def77f240a4ac590df Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Mon, 7 Jul 2025 19:56:18 +0530 Subject: [PATCH 3/6] Changes for POC Feedback --- .../src/auth0_api_python/api_client.py | 50 +++++++++++-------- .../src/auth0_api_python/config.py | 1 + 2 files changed, 31 insertions(+), 20 deletions(-) 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 0f4b2ea..8902101 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 @@ -31,28 +31,34 @@ def __init__(self, options: ApiClientOptions): self._dpop_jwt = JsonWebToken(self._dpop_algorithms) - def _build_www_authenticate(self) -> List[Tuple[str,str]]: + def _build_www_authenticate( + self, + *, + dpop_error: Optional[str] = None, + dpop_error_description: Optional[str] = None + ) -> List[Tuple[str, str]]: """ - Build one or two WWW-Authenticate headers: - - Required mode: single DPoP challenge - - Allowed mode: Bearer + DPoP challenges + 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 = self.options.realm \ - or f'https://{self.options.domain}' + realm = getattr(self.options, "realm", None) or f'https://{self.options.domain}' algs = " ".join(self._dpop_algorithms) - # 1) If DPoP *required*, only send a DPoP header - if self.options.dpop_required: - return [ - ("WWW-Authenticate", f'DPoP algs="{algs}"') - ] + 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)] - # 2) Otherwise, send Bearer then DPoP - bearer = f'Bearer realm="{realm}"' - dpop = f'DPoP algs="{algs}"' + bearer_header = f'Bearer realm="{realm}"' return [ - ("WWW-Authenticate", bearer), - ("WWW-Authenticate", dpop) + ("WWW-Authenticate", bearer_header), + ("WWW-Authenticate", dpop_header), ] async def _discover(self) -> Dict[str, Any]: @@ -205,11 +211,14 @@ async def verify_dpop_proof( 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") - leeway = getattr(self.options, "dpop_iat_leeway", 30) - if abs(now - iat) > leeway: - raise InvalidDpopProofError("iat timestamp check failed") + + 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") @@ -257,9 +266,10 @@ async def verify_request( raise MissingRequiredArgumentError( "http_method and http_url are required for DPoP" ) + await self.verify_access_token(token) return await self.verify_dpop_proof( access_token=token, - dpop_proof=dpop_proof, + proof=dpop_proof, http_method=http_method, http_url=http_url ) 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 fcb745b..8418542 100644 --- a/packages/auth0_api_python/src/auth0_api_python/config.py +++ b/packages/auth0_api_python/src/auth0_api_python/config.py @@ -27,3 +27,4 @@ def __init__( self.dpop_enabled = True self.dpop_required = False self.dpop_iat_leeway = 30 + self.dpop_iat_offset = 300 From 8be3d7c3f4e1ae1ba07c9491c11b703d1a349127 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Tue, 8 Jul 2025 22:42:01 +0530 Subject: [PATCH 4/6] fix: verify_request accepts a Dict/Mapping --- .../src/auth0_api_python/api_client.py | 46 ++++++++++++++----- .../src/auth0_api_python/types.py | 7 +++ .../src/auth0_api_python/utils.py | 27 ++++++++++- 3 files changed, 68 insertions(+), 12 deletions(-) create mode 100644 packages/auth0_api_python/src/auth0_api_python/types.py 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 8902101..fa5ba8b 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 @@ -6,7 +6,8 @@ from .config import ApiClientOptions from .errors import MissingRequiredArgumentError, VerifyAccessTokenError, InvalidAuthSchemeError, InvalidDpopProofError -from .utils import fetch_oidc_metadata, fetch_jwks, get_unverified_header, normalize_url_for_htu, sha256_base64url +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: """ @@ -45,6 +46,7 @@ def _build_www_authenticate( 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}"') @@ -55,6 +57,7 @@ def _build_www_authenticate( 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), @@ -233,21 +236,35 @@ async def verify_dpop_proof( async def verify_request( self, - authorization_header: str, - dpop_proof: Optional[str], - http_method: Optional[str], - http_url: Optional[str] + request: RequestData, # More specific than Mapping[str, Any] ) -> Dict[str, Any]: """ Dispatch based on Authorization scheme: - • If scheme is 'DPoP', calls verify_dpop_request() - • Else treats as Bearer and calls verify_access_token() + • 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 - InvalidDpopSchemeError if an unsupported scheme is provided + 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") try: @@ -266,13 +283,20 @@ async def verify_request( raise MissingRequiredArgumentError( "http_method and http_url are required for DPoP" ) - await self.verify_access_token(token) - return await self.verify_dpop_proof( + 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) 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 c2bbfd3..3131881 100644 --- a/packages/auth0_api_python/src/auth0_api_python/utils.py +++ b/packages/auth0_api_python/src/auth0_api_python/utils.py @@ -106,4 +106,29 @@ def sha256_base64url(input_str: str) -> str: """ digest = hashlib.sha256(input_str.encode("utf-8")).digest() b64 = base64.urlsafe_b64encode(digest).decode("utf-8") - return b64.rstrip("=") \ No newline at end of file + 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 + - 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": + members = ("crv", "kty", "x", "y") + elif kty == "RSA": + members = ("e", "kty", "n") + else: + members = tuple(sorted(k for k,v in jwk.items() if isinstance(v, str))) + + 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 From 31979c22e0fd5ceceb7739acf1a3b729c74c5686 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Tue, 8 Jul 2025 22:52:52 +0530 Subject: [PATCH 5/6] calculate_jwk_thumbprint now matches "jose" logic --- .../src/auth0_api_python/utils.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) 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 3131881..e3835b8 100644 --- a/packages/auth0_api_python/src/auth0_api_python/utils.py +++ b/packages/auth0_api_python/src/auth0_api_python/utils.py @@ -114,20 +114,35 @@ def calculate_jwk_thumbprint(jwk: Dict[str, str]) -> str: - 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: - members = tuple(sorted(k for k,v in jwk.items() if isinstance(v, str))) + 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) + thumbprint_json = json.dumps(ordered, separators=(",", ":"), sort_keys=True) digest = hashlib.sha256(thumbprint_json.encode("utf-8")).digest() From 0303484624c07128c66876adca2ace16d66fe84e Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Thu, 24 Jul 2025 16:57:57 +0530 Subject: [PATCH 6/6] security: Fixes for SEC Tickets Feedback --- .../src/auth0_api_python/api_client.py | 57 ++++++++++++++++++- 1 file changed, 54 insertions(+), 3 deletions(-) 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 fa5ba8b..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 @@ -199,10 +199,18 @@ async def verify_dpop_proof( 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 not jwk_dict or "d" in jwk_dict: - raise InvalidDpopProofError("Missing or private jwk in header") + + 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) @@ -210,6 +218,16 @@ async def verify_dpop_proof( 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") @@ -259,6 +277,7 @@ async def verify_request( InvalidDpopProofError: If DPoP verification fails VerifyAccessTokenError: If access token verification fails """ + authorization_header = request.get("authorization_header", "") dpop_proof = request.get("dpop_proof") @@ -267,6 +286,9 @@ async def verify_request( 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: @@ -301,4 +323,33 @@ async def verify_request( if scheme == "bearer": return await self.verify_access_token(token) - raise InvalidAuthSchemeError(f"Unsupported auth scheme: {scheme}") \ No newline at end of file + 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)