diff --git a/packages/google-auth/google/auth/dpop.py b/packages/google-auth/google/auth/dpop.py new file mode 100644 index 000000000000..b60881a7cbda --- /dev/null +++ b/packages/google-auth/google/auth/dpop.py @@ -0,0 +1,494 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Demonstrating Proof-of-Possession (DPoP) support per RFC 9449. + +This module provides the abstract base class :class:`DPoPSigner` and a +concrete implementation :class:`SoftwareDPoPSigner`, as well as +cryptographic helpers for creating DPoP proof JWTs and computing +RFC 7638 JWK thumbprints. + +Third-party keystores (such as hardware tokens, TPMs, YubiKeys, +PKCS#11 devices, or cloud KMS) can be integrated by subclassing +:class:`DPoPSigner` and implementing its :attr:`algorithm`, +:attr:`public_jwk`, and :meth:`sign` methods. +""" + +import abc +import hashlib +import json +import time +import urllib.parse +import uuid + +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa +from cryptography.hazmat.primitives.asymmetric.utils import ( + decode_dss_signature, + encode_dss_signature, +) + +from google.auth import _helpers, exceptions + + +def compute_jwk_thumbprint(jwk): + """Computes the RFC 7638 SHA-256 JSON Web Key (JWK) Thumbprint. + + The thumbprint is calculated by canonicalizing the required public + key members in lexicographical order, encoding as UTF-8 JSON with no + extra whitespace, and computing the unpadded URL-safe base64 SHA-256 + hash. + + Args: + jwk (Mapping[str, Any]): The public JWK dictionary. + + Returns: + str: The base64url-encoded SHA-256 thumbprint (used for + ``dpop_jkt``). + + Raises: + google.auth.exceptions.InvalidValue: If the JWK is missing + required fields for its key type. + """ + kty = jwk.get("kty") + if kty == "EC": + required = ["crv", "kty", "x", "y"] + elif kty == "RSA": + required = ["e", "kty", "n"] + else: + raise exceptions.InvalidValue( + "Unsupported key type '{}' for JWK thumbprint calculation.".format(kty) + ) + + for field in required: + if field not in jwk: + raise exceptions.InvalidValue( + "JWK is missing required field '{}' for key type '{}'.".format( + field, kty + ) + ) + + canonical_dict = {field: jwk[field] for field in required} + canonical_json = json.dumps( + canonical_dict, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + + digest = hashlib.sha256(canonical_json).digest() + return _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + + +def compute_access_token_hash(access_token): + """Computes the RFC 9449 access token hash (``ath`` claim). + + The ``ath`` claim value is the unpadded base64url encoding of the + SHA-256 hash of the ASCII-encoded access token string. + + Args: + access_token (Union[str, bytes]): The access token. + + Returns: + str: Base64url-encoded SHA-256 hash of the token. + """ + token_bytes = _helpers.to_bytes(access_token) + digest = hashlib.sha256(token_bytes).digest() + return _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + + +def compute_auth_code_jti(code): + """Computes the DPoP ``jti`` claim for authorization code exchange. + + Google's DPoP implementation requires + ``jti = BASE64URL(SHA-256(code))`` for authorization code exchanges + to cryptographically bind the proof to the issued authorization code. + + Args: + code (Union[str, bytes]): The authorization code received from + Google. + + Returns: + str: Unpadded Base64URL-encoded SHA-256 hash of the + authorization code. + """ + code_bytes = _helpers.to_bytes(code) + digest = hashlib.sha256(code_bytes).digest() + return _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + + +def normalize_dpop_url(url): + """Normalizes a URL for the DPoP ``htu`` claim per RFC 9449 §4.3. + + The URL is normalized to lowercase scheme and host, with the path + preserved and any query string or fragment stripped. + + Args: + url (str): The target HTTP URI. + + Returns: + str: The normalized URI without query parameters or fragment. + """ + parsed = urllib.parse.urlsplit(url) + scheme = parsed.scheme.lower() + netloc = parsed.netloc.lower() + path = parsed.path or "/" + return urllib.parse.urlunsplit((scheme, netloc, path, "", "")) + + +def der_to_raw_ecdsa_signature(der_signature, curve_byte_size=32): + """Converts an ASN.1 DER ECDSA signature to raw (R || S) JWS format. + + Many hardware keystores, PKCS#11 libraries, and OS security APIs + output ECDSA signatures in ASN.1 DER format. JWS (RFC 7515 + Section 3.4) and DPoP mandate raw (R || S) signature bytes. + + Args: + der_signature (bytes): The ASN.1 DER-encoded ECDSA signature. + curve_byte_size (int): The coordinate size in bytes (32 for + P-256/ES256, 48 for P-384/ES384). Default is 32. + + Returns: + bytes: Raw (R || S) concatenated byte signature of length + ``2 * curve_byte_size``. + """ + r, s = decode_dss_signature(der_signature) + return r.to_bytes(curve_byte_size, byteorder="big") + s.to_bytes( + curve_byte_size, byteorder="big" + ) + + +def raw_to_der_ecdsa_signature(raw_signature): + """Converts a raw (R || S) ECDSA signature to ASN.1 DER format. + + Args: + raw_signature (bytes): The raw concatenated (R || S) signature. + + Returns: + bytes: ASN.1 DER-encoded signature. + """ + half_len = len(raw_signature) // 2 + r = int.from_bytes(raw_signature[:half_len], byteorder="big") + s = int.from_bytes(raw_signature[half_len:], byteorder="big") + return encode_dss_signature(r, s) + + +class DPoPSigner(metaclass=abc.ABCMeta): + """Abstract base class for DPoP signers. + + This interface allows custom keystore backends — such as TPMs, + YubiKeys, smart cards (PKCS#11), Apple Secure Enclave, Windows CNG, + or Cloud KMS — to supply DPoP signatures without exposing private + key material. + """ + + @property + @abc.abstractmethod + def algorithm(self): + """str: The JWS signing algorithm (e.g. ``'ES256'``, ``'RS256'``).""" + raise NotImplementedError("Derived classes must implement algorithm") + + @property + @abc.abstractmethod + def public_jwk(self): + """Mapping[str, Any]: The public key as an RFC 7517 JWK dict. + + Must NOT contain any private key components (e.g. ``'d'``, + ``'p'``, ``'q'``). + """ + raise NotImplementedError("Derived classes must implement public_jwk") + + @property + def key_id(self): + """Optional[str]: An identifier for this key.""" + return None + + @property + def thumbprint(self): + """str: RFC 7638 SHA-256 JWK thumbprint.""" + return compute_jwk_thumbprint(self.public_jwk) + + @abc.abstractmethod + def sign(self, data): + """Signs the input bytes. + + For ECDSA algorithms (e.g. ES256), the return value MUST be raw + concatenated (R || S) bytes per RFC 7515 Section 3.4 (e.g. 64 + bytes for P-256), not ASN.1 DER. See + :func:`der_to_raw_ecdsa_signature`. + + Args: + data (bytes): The ASCII data to sign (typically + ``b'.'``). + + Returns: + bytes: The signature bytes. + """ + raise NotImplementedError("Derived classes must implement sign") + + def create_proof( + self, + method, + url, + nonce=None, + access_token=None, + jti=None, + extra_claims=None, + ): + """Creates a signed RFC 9449 DPoP proof JWT. + + Args: + method (str): The HTTP method (e.g. ``'POST'``, ``'GET'``). + url (str): The target HTTP URI. + nonce (Optional[str]): A server-provided DPoP nonce, if + available. + access_token (Optional[Union[str, bytes]]): The access token + to bind via the ``ath`` claim per RFC 9449 §4.3. + Google resource servers currently ignore this header + entirely; it is included for forward compatibility. + jti (Optional[str]): Explicit unique JWT ID. If omitted, a + random UUIDv4 is generated. For Google OAuth code + exchange, this MUST be ``compute_auth_code_jti(code)``. + extra_claims (Optional[Mapping[str, Any]]): Additional + custom claims to include in the payload. + + Returns: + str: The compact serialized DPoP proof JWT + (``
..``). + """ + header = { + "typ": "dpop+jwt", + "alg": self.algorithm, + "jwk": dict(self.public_jwk), + } + + payload = { + "jti": jti or str(uuid.uuid4()), + "htm": method.upper(), + "htu": normalize_dpop_url(url), + "iat": int(time.time()), + } + + if nonce: + payload["nonce"] = nonce + + if access_token: + payload["ath"] = compute_access_token_hash(access_token) + + if extra_claims: + payload.update(extra_claims) + + header_b64 = _helpers.unpadded_urlsafe_b64encode( + json.dumps(header, separators=(",", ":")).encode("utf-8") + ) + payload_b64 = _helpers.unpadded_urlsafe_b64encode( + json.dumps(payload, separators=(",", ":")).encode("utf-8") + ) + + signing_input = header_b64 + b"." + payload_b64 + signature = self.sign(signing_input) + sig_b64 = _helpers.unpadded_urlsafe_b64encode(signature) + + return (header_b64 + b"." + payload_b64 + b"." + sig_b64).decode("ascii") + + +class SoftwareDPoPSigner(DPoPSigner): + """Concrete software-backed DPoP signer using ``cryptography``. + + Supports NIST curves (P-256 for ES256, P-384 for ES384) and RSA + (RS256). + + The private key must be created externally and supplied either as a + ``cryptography`` key object (via the constructor) or as PEM data + (via :meth:`from_pem`). + + .. tip:: Generating a key with ``openssl`` + + .. code-block:: bash + + # ES256 (recommended) + openssl ecparam -name prime256v1 -genkey -noout \\ + -out dpop_key.pem + + # ES384 + openssl ecparam -name secp384r1 -genkey -noout \\ + -out dpop_key.pem + + # RS256 + openssl genrsa -out dpop_key.pem 2048 + + .. note:: + + For production deployments, prefer a hardware-backed signer + (YubiKey, TPM, Cloud KMS) by subclassing :class:`DPoPSigner`. + Software keys are suitable for development and testing. + + Args: + private_key (Union[ec.EllipticCurvePrivateKey, + rsa.RSAPrivateKey]): The private key. + key_id (Optional[str]): An optional key identifier. + + Raises: + google.auth.exceptions.InvalidValue: If the key type or curve + is not supported. + """ + + def __init__(self, private_key, key_id=None): + """ + Args: + private_key (Union[ec.EllipticCurvePrivateKey, + rsa.RSAPrivateKey]): The private key. + key_id (Optional[str]): An optional key identifier. + """ + self._private_key = private_key + self._key_id = key_id + + if isinstance(private_key, ec.EllipticCurvePrivateKey): + curve = private_key.curve + if isinstance(curve, ec.SECP256R1): + self._alg = "ES256" + self._crv_size = 32 + crv_name = "P-256" + elif isinstance(curve, ec.SECP384R1): + self._alg = "ES384" + self._crv_size = 48 + crv_name = "P-384" + else: + raise exceptions.InvalidValue( + "Unsupported elliptic curve: {}".format(curve.name) + ) + + pub_numbers = private_key.public_key().public_numbers() + x_bytes = pub_numbers.x.to_bytes(self._crv_size, byteorder="big") + y_bytes = pub_numbers.y.to_bytes(self._crv_size, byteorder="big") + + self._public_jwk = { + "kty": "EC", + "crv": crv_name, + "x": _helpers.unpadded_urlsafe_b64encode(x_bytes).decode("ascii"), + "y": _helpers.unpadded_urlsafe_b64encode(y_bytes).decode("ascii"), + } + elif isinstance(private_key, rsa.RSAPrivateKey): + self._alg = "RS256" + self._crv_size = None + pub_numbers = private_key.public_key().public_numbers() + e_bytes = pub_numbers.e.to_bytes( + (pub_numbers.e.bit_length() + 7) // 8, byteorder="big" + ) + n_bytes = pub_numbers.n.to_bytes( + (pub_numbers.n.bit_length() + 7) // 8, byteorder="big" + ) + + self._public_jwk = { + "kty": "RSA", + "n": _helpers.unpadded_urlsafe_b64encode(n_bytes).decode("ascii"), + "e": _helpers.unpadded_urlsafe_b64encode(e_bytes).decode("ascii"), + } + else: + raise exceptions.InvalidValue( + "Unsupported private key type: {}".format(type(private_key)) + ) + + @property + def algorithm(self): + """str: The JWS signing algorithm name.""" + return self._alg + + @property + def public_jwk(self): + """Mapping[str, Any]: The public JWK dictionary.""" + return self._public_jwk + + @property + def key_id(self): + """Optional[str]: The key identifier.""" + return self._key_id + + @property + def private_key(self): + """Union[ec.EllipticCurvePrivateKey, rsa.RSAPrivateKey]: The + underlying private key object.""" + return self._private_key + + def sign(self, data): + """Signs the input bytes using the private key. + + Args: + data (bytes): The data to sign. + + Returns: + bytes: The signature bytes (raw R||S for EC, PKCS1v15 for + RSA). + + Raises: + google.auth.exceptions.InvalidValue: If the private key + type is not supported. + """ + if isinstance(self._private_key, ec.EllipticCurvePrivateKey): + sha_algo = hashes.SHA256() if self._alg == "ES256" else hashes.SHA384() + der_sig = self._private_key.sign(data, ec.ECDSA(sha_algo)) + return der_to_raw_ecdsa_signature(der_sig, self._crv_size) + elif isinstance(self._private_key, rsa.RSAPrivateKey): + return self._private_key.sign( + data, + padding.PKCS1v15(), + hashes.SHA256(), + ) + else: + raise exceptions.InvalidValue("Unsupported private key type for signing.") + + @classmethod + def from_pem(cls, pem_bytes, password=None, key_id=None): + """Loads a signer from PEM-encoded private key data. + + Args: + pem_bytes (Union[str, bytes]): The PEM private key data. + password (Optional[bytes]): Passphrase if the key is + encrypted. + key_id (Optional[str]): Optional key identifier. + + Returns: + SoftwareDPoPSigner: The constructed signer. + + Raises: + google.auth.exceptions.InvalidValue: If the loaded key is + not an EC or RSA private key. + """ + pem_data = _helpers.to_bytes(pem_bytes) + key = serialization.load_pem_private_key(pem_data, password=password) + if not isinstance( + key, + ( + ec.EllipticCurvePrivateKey, + rsa.RSAPrivateKey, + ), + ): + raise exceptions.InvalidValue( + "Loaded key must be an EC or RSA private key, got {}".format(type(key)) + ) + return cls(key, key_id=key_id) + + +def extract_dpop_nonce(headers): + """Extracts the ``DPoP-Nonce`` header value (case-insensitive). + + Args: + headers (Mapping[str, str]): HTTP response headers. + + Returns: + Optional[str]: The nonce value, or ``None`` if not present. + """ + if not headers: + return None + for key in headers: + if key.lower() == "dpop-nonce": + return headers[key] + return None diff --git a/packages/google-auth/google/oauth2/_client.py b/packages/google-auth/google/oauth2/_client.py index 1afa7ab55d32..f1abf2c8e235 100644 --- a/packages/google-auth/google/oauth2/_client.py +++ b/packages/google-auth/google/oauth2/_client.py @@ -147,6 +147,7 @@ def _token_endpoint_request_no_throw( use_json=False, can_retry=True, headers=None, + return_response_headers=False, **kwargs, ): """Makes a request to the OAuth 2.0 authorization server's token endpoint. @@ -163,6 +164,9 @@ def _token_endpoint_request_no_throw( content type. The default value is False. can_retry (bool): Enable or disable request retry behavior. headers (Optional[Mapping[str, str]]): The headers for the request. + return_response_headers (Optional[bool]): If True, the response + headers will be included as a fourth element in the returned + tuple. The default value is False. kwargs: Additional arguments passed on to the request method. The kwargs will be passed to `requests.request` method, see: https://docs.python-requests.org/en/latest/api/#requests.request. @@ -175,7 +179,8 @@ def _token_endpoint_request_no_throw( Tuple(bool, Mapping[str, str], Optional[bool]): A boolean indicating if the request is successful, a mapping for the JSON-decoded response data and in the case of an error a boolean indicating if the error - is retryable. + is retryable. If ``return_response_headers`` is True, a fourth + element containing the response headers mapping is appended. """ if use_json: headers_to_use = {"Content-Type": _JSON_CONTENT_TYPE} @@ -192,12 +197,14 @@ def _token_endpoint_request_no_throw( response_data = {} retryable_error = False + response_headers = {} retries = _exponential_backoff.ExponentialBackoff() for _ in retries: response = request( method="POST", url=token_uri, headers=headers_to_use, body=body, **kwargs ) + response_headers = getattr(response, "headers", {}) response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") @@ -211,6 +218,8 @@ def _token_endpoint_request_no_throw( response_data = response_body if response.status == http_client.OK: + if return_response_headers: + return True, response_data, None, response_headers return True, response_data, None retryable_error = _can_retry( @@ -218,8 +227,12 @@ def _token_endpoint_request_no_throw( ) if not can_retry or not retryable_error: + if return_response_headers: + return False, response_data, retryable_error, response_headers return False, response_data, retryable_error + if return_response_headers: + return False, response_data, retryable_error, response_headers return False, response_data, retryable_error @@ -246,6 +259,9 @@ def _token_endpoint_request( content type. The default value is False. can_retry (bool): Enable or disable request retry behavior. headers (Optional[Mapping[str, str]]): The headers for the request. + return_response_headers (Optional[bool]): If True, the response + headers will be included as a fourth element in the returned + tuple. The default value is False. kwargs: Additional arguments passed on to the request method. The kwargs will be passed to `requests.request` method, see: https://docs.python-requests.org/en/latest/api/#requests.request. diff --git a/packages/google-auth/google/oauth2/credentials.py b/packages/google-auth/google/oauth2/credentials.py index cff8647d3bd9..0a3424c1fa3b 100644 --- a/packages/google-auth/google/oauth2/credentials.py +++ b/packages/google-auth/google/oauth2/credentials.py @@ -434,7 +434,7 @@ def _perform_refresh_token(self, request): expiry, grant_response, rapt_token, - ) = reauth.refresh_grant( + ) = self._call_refresh_grant( request, self._token_uri, self._refresh_token, @@ -442,7 +442,6 @@ def _perform_refresh_token(self, request): self._client_secret, scopes=scopes, rapt_token=self._rapt_token, - enable_reauth_refresh=self._enable_reauth_refresh, ) self.token = access_token @@ -468,6 +467,51 @@ def _perform_refresh_token(self, request): ) ) + def _call_refresh_grant( + self, + request, + token_uri, + refresh_token, + client_id, + client_secret, + scopes=None, + rapt_token=None, + ): + """Performs the token refresh grant call. + + Subclasses may override this to customize the token request, + for example to add DPoP proof-of-possession headers. + + Args: + request (google.auth.transport.Request): The object used to make + HTTP requests. + token_uri (str): The OAuth 2.0 token endpoint URI. + refresh_token (str): The refresh token. + client_id (str): The OAuth 2.0 client ID. + client_secret (str): The OAuth 2.0 client secret. + scopes (Optional[Sequence[str]]): The requested scopes. + rapt_token (Optional[str]): The reauth Proof Token. + + Returns: + Tuple[str, str, Optional[datetime], Mapping[str, str], str]: + The access token, new refresh token, expiration, the full + response data, and the rapt token. + + Raises: + google.auth.exceptions.RefreshError: If the token endpoint + returned an error. + """ + return reauth.refresh_grant( + request, + token_uri, + refresh_token, + client_id, + client_secret, + scopes=scopes, + rapt_token=rapt_token, + enable_reauth_refresh=self._enable_reauth_refresh, + ) + @classmethod def from_authorized_user_info(cls, info, scopes=None): """Creates a Credentials instance from parsed authorized user info. diff --git a/packages/google-auth/google/oauth2/dpop_credentials.py b/packages/google-auth/google/oauth2/dpop_credentials.py new file mode 100644 index 000000000000..5e5598ad857c --- /dev/null +++ b/packages/google-auth/google/oauth2/dpop_credentials.py @@ -0,0 +1,392 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""OAuth 2.0 credentials with DPoP (RFC 9449) proof-of-possession. + +This module provides :class:`DPoPCredentials`, a subclass of +:class:`google.oauth2.credentials.Credentials` that binds OAuth 2.0 +refresh and access tokens to a cryptographic key using Demonstrating +Proof-of-Possession (DPoP) as defined in `RFC 9449`_. + +In Google's implementation, DPoP proofs are validated by the +authorization server during token refresh and code exchange. A +stolen refresh token cannot be used without the corresponding +private key. + +Usage with an externally-generated PEM key:: + + from google.auth.dpop import SoftwareDPoPSigner + from google.oauth2.dpop_credentials import DPoPCredentials + + signer = SoftwareDPoPSigner.from_pem( + open("dpop_key.pem", "rb").read() + ) + creds = DPoPCredentials.from_authorized_user_file( + "credentials.json", dpop_signer=signer + ) + +.. _RFC 9449: https://www.rfc-editor.org/rfc/rfc9449 +""" + +import io +import json + +from google.auth import exceptions, metrics +from google.auth.dpop import extract_dpop_nonce +from google.oauth2 import _client, reauth +from google.oauth2 import credentials as _oauth2_credentials + +_USE_DPOP_NONCE_ERROR = "use_dpop_nonce" + + +class DPoPCredentials(_oauth2_credentials.Credentials): + """OAuth 2.0 credentials with DPoP proof-of-possession. + + This is a subclass of :class:`google.oauth2.credentials.Credentials` + that adds DPoP proof generation to token refresh requests, binding + OAuth 2.0 tokens to a cryptographic key per `RFC 9449`_. + + In Google's implementation, DPoP proofs are validated at the + authorization server (token endpoint) during token refresh and + code exchange. This prevents stolen refresh tokens from being + used without the corresponding private key. + + .. _RFC 9449: https://www.rfc-editor.org/rfc/rfc9449 + """ + + def __init__(self, token, dpop_signer=None, dpop_nonce=None, **kwargs): + """ + Args: + token (Optional[str]): The OAuth 2.0 access token. Can be + None if refresh information is provided. + dpop_signer (Optional[google.auth.dpop.DPoPSigner]): The + DPoP signer used to create proof-of-possession JWTs. + When provided, DPoP proofs are attached to token + refresh requests sent to Google's authorization + server. + dpop_nonce (Optional[str]): A server-provided DPoP nonce + from a previous token response. If available, it will + be included in the next DPoP proof. + **kwargs: All remaining keyword arguments are forwarded to + :class:`google.oauth2.credentials.Credentials`. + """ + super(DPoPCredentials, self).__init__(token, **kwargs) + self._dpop_signer = dpop_signer + self._dpop_nonce = dpop_nonce + + @property + def dpop_signer(self): + """Optional[google.auth.dpop.DPoPSigner]: The DPoP signer.""" + return self._dpop_signer + + @property + def dpop_nonce(self): + """Optional[str]: The server-provided DPoP nonce.""" + return self._dpop_nonce + + def _after_refresh(self, request, method, url, headers): + """Attaches a DPoP proof for the resource server request. + + After credentials are refreshed, this hook creates a DPoP proof + JWT bound to the target URL and the current access token (via + the ``ath`` claim) per RFC 9449 §4.3. + + .. note:: + + Google resource servers currently ignore the ``DPoP`` + header entirely — the ``Authorization`` header uses the + ``Bearer`` scheme and the ``DPoP`` proof is not validated. + This header is included for RFC 9449 compliance and + forward compatibility with stricter enforcement. + + Args: + request (google.auth.transport.Request): The object used to + make HTTP requests. + method (str): The HTTP method of the request. + url (str): The target resource server URI. + headers (Mapping): The request headers to update. + """ + super(DPoPCredentials, self)._after_refresh(request, method, url, headers) + if self._dpop_signer is not None and self.token: + headers["DPoP"] = self._dpop_signer.create_proof( + method, + url, + nonce=self._dpop_nonce, + access_token=self.token, + ) + + def _call_refresh_grant( + self, + request, + token_uri, + refresh_token, + client_id, + client_secret, + scopes=None, + rapt_token=None, + ): + """DPoP-aware token refresh grant. + + Creates a DPoP proof for the token endpoint, handles + ``use_dpop_nonce`` challenges with a single retry, and supports + the reauth/RAPT flow when ``enable_reauth_refresh`` is set. + + Args: + request (google.auth.transport.Request): The object used to + make HTTP requests. + token_uri (str): The OAuth 2.0 token endpoint URI. + refresh_token (str): The refresh token. + client_id (str): The OAuth 2.0 client ID. + client_secret (str): The OAuth 2.0 client secret. + scopes (Optional[Sequence[str]]): The requested scopes. + rapt_token (Optional[str]): The reauth Proof Token. + + Returns: + Tuple[str, str, Optional[datetime], Mapping[str, str], str]: + The access token, new refresh token, expiration, the + full response data, and the rapt token. + + Raises: + google.auth.exceptions.RefreshError: If the token endpoint + returned an error. + ValueError: If no ``dpop_signer`` has been set on this + instance. + """ + if self._dpop_signer is None: + raise ValueError( + "DPoPCredentials requires a dpop_signer for token " + "refresh. Use google.oauth2.credentials.Credentials " + "for non-DPoP flows." + ) + + body = { + "grant_type": _client._REFRESH_GRANT_TYPE, + "client_id": client_id, + "client_secret": client_secret, + "refresh_token": refresh_token, + } + if scopes: + body["scope"] = " ".join(scopes) + if rapt_token: + body["rapt"] = rapt_token + + headers = {metrics.API_CLIENT_HEADER: metrics.token_request_user()} + headers["DPoP"] = self._dpop_signer.create_proof( + "POST", token_uri, nonce=self._dpop_nonce + ) + + ( + response_status_ok, + response_data, + retryable_error, + response_headers, + ) = _client._token_endpoint_request_no_throw( + request, + token_uri, + body, + headers=headers, + return_response_headers=True, + ) + + # Always capture the server-provided nonce from any response. + self._dpop_nonce = extract_dpop_nonce(response_headers) or self._dpop_nonce + + # Handle use_dpop_nonce challenge — retry once with the new + # nonce the server just gave us. + if ( + not response_status_ok + and isinstance(response_data, dict) + and response_data.get("error") == _USE_DPOP_NONCE_ERROR + ): + headers["DPoP"] = self._dpop_signer.create_proof( + "POST", token_uri, nonce=self._dpop_nonce + ) + ( + response_status_ok, + response_data, + retryable_error, + response_headers, + ) = _client._token_endpoint_request_no_throw( + request, + token_uri, + body, + headers=headers, + return_response_headers=True, + ) + self._dpop_nonce = extract_dpop_nonce(response_headers) or self._dpop_nonce + + # Handle reauth/RAPT challenge. + if ( + not response_status_ok + and isinstance(response_data, dict) + and response_data.get("error") == reauth._REAUTH_NEEDED_ERROR + and ( + response_data.get("error_subtype") + == reauth._REAUTH_NEEDED_ERROR_INVALID_RAPT + or response_data.get("error_subtype") + == reauth._REAUTH_NEEDED_ERROR_RAPT_REQUIRED + ) + ): + if not self._enable_reauth_refresh: + raise exceptions.RefreshError( + "Reauthentication is needed. Please run " + "`gcloud auth application-default login` to " + "reauthenticate." + ) + + rapt_token = reauth.get_rapt_token( + request, + client_id, + client_secret, + refresh_token, + token_uri, + scopes=scopes, + ) + body["rapt"] = rapt_token + headers["DPoP"] = self._dpop_signer.create_proof( + "POST", token_uri, nonce=self._dpop_nonce + ) + ( + response_status_ok, + response_data, + retryable_error, + response_headers, + ) = _client._token_endpoint_request_no_throw( + request, + token_uri, + body, + headers=headers, + return_response_headers=True, + ) + self._dpop_nonce = extract_dpop_nonce(response_headers) or self._dpop_nonce + + if not response_status_ok: + _client._handle_error_response(response_data, retryable_error) + + return _client._handle_refresh_grant_response(response_data, refresh_token) + ( + rapt_token, + ) + + def _make_copy(self): + cred = self.__class__( + self.token, + dpop_signer=self._dpop_signer, + dpop_nonce=self._dpop_nonce, + refresh_token=self.refresh_token, + id_token=self.id_token, + token_uri=self.token_uri, + client_id=self.client_id, + client_secret=self.client_secret, + scopes=self.scopes, + default_scopes=self.default_scopes, + granted_scopes=self.granted_scopes, + quota_project_id=self.quota_project_id, + rapt_token=self.rapt_token, + enable_reauth_refresh=self._enable_reauth_refresh, + trust_boundary=self._trust_boundary, + universe_domain=self._universe_domain, + account=self._account, + ) + cred._cred_file_path = self._cred_file_path + return cred + + def to_json(self, strip=None): + """Utility function that creates a JSON representation of a + DPoPCredentials object. + + Args: + strip (Sequence[str]): Optional list of members to exclude + from the generated JSON. + + Returns: + str: A JSON representation of this instance, including + DPoP metadata (nonce, key ID, thumbprint). Private key + material is never included. + """ + prep = json.loads(super(DPoPCredentials, self).to_json(strip)) + + if self._dpop_nonce: + prep["dpop_nonce"] = self._dpop_nonce + if self._dpop_signer is not None: + prep["dpop_thumbprint"] = self._dpop_signer.thumbprint + if self._dpop_signer.key_id is not None: + prep["dpop_key_id"] = self._dpop_signer.key_id + + return json.dumps(prep) + + @classmethod + def from_authorized_user_info(cls, info, scopes=None, dpop_signer=None): + """Creates a DPoPCredentials instance from parsed authorized + user info. + + Args: + info (Mapping[str, str]): The authorized user info in + Google format. + scopes (Sequence[str]): Optional list of scopes to include + in the credentials. + dpop_signer (Optional[google.auth.dpop.DPoPSigner]): The + DPoP signer to bind to these credentials. + + Returns: + google.oauth2.dpop_credentials.DPoPCredentials: The + constructed credentials. + + Raises: + ValueError: If the info is not in the expected format, or + if a ``dpop_thumbprint`` is present in the info and + does not match the provided signer's thumbprint. + """ + # Let the parent parse and create using cls() — which will be + # DPoPCredentials with dpop_signer=None initially. + creds = super(DPoPCredentials, cls).from_authorized_user_info(info, scopes) + + saved_thumbprint = info.get("dpop_thumbprint") + if dpop_signer is not None and saved_thumbprint: + if dpop_signer.thumbprint != saved_thumbprint: + raise ValueError( + "DPoP signer thumbprint '{}' does not match the " + "thumbprint '{}' saved in the credentials. The " + "refresh token is bound to the original key and " + "cannot be used with a different signer.".format( + dpop_signer.thumbprint, saved_thumbprint + ) + ) + + creds._dpop_signer = dpop_signer + creds._dpop_nonce = info.get("dpop_nonce") + return creds + + @classmethod + def from_authorized_user_file(cls, filename, scopes=None, dpop_signer=None): + """Creates a DPoPCredentials instance from an authorized user + json file. + + Args: + filename (str): The path to the authorized user json file. + scopes (Sequence[str]): Optional list of scopes to include + in the credentials. + dpop_signer (Optional[google.auth.dpop.DPoPSigner]): The + DPoP signer to bind to these credentials. + + Returns: + google.oauth2.dpop_credentials.DPoPCredentials: The + constructed credentials. + + Raises: + ValueError: If the file is not in the expected format. + """ + with io.open(filename, "r", encoding="utf-8") as json_file: + data = json.load(json_file) + return cls.from_authorized_user_info(data, scopes, dpop_signer=dpop_signer) + diff --git a/packages/google-auth/tests/oauth2/test_dpop_credentials.py b/packages/google-auth/tests/oauth2/test_dpop_credentials.py new file mode 100644 index 000000000000..bf4ba5fc2973 --- /dev/null +++ b/packages/google-auth/tests/oauth2/test_dpop_credentials.py @@ -0,0 +1,835 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +import os +import tempfile +from unittest import mock + +import pytest # type: ignore +from cryptography.hazmat.primitives.asymmetric import ec + +from google.auth import dpop, exceptions, transport +from google.auth.dpop import SoftwareDPoPSigner +from google.oauth2 import _client, dpop_credentials, reauth +from google.oauth2 import credentials as _oauth2_credentials + +_TOKEN_URI = "https://oauth2.googleapis.com/token" +_REFRESH_TOKEN = "refresh_token" +_CLIENT_ID = "client_id" +_CLIENT_SECRET = "client_secret" +_ACCESS_TOKEN = "access-token-123" + +_SUCCESS_RESPONSE_DATA = { + "access_token": "new-access-token", + "refresh_token": "new-refresh-token", + "expires_in": 3600, +} + +_NONCE_HEADERS = {"DPoP-Nonce": "server-nonce-xyz"} + + +class TestDPoPCredentials(object): + TOKEN_URI = _TOKEN_URI + REFRESH_TOKEN = _REFRESH_TOKEN + CLIENT_ID = _CLIENT_ID + CLIENT_SECRET = _CLIENT_SECRET + + @classmethod + def make_dpop_signer(cls): + """Create a SoftwareDPoPSigner backed by a fresh EC P-256 key.""" + private_key = ec.generate_private_key(ec.SECP256R1()) + return SoftwareDPoPSigner(private_key) + + @classmethod + def make_credentials(cls, dpop_signer=None, dpop_nonce=None): + """Create a DPoPCredentials instance for testing.""" + return dpop_credentials.DPoPCredentials( + token=_ACCESS_TOKEN, + dpop_signer=dpop_signer, + dpop_nonce=dpop_nonce, + refresh_token=cls.REFRESH_TOKEN, + token_uri=cls.TOKEN_URI, + client_id=cls.CLIENT_ID, + client_secret=cls.CLIENT_SECRET, + ) + + # ------------------------------------------------------------------ + # Construction + # ------------------------------------------------------------------ + + def test_constructor(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer, dpop_nonce="nonce1") + + assert creds.token == _ACCESS_TOKEN + assert creds.dpop_signer is signer + assert creds.dpop_nonce == "nonce1" + assert creds.refresh_token == self.REFRESH_TOKEN + assert creds.token_uri == self.TOKEN_URI + assert creds.client_id == self.CLIENT_ID + assert creds.client_secret == self.CLIENT_SECRET + + def test_constructor_without_signer(self): + creds = self.make_credentials() + + assert creds.dpop_signer is None + assert creds.dpop_nonce is None + + def test_dpop_signer_property(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + + assert creds.dpop_signer is signer + + def test_dpop_nonce_property(self): + creds = self.make_credentials(dpop_nonce="test-nonce") + + assert creds.dpop_nonce == "test-nonce" + + # ------------------------------------------------------------------ + # _apply() — always uses Bearer (Google APIs require it) + # ------------------------------------------------------------------ + + def test_apply_with_signer_uses_bearer(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + headers = {} + + creds._apply(headers) + + assert headers["authorization"] == "Bearer {}".format(_ACCESS_TOKEN) + + def test_apply_without_signer(self): + creds = self.make_credentials() + headers = {} + + creds._apply(headers) + + assert headers["authorization"] == "Bearer {}".format(_ACCESS_TOKEN) + + # ------------------------------------------------------------------ + # _after_refresh() + # ------------------------------------------------------------------ + + def test_after_refresh_injects_dpop_proof(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer, dpop_nonce="after-nonce") + request = mock.create_autospec(transport.Request) + headers = {} + + creds._after_refresh(request, "GET", "https://api.example.com/data", headers) + + assert "DPoP" in headers + proof = headers["DPoP"] + # DPoP proof is a compact JWS: header.payload.signature + parts = proof.split(".") + assert len(parts) == 3 + + def test_after_refresh_no_signer(self): + creds = self.make_credentials() + request = mock.create_autospec(transport.Request) + headers = {} + + creds._after_refresh(request, "GET", "https://api.example.com/data", headers) + + assert "DPoP" not in headers + + def test_after_refresh_no_token(self): + signer = self.make_dpop_signer() + creds = dpop_credentials.DPoPCredentials( + token=None, + dpop_signer=signer, + refresh_token=self.REFRESH_TOKEN, + token_uri=self.TOKEN_URI, + client_id=self.CLIENT_ID, + client_secret=self.CLIENT_SECRET, + ) + request = mock.create_autospec(transport.Request) + headers = {} + + creds._after_refresh(request, "GET", "https://api.example.com/data", headers) + + assert "DPoP" not in headers + + @mock.patch.object( + _oauth2_credentials.Credentials, "_after_refresh", autospec=True + ) + def test_after_refresh_calls_super(self, mock_parent_after_refresh): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + headers = {} + + creds._after_refresh(request, "GET", "https://api.example.com/data", headers) + + # Verify super()._after_refresh() was called, which is + # required for Regional Access Boundary support. + mock_parent_after_refresh.assert_called_once_with( + creds, request, "GET", "https://api.example.com/data", headers + ) + + # ------------------------------------------------------------------ + # before_request() integration (refresh → _after_refresh → apply) + # ------------------------------------------------------------------ + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_before_request_full_cycle(self, mock_token_request): + """Verifies the complete before_request() flow produces both + Authorization: Bearer and DPoP headers on the outgoing request. + """ + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + # Expire the token so before_request triggers a refresh. + creds.token = None + creds.expiry = None + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + True, + _SUCCESS_RESPONSE_DATA, + None, + _NONCE_HEADERS, + ) + + headers = {} + creds.before_request( + request, "GET", "https://api.example.com/resource", headers + ) + + # Bearer token in Authorization header. + assert headers["authorization"] == "Bearer new-access-token" + + # DPoP proof in DPoP header. + assert "DPoP" in headers + parts = headers["DPoP"].split(".") + assert len(parts) == 3 + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_before_request_valid_token_skips_refresh(self, mock_token_request): + """When the token is valid, before_request() should skip refresh + but still apply the Bearer token. + """ + import datetime + + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + # Set a far-future expiry so token is valid. + creds.expiry = datetime.datetime.max + + headers = {} + creds.before_request( + mock.create_autospec(transport.Request), + "POST", + "https://api.example.com/data", + headers, + ) + + # No refresh should have occurred. + mock_token_request.assert_not_called() + + # Bearer header is applied. + assert headers["authorization"] == "Bearer {}".format(_ACCESS_TOKEN) + + # ------------------------------------------------------------------ + # _call_refresh_grant() + # ------------------------------------------------------------------ + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_happy_path(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + True, + _SUCCESS_RESPONSE_DATA, + None, + _NONCE_HEADERS, + ) + + result = creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + # Verify the endpoint was called once. + assert mock_token_request.call_count == 1 + + call_kwargs = mock_token_request.call_args + # Verify DPoP header was sent. + sent_headers = call_kwargs[1].get("headers") or call_kwargs[0][3] + assert "DPoP" in sent_headers + + # Verify return_response_headers=True was passed. + assert call_kwargs[1]["return_response_headers"] is True + + # Verify the result contains the expected tokens. + access_token, refresh_token, expiry, response_data, rapt = result + assert access_token == "new-access-token" + assert refresh_token == "new-refresh-token" + + # Verify nonce was captured from response headers. + assert creds.dpop_nonce == "server-nonce-xyz" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_nonce_challenge(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + # First call: use_dpop_nonce error with a nonce header. + # Second call: success. + mock_token_request.side_effect = [ + ( + False, + {"error": "use_dpop_nonce"}, + False, + {"DPoP-Nonce": "retry-nonce-abc"}, + ), + ( + True, + _SUCCESS_RESPONSE_DATA, + None, + {"DPoP-Nonce": "final-nonce-def"}, + ), + ] + + result = creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + # Verify two calls were made (initial + retry). + assert mock_token_request.call_count == 2 + + # Verify nonce was updated to the final value. + assert creds.dpop_nonce == "final-nonce-def" + + access_token = result[0] + assert access_token == "new-access-token" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_nonce_captured_on_success(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + True, + _SUCCESS_RESPONSE_DATA, + None, + {"DPoP-Nonce": "captured-nonce"}, + ) + + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + assert creds.dpop_nonce == "captured-nonce" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_error(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + False, + {"error": "invalid_grant", "error_description": "Token expired"}, + False, + {}, + ) + + with pytest.raises(exceptions.RefreshError): + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + def test_refresh_without_signer_raises(self): + creds = self.make_credentials() # no signer + request = mock.create_autospec(transport.Request) + + with pytest.raises(ValueError, match="requires a dpop_signer"): + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + @mock.patch("google.oauth2.reauth.get_rapt_token") + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_rapt_challenge_with_reauth_enabled( + self, mock_token_request, mock_get_rapt + ): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + creds._enable_reauth_refresh = True + request = mock.create_autospec(transport.Request) + mock_get_rapt.return_value = "rapt-token-123" + + # First call: RAPT challenge. + # Second call: success after RAPT is included. + mock_token_request.side_effect = [ + ( + False, + { + "error": reauth._REAUTH_NEEDED_ERROR, + "error_subtype": reauth._REAUTH_NEEDED_ERROR_RAPT_REQUIRED, + }, + False, + {}, + ), + ( + True, + _SUCCESS_RESPONSE_DATA, + None, + _NONCE_HEADERS, + ), + ] + + result = creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + # Verify reauth.get_rapt_token was called. + mock_get_rapt.assert_called_once_with( + request, + self.CLIENT_ID, + self.CLIENT_SECRET, + self.REFRESH_TOKEN, + self.TOKEN_URI, + scopes=None, + ) + + # Two calls to the endpoint: initial + retry with RAPT. + assert mock_token_request.call_count == 2 + + # Verify the second call includes the rapt token in the body. + second_call_body = mock_token_request.call_args_list[1][0][2] + assert second_call_body["rapt"] == "rapt-token-123" + + access_token = result[0] + assert access_token == "new-access-token" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_rapt_challenge_without_reauth_raises( + self, mock_token_request + ): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + creds._enable_reauth_refresh = False + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + False, + { + "error": reauth._REAUTH_NEEDED_ERROR, + "error_subtype": reauth._REAUTH_NEEDED_ERROR_INVALID_RAPT, + }, + False, + {}, + ) + + with pytest.raises(exceptions.RefreshError, match="Reauthentication"): + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + @mock.patch("google.oauth2.reauth.get_rapt_token") + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_rapt_challenge_invalid_rapt_subtype( + self, mock_token_request, mock_get_rapt + ): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + creds._enable_reauth_refresh = True + request = mock.create_autospec(transport.Request) + mock_get_rapt.return_value = "rapt-token-456" + + mock_token_request.side_effect = [ + ( + False, + { + "error": reauth._REAUTH_NEEDED_ERROR, + "error_subtype": reauth._REAUTH_NEEDED_ERROR_INVALID_RAPT, + }, + False, + {}, + ), + ( + True, + _SUCCESS_RESPONSE_DATA, + None, + {}, + ), + ] + + result = creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + ) + + mock_get_rapt.assert_called_once() + assert result[0] == "new-access-token" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_with_scopes_passed_to_body(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + True, + _SUCCESS_RESPONSE_DATA, + None, + {}, + ) + + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + scopes=["email", "profile"], + ) + + call_body = mock_token_request.call_args[0][2] + assert call_body["scope"] == "email profile" + + @mock.patch("google.oauth2._client._token_endpoint_request_no_throw") + def test_refresh_with_rapt_token_passed_to_body(self, mock_token_request): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + request = mock.create_autospec(transport.Request) + + mock_token_request.return_value = ( + True, + _SUCCESS_RESPONSE_DATA, + None, + {}, + ) + + creds._call_refresh_grant( + request, + self.TOKEN_URI, + self.REFRESH_TOKEN, + self.CLIENT_ID, + self.CLIENT_SECRET, + rapt_token="existing-rapt", + ) + + call_body = mock_token_request.call_args[0][2] + assert call_body["rapt"] == "existing-rapt" + + # ------------------------------------------------------------------ + # _make_copy() + # ------------------------------------------------------------------ + + def test_make_copy_preserves_dpop_fields(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer, dpop_nonce="copy-nonce") + + copy = creds._make_copy() + + assert copy.dpop_signer is signer + assert copy.dpop_nonce == "copy-nonce" + assert copy.refresh_token == self.REFRESH_TOKEN + assert copy.token_uri == self.TOKEN_URI + assert copy.client_id == self.CLIENT_ID + assert copy.client_secret == self.CLIENT_SECRET + assert isinstance(copy, dpop_credentials.DPoPCredentials) + + # ------------------------------------------------------------------ + # to_json() + # ------------------------------------------------------------------ + + def test_to_json_includes_dpop_metadata(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer, dpop_nonce="json-nonce") + + result = json.loads(creds.to_json()) + + assert result["dpop_nonce"] == "json-nonce" + assert "dpop_thumbprint" in result + assert result["dpop_thumbprint"] == signer.thumbprint + + def test_to_json_no_private_key_material(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer) + + raw = creds.to_json() + + # Private EC key components must never appear in the JSON. + assert '"d"' not in raw + assert '"p"' not in raw + assert '"q"' not in raw + assert "PRIVATE KEY" not in raw + + # ------------------------------------------------------------------ + # from_authorized_user_info() + # ------------------------------------------------------------------ + + def test_from_authorized_user_info_with_signer(self): + signer = self.make_dpop_signer() + info = { + "client_id": self.CLIENT_ID, + "client_secret": self.CLIENT_SECRET, + "refresh_token": self.REFRESH_TOKEN, + } + + creds = dpop_credentials.DPoPCredentials.from_authorized_user_info( + info, dpop_signer=signer + ) + + assert isinstance(creds, dpop_credentials.DPoPCredentials) + assert creds.dpop_signer is signer + assert creds.client_id == self.CLIENT_ID + assert creds.client_secret == self.CLIENT_SECRET + assert creds.refresh_token == self.REFRESH_TOKEN + + def test_from_authorized_user_info_nonce_from_info(self): + info = { + "client_id": self.CLIENT_ID, + "client_secret": self.CLIENT_SECRET, + "refresh_token": self.REFRESH_TOKEN, + "dpop_nonce": "saved-nonce", + } + + creds = dpop_credentials.DPoPCredentials.from_authorized_user_info(info) + + assert creds.dpop_nonce == "saved-nonce" + + def test_from_authorized_user_info_missing_fields_raises(self): + info = { + "client_id": self.CLIENT_ID, + # Missing client_secret and refresh_token. + } + + with pytest.raises(ValueError, match="missing"): + dpop_credentials.DPoPCredentials.from_authorized_user_info(info) + + def test_from_authorized_user_info_thumbprint_mismatch_raises(self): + signer = self.make_dpop_signer() + wrong_key = ec.generate_private_key(ec.SECP256R1()) + wrong_signer = SoftwareDPoPSigner(wrong_key) + info = { + "client_id": self.CLIENT_ID, + "client_secret": self.CLIENT_SECRET, + "refresh_token": self.REFRESH_TOKEN, + "dpop_thumbprint": signer.thumbprint, + } + + with pytest.raises(ValueError, match="does not match"): + dpop_credentials.DPoPCredentials.from_authorized_user_info( + info, dpop_signer=wrong_signer + ) + + def test_from_authorized_user_info_thumbprint_match_succeeds(self): + signer = self.make_dpop_signer() + info = { + "client_id": self.CLIENT_ID, + "client_secret": self.CLIENT_SECRET, + "refresh_token": self.REFRESH_TOKEN, + "dpop_thumbprint": signer.thumbprint, + } + + creds = dpop_credentials.DPoPCredentials.from_authorized_user_info( + info, dpop_signer=signer + ) + + assert creds.dpop_signer is signer + + def test_from_authorized_user_info_no_saved_thumbprint_skips_check(self): + signer = self.make_dpop_signer() + info = { + "client_id": self.CLIENT_ID, + "client_secret": self.CLIENT_SECRET, + "refresh_token": self.REFRESH_TOKEN, + } + + creds = dpop_credentials.DPoPCredentials.from_authorized_user_info( + info, dpop_signer=signer + ) + + assert creds.dpop_signer is signer + + # ------------------------------------------------------------------ + # from_authorized_user_file() + # ------------------------------------------------------------------ + + def test_from_authorized_user_file_roundtrip(self): + signer = self.make_dpop_signer() + creds = self.make_credentials(dpop_signer=signer, dpop_nonce="roundtrip-nonce") + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + f.write(creds.to_json()) + tmp_path = f.name + + try: + loaded = dpop_credentials.DPoPCredentials.from_authorized_user_file( + tmp_path, dpop_signer=signer + ) + + assert isinstance(loaded, dpop_credentials.DPoPCredentials) + assert loaded.dpop_signer is signer + assert loaded.dpop_nonce == "roundtrip-nonce" + assert loaded.client_id == self.CLIENT_ID + assert loaded.client_secret == self.CLIENT_SECRET + assert loaded.refresh_token == self.REFRESH_TOKEN + finally: + os.unlink(tmp_path) + + +class TestExtractDpopNonce(object): + """Tests for dpop.extract_dpop_nonce() helper.""" + + def test_extracts_nonce_exact_case(self): + headers = {"DPoP-Nonce": "abc"} + + result = dpop.extract_dpop_nonce(headers) + + assert result == "abc" + + def test_extracts_nonce_lowercase(self): + headers = {"dpop-nonce": "abc"} + + result = dpop.extract_dpop_nonce(headers) + + assert result == "abc" + + def test_returns_none_when_absent(self): + headers = {"other": "header"} + + result = dpop.extract_dpop_nonce(headers) + + assert result is None + + def test_returns_none_for_empty_headers(self): + result = dpop.extract_dpop_nonce({}) + + assert result is None + + def test_returns_none_for_none(self): + result = dpop.extract_dpop_nonce(None) + + assert result is None + + +class TestTokenEndpointRequestNoThrowWithResponseHeaders(object): + """Tests for the return_response_headers flag on + _client._token_endpoint_request_no_throw. + """ + + @classmethod + def _make_mock_response(cls, status=200, data=None, headers=None): + """Create a mock HTTP response object.""" + response = mock.Mock() + response.status = status + body = json.dumps(data or {}).encode("utf-8") + response.data = body + response.headers = headers or {} + return response + + def test_returns_3_tuple_by_default(self): + request = mock.create_autospec(transport.Request) + request.return_value = self._make_mock_response( + status=200, + data={"access_token": "tok"}, + headers={"X-Custom": "val"}, + ) + + result = _client._token_endpoint_request_no_throw( + request, + _TOKEN_URI, + {"grant_type": "authorization_code"}, + ) + + # Default: 3-tuple (success, data, retryable_error). + assert len(result) == 3 + assert result[0] is True + assert result[1]["access_token"] == "tok" + + def test_returns_4_tuple_when_requested(self): + response_headers = {"DPoP-Nonce": "server-nonce", "X-Other": "value"} + request = mock.create_autospec(transport.Request) + request.return_value = self._make_mock_response( + status=200, + data={"access_token": "tok"}, + headers=response_headers, + ) + + result = _client._token_endpoint_request_no_throw( + request, + _TOKEN_URI, + {"grant_type": "authorization_code"}, + return_response_headers=True, + ) + + # With return_response_headers=True: 4-tuple. + assert len(result) == 4 + assert result[0] is True + assert result[1]["access_token"] == "tok" + assert result[2] is None + assert result[3]["DPoP-Nonce"] == "server-nonce" + + def test_response_headers_on_error(self): + error_headers = {"DPoP-Nonce": "error-nonce"} + request = mock.create_autospec(transport.Request) + request.return_value = self._make_mock_response( + status=400, + data={"error": "invalid_grant"}, + headers=error_headers, + ) + + result = _client._token_endpoint_request_no_throw( + request, + _TOKEN_URI, + {"grant_type": "authorization_code"}, + return_response_headers=True, + can_retry=False, + ) + + assert len(result) == 4 + assert result[0] is False + assert result[1]["error"] == "invalid_grant" + assert result[3]["DPoP-Nonce"] == "error-nonce" diff --git a/packages/google-auth/tests/test_dpop.py b/packages/google-auth/tests/test_dpop.py new file mode 100644 index 000000000000..81e751378840 --- /dev/null +++ b/packages/google-auth/tests/test_dpop.py @@ -0,0 +1,734 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import base64 +import hashlib +import json +import unittest.mock as mock + +import pytest +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa + +from google.auth import _helpers, dpop, exceptions + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + + +def _b64url_decode(data): + """Base64url-decode without padding.""" + if isinstance(data, str): + data = data.encode("ascii") + padding_needed = 4 - len(data) % 4 + if padding_needed != 4: + data += b"=" * padding_needed + return base64.urlsafe_b64decode(data) + + +def _decode_jwt(token): + """Decode a compact JWT into (header_dict, payload_dict, sig_bytes).""" + parts = token.split(".") + header = json.loads(_b64url_decode(parts[0])) + payload = json.loads(_b64url_decode(parts[1])) + signature = _b64url_decode(parts[2]) + return header, payload, signature + + +# --------------------------------------------------------------------------- +# Test classes +# --------------------------------------------------------------------------- + + +class TestComputeJwkThumbprint(object): + """Tests for :func:`dpop.compute_jwk_thumbprint`.""" + + @classmethod + def make_ec256_signer(cls): + key = ec.generate_private_key(ec.SECP256R1()) + return dpop.SoftwareDPoPSigner(key) + + @classmethod + def make_ec384_signer(cls): + key = ec.generate_private_key(ec.SECP384R1()) + return dpop.SoftwareDPoPSigner(key) + + @classmethod + def make_rsa_signer(cls): + key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + return dpop.SoftwareDPoPSigner(key) + + def test_ec256_thumbprint_matches_manual_computation(self): + signer = self.make_ec256_signer() + jwk = signer.public_jwk + # Manually compute expected thumbprint. + canonical = { + "crv": jwk["crv"], + "kty": jwk["kty"], + "x": jwk["x"], + "y": jwk["y"], + } + canonical_json = json.dumps( + canonical, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + digest = hashlib.sha256(canonical_json).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + + result = dpop.compute_jwk_thumbprint(jwk) + assert result == expected + + def test_ec384_thumbprint(self): + signer = self.make_ec384_signer() + jwk = signer.public_jwk + result = dpop.compute_jwk_thumbprint(jwk) + # Thumbprint is a non-empty base64url string. + assert isinstance(result, str) + assert len(result) > 0 + # SHA-256 digest is 32 bytes → 43 base64url characters. + assert len(result) == 43 + + def test_rsa_thumbprint(self): + signer = self.make_rsa_signer() + jwk = signer.public_jwk + # Manually compute expected thumbprint using RSA required fields. + canonical = { + "e": jwk["e"], + "kty": jwk["kty"], + "n": jwk["n"], + } + canonical_json = json.dumps( + canonical, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + digest = hashlib.sha256(canonical_json).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + + result = dpop.compute_jwk_thumbprint(jwk) + assert result == expected + + def test_missing_required_field_raises(self): + # An EC JWK missing the "x" field. + jwk = {"kty": "EC", "crv": "P-256", "y": "some_y_value"} + with pytest.raises(exceptions.InvalidValue, match="missing required"): + dpop.compute_jwk_thumbprint(jwk) + + def test_unsupported_key_type_raises(self): + jwk = {"kty": "OKP", "crv": "Ed25519", "x": "abc"} + with pytest.raises(exceptions.InvalidValue, match="Unsupported key type"): + dpop.compute_jwk_thumbprint(jwk) + + def test_none_key_type_raises(self): + jwk = {"crv": "P-256", "x": "abc", "y": "def"} + with pytest.raises(exceptions.InvalidValue, match="Unsupported key type"): + dpop.compute_jwk_thumbprint(jwk) + + +class TestComputeAccessTokenHash(object): + """Tests for :func:`dpop.compute_access_token_hash`.""" + + def test_known_token_hash(self): + token = "test-access-token-12345" + digest = hashlib.sha256(token.encode("ascii")).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + result = dpop.compute_access_token_hash(token) + assert result == expected + + def test_bytes_input(self): + token = b"bytes-access-token" + digest = hashlib.sha256(token).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + result = dpop.compute_access_token_hash(token) + assert result == expected + + def test_hash_is_base64url_string(self): + result = dpop.compute_access_token_hash("some-token") + assert isinstance(result, str) + # Should be 43 characters (256 bits base64url without padding). + assert len(result) == 43 + + +class TestComputeAuthCodeJti(object): + """Tests for :func:`dpop.compute_auth_code_jti`.""" + + def test_known_code_hash(self): + code = "4/0AQSTgQF_auth_code_example" + digest = hashlib.sha256(code.encode("ascii")).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + result = dpop.compute_auth_code_jti(code) + assert result == expected + + def test_bytes_input(self): + code = b"auth-code-bytes" + digest = hashlib.sha256(code).digest() + expected = _helpers.unpadded_urlsafe_b64encode(digest).decode("ascii") + result = dpop.compute_auth_code_jti(code) + assert result == expected + + +class TestNormalizeDpopUrl(object): + """Tests for :func:`dpop.normalize_dpop_url`.""" + + def test_strips_query_string(self): + url = "https://example.com/path?foo=bar&baz=qux" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/path" + + def test_strips_fragment(self): + url = "https://example.com/path#section" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/path" + + def test_lowercases_scheme_and_host(self): + url = "HTTPS://Example.COM/Path" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/Path" + + def test_preserves_path(self): + url = "https://example.com/api/v1/Resource" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/api/v1/Resource" + + def test_adds_trailing_slash_when_path_empty(self): + url = "https://example.com" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/" + + def test_strips_query_and_fragment_together(self): + url = "https://example.com/path?q=1#frag" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com/path" + + def test_preserves_port(self): + url = "https://example.com:8443/resource" + result = dpop.normalize_dpop_url(url) + assert result == "https://example.com:8443/resource" + + +class TestDerRawEcdsaConversion(object): + """Tests for DER ↔ raw ECDSA signature conversion.""" + + @classmethod + def make_ec256_key(cls): + return ec.generate_private_key(ec.SECP256R1()) + + @classmethod + def make_ec384_key(cls): + return ec.generate_private_key(ec.SECP384R1()) + + def test_roundtrip_p256(self): + key = self.make_ec256_key() + data = b"test data to sign" + der_sig = key.sign(data, ec.ECDSA(hashes.SHA256())) + raw_sig = dpop.der_to_raw_ecdsa_signature(der_sig, 32) + der_sig_again = dpop.raw_to_der_ecdsa_signature(raw_sig) + raw_sig_again = dpop.der_to_raw_ecdsa_signature(der_sig_again, 32) + assert raw_sig == raw_sig_again + + def test_roundtrip_p384(self): + key = self.make_ec384_key() + data = b"test data to sign" + der_sig = key.sign(data, ec.ECDSA(hashes.SHA384())) + raw_sig = dpop.der_to_raw_ecdsa_signature(der_sig, 48) + der_sig_again = dpop.raw_to_der_ecdsa_signature(raw_sig) + raw_sig_again = dpop.der_to_raw_ecdsa_signature(der_sig_again, 48) + assert raw_sig == raw_sig_again + + def test_p256_raw_length(self): + key = self.make_ec256_key() + data = b"hello" + der_sig = key.sign(data, ec.ECDSA(hashes.SHA256())) + raw_sig = dpop.der_to_raw_ecdsa_signature(der_sig, 32) + assert len(raw_sig) == 64 + + def test_p384_raw_length(self): + key = self.make_ec384_key() + data = b"hello" + der_sig = key.sign(data, ec.ECDSA(hashes.SHA384())) + raw_sig = dpop.der_to_raw_ecdsa_signature(der_sig, 48) + assert len(raw_sig) == 96 + + def test_der_roundtrip_verifies(self): + """Verify that a DER→raw→DER signature still passes verification.""" + key = self.make_ec256_key() + data = b"verify me" + der_sig = key.sign(data, ec.ECDSA(hashes.SHA256())) + raw_sig = dpop.der_to_raw_ecdsa_signature(der_sig, 32) + der_sig_back = dpop.raw_to_der_ecdsa_signature(raw_sig) + # Should not raise. + key.public_key().verify(der_sig_back, data, ec.ECDSA(hashes.SHA256())) + + +class TestDPoPSigner(object): + """Tests for the abstract :class:`dpop.DPoPSigner`.""" + + def test_cannot_instantiate_directly(self): + with pytest.raises(TypeError): + dpop.DPoPSigner() + + def test_thumbprint_property_uses_compute_jwk_thumbprint(self): + """thumbprint delegates to compute_jwk_thumbprint.""" + ec_key = ec.generate_private_key(ec.SECP256R1()) + signer = dpop.SoftwareDPoPSigner(ec_key) + expected = dpop.compute_jwk_thumbprint(signer.public_jwk) + assert signer.thumbprint == expected + + def test_key_id_defaults_to_none(self): + ec_key = ec.generate_private_key(ec.SECP256R1()) + signer = dpop.SoftwareDPoPSigner(ec_key) + assert signer.key_id is None + + +class TestSoftwareDPoPSigner(object): + """Tests for :class:`dpop.SoftwareDPoPSigner`.""" + + @classmethod + def make_ec256_key(cls): + return ec.generate_private_key(ec.SECP256R1()) + + @classmethod + def make_ec384_key(cls): + return ec.generate_private_key(ec.SECP384R1()) + + @classmethod + def make_rsa_key(cls): + return rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + + # -- ES256 --------------------------------------------------------------- + + def test_es256_algorithm(self): + signer = dpop.SoftwareDPoPSigner(self.make_ec256_key()) + assert signer.algorithm == "ES256" + + def test_es256_public_jwk_has_required_fields(self): + signer = dpop.SoftwareDPoPSigner(self.make_ec256_key()) + jwk = signer.public_jwk + assert jwk["kty"] == "EC" + assert jwk["crv"] == "P-256" + assert "x" in jwk + assert "y" in jwk + # Must NOT contain private key material. + assert "d" not in jwk + + def test_es256_sign_produces_valid_output(self): + key = self.make_ec256_key() + signer = dpop.SoftwareDPoPSigner(key) + data = b"data to sign" + sig = signer.sign(data) + # Raw R||S for P-256 must be 64 bytes. + assert len(sig) == 64 + + def test_es256_sign_verify_roundtrip(self): + key = self.make_ec256_key() + signer = dpop.SoftwareDPoPSigner(key) + data = b"roundtrip test data" + raw_sig = signer.sign(data) + der_sig = dpop.raw_to_der_ecdsa_signature(raw_sig) + # Should not raise. + key.public_key().verify(der_sig, data, ec.ECDSA(hashes.SHA256())) + + # -- ES384 --------------------------------------------------------------- + + def test_es384_algorithm(self): + signer = dpop.SoftwareDPoPSigner(self.make_ec384_key()) + assert signer.algorithm == "ES384" + + def test_es384_public_jwk_curve(self): + signer = dpop.SoftwareDPoPSigner(self.make_ec384_key()) + assert signer.public_jwk["crv"] == "P-384" + + def test_es384_sign_length(self): + key = self.make_ec384_key() + signer = dpop.SoftwareDPoPSigner(key) + sig = signer.sign(b"data") + # Raw R||S for P-384 must be 96 bytes. + assert len(sig) == 96 + + # -- RS256 --------------------------------------------------------------- + + def test_rs256_algorithm(self): + signer = dpop.SoftwareDPoPSigner(self.make_rsa_key()) + assert signer.algorithm == "RS256" + + def test_rs256_public_jwk_has_required_fields(self): + signer = dpop.SoftwareDPoPSigner(self.make_rsa_key()) + jwk = signer.public_jwk + assert jwk["kty"] == "RSA" + assert "n" in jwk + assert "e" in jwk + # Must NOT contain private key material. + assert "d" not in jwk + assert "p" not in jwk + assert "q" not in jwk + + def test_rs256_sign_produces_output(self): + key = self.make_rsa_key() + signer = dpop.SoftwareDPoPSigner(key) + sig = signer.sign(b"data") + assert len(sig) > 0 + + def test_rs256_sign_verify_roundtrip(self): + key = self.make_rsa_key() + signer = dpop.SoftwareDPoPSigner(key) + data = b"roundtrip rsa test" + sig = signer.sign(data) + # Should not raise. + key.public_key().verify(sig, data, padding.PKCS1v15(), hashes.SHA256()) + + # -- Unsupported key types / curves -------------------------------------- + + def test_unsupported_curve_raises(self): + key = ec.generate_private_key(ec.SECP521R1()) + with pytest.raises(exceptions.InvalidValue, match="Unsupported"): + dpop.SoftwareDPoPSigner(key) + + def test_unsupported_key_type_raises(self): + from cryptography.hazmat.primitives.asymmetric import ed25519 + + key = ed25519.Ed25519PrivateKey.generate() + with pytest.raises(exceptions.InvalidValue, match="Unsupported"): + dpop.SoftwareDPoPSigner(key) + + # -- from_pem ------------------------------------------------------------ + + def test_from_pem_ec_key(self): + key = self.make_ec256_key() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + signer = dpop.SoftwareDPoPSigner.from_pem(pem) + assert signer.algorithm == "ES256" + assert signer.public_jwk["kty"] == "EC" + + def test_from_pem_rsa_key(self): + key = self.make_rsa_key() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + signer = dpop.SoftwareDPoPSigner.from_pem(pem) + assert signer.algorithm == "RS256" + assert signer.public_jwk["kty"] == "RSA" + + def test_from_pem_rejects_unsupported_key_type(self): + from cryptography.hazmat.primitives.asymmetric import ed25519 + + key = ed25519.Ed25519PrivateKey.generate() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + with pytest.raises(exceptions.InvalidValue, match="EC or RSA"): + dpop.SoftwareDPoPSigner.from_pem(pem) + + def test_from_pem_password_protected(self): + key = self.make_ec256_key() + password = b"test-password-123" + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.BestAvailableEncryption(password), + ) + signer = dpop.SoftwareDPoPSigner.from_pem(pem, password=password) + assert signer.algorithm == "ES256" + + def test_from_pem_string_input(self): + key = self.make_ec256_key() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + # Pass as str instead of bytes. + signer = dpop.SoftwareDPoPSigner.from_pem(pem.decode("utf-8")) + assert signer.algorithm == "ES256" + + def test_from_pem_with_key_id(self): + key = self.make_ec256_key() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + signer = dpop.SoftwareDPoPSigner.from_pem(pem, key_id="my-key-id") + assert signer.key_id == "my-key-id" + + # -- Properties ---------------------------------------------------------- + + def test_key_id_property(self): + key = self.make_ec256_key() + signer = dpop.SoftwareDPoPSigner(key, key_id="kid-123") + assert signer.key_id == "kid-123" + + def test_key_id_default_none(self): + key = self.make_ec256_key() + signer = dpop.SoftwareDPoPSigner(key) + assert signer.key_id is None + + def test_private_key_property(self): + key = self.make_ec256_key() + signer = dpop.SoftwareDPoPSigner(key) + assert signer.private_key is key + + +class TestCreateProof(object): + """Tests for :meth:`dpop.DPoPSigner.create_proof`.""" + + FIXED_TIME = 1700000000 + FIXED_UUID = "test-uuid-1234-5678" + + @classmethod + def make_ec256_signer(cls): + key = ec.generate_private_key(ec.SECP256R1()) + return dpop.SoftwareDPoPSigner(key) + + @classmethod + def make_rsa_signer(cls): + key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + return dpop.SoftwareDPoPSigner(key) + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_creates_valid_jwt_structure(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + parts = proof.split(".") + assert len(parts) == 3 + # Each part should be non-empty. + for part in parts: + assert len(part) > 0 + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_header_has_correct_fields(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + header, _, _ = _decode_jwt(proof) + assert header["typ"] == "dpop+jwt" + assert header["alg"] == "ES256" + assert "jwk" in header + assert header["jwk"]["kty"] == "EC" + assert header["jwk"]["crv"] == "P-256" + assert "x" in header["jwk"] + assert "y" in header["jwk"] + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_payload_has_required_fields(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + _, payload, _ = _decode_jwt(proof) + assert payload["jti"] == self.FIXED_UUID + assert payload["htm"] == "POST" + assert payload["htu"] == "https://example.com/token" + assert payload["iat"] == self.FIXED_TIME + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_htm_is_uppercased(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("post", "https://example.com/token") + _, payload, _ = _decode_jwt(proof) + assert payload["htm"] == "POST" + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_nonce_included_when_provided(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof( + "POST", + "https://example.com/token", + nonce="server-nonce-xyz", + ) + _, payload, _ = _decode_jwt(proof) + assert payload["nonce"] == "server-nonce-xyz" + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_nonce_absent_when_not_provided(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + _, payload, _ = _decode_jwt(proof) + assert "nonce" not in payload + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_access_token_hash_included(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + access_token = "my-access-token" + proof = signer.create_proof( + "GET", + "https://resource.example.com/api", + access_token=access_token, + ) + _, payload, _ = _decode_jwt(proof) + expected_ath = dpop.compute_access_token_hash(access_token) + assert payload["ath"] == expected_ath + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_ath_absent_when_no_access_token(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + _, payload, _ = _decode_jwt(proof) + assert "ath" not in payload + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_extra_claims_included(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof( + "POST", + "https://example.com/token", + extra_claims={"custom_key": "custom_value", "num": 42}, + ) + _, payload, _ = _decode_jwt(proof) + assert payload["custom_key"] == "custom_value" + assert payload["num"] == 42 + + @mock.patch("time.time", return_value=1700000000) + def test_custom_jti_is_used(self, mock_time): + signer = self.make_ec256_signer() + custom_jti = "my-custom-jti-value" + proof = signer.create_proof( + "POST", + "https://example.com/token", + jti=custom_jti, + ) + _, payload, _ = _decode_jwt(proof) + assert payload["jti"] == custom_jti + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_url_is_normalized_in_htu(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof( + "POST", + "HTTPS://Example.COM/token?code=abc#frag", + ) + _, payload, _ = _decode_jwt(proof) + assert payload["htu"] == "https://example.com/token" + + def test_signature_verifies_ec256(self): + """Verify the proof signature using the public key.""" + key = ec.generate_private_key(ec.SECP256R1()) + signer = dpop.SoftwareDPoPSigner(key) + proof = signer.create_proof("POST", "https://example.com/token") + parts = proof.split(".") + signing_input = "{}.{}".format(parts[0], parts[1]).encode("ascii") + raw_sig = _b64url_decode(parts[2]) + der_sig = dpop.raw_to_der_ecdsa_signature(raw_sig) + # Should not raise. + key.public_key().verify(der_sig, signing_input, ec.ECDSA(hashes.SHA256())) + + def test_signature_verifies_rsa(self): + """Verify the proof signature using the RSA public key.""" + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + signer = dpop.SoftwareDPoPSigner(key) + proof = signer.create_proof("POST", "https://example.com/token") + parts = proof.split(".") + signing_input = "{}.{}".format(parts[0], parts[1]).encode("ascii") + sig = _b64url_decode(parts[2]) + # Should not raise. + key.public_key().verify( + sig, + signing_input, + padding.PKCS1v15(), + hashes.SHA256(), + ) + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_rsa_proof_header(self, mock_uuid, mock_time): + signer = self.make_rsa_signer() + proof = signer.create_proof("POST", "https://example.com/token") + header, _, _ = _decode_jwt(proof) + assert header["typ"] == "dpop+jwt" + assert header["alg"] == "RS256" + assert header["jwk"]["kty"] == "RSA" + assert "n" in header["jwk"] + assert "e" in header["jwk"] + + @mock.patch("time.time", return_value=1700000000) + def test_auth_code_jti_flow(self, mock_time): + """Test code exchange flow with compute_auth_code_jti.""" + signer = self.make_ec256_signer() + code = "4/0AQSTgQF_some_code" + jti = dpop.compute_auth_code_jti(code) + proof = signer.create_proof( + "POST", + "https://oauth2.googleapis.com/token", + jti=jti, + ) + _, payload, _ = _decode_jwt(proof) + assert payload["jti"] == jti + # Verify jti matches the hash of the code. + expected_jti = dpop.compute_auth_code_jti(code) + assert payload["jti"] == expected_jti + + @mock.patch("time.time", return_value=1700000000) + @mock.patch( + "uuid.uuid4", + return_value=mock.Mock(__str__=lambda self: "test-uuid-1234-5678"), + ) + def test_jwk_in_header_matches_signer_public_jwk(self, mock_uuid, mock_time): + signer = self.make_ec256_signer() + proof = signer.create_proof("POST", "https://example.com/token") + header, _, _ = _decode_jwt(proof) + assert header["jwk"] == signer.public_jwk