diff --git a/google/auth/_agent_identity_utils.py b/google/auth/_agent_identity_utils.py new file mode 100644 index 000000000..b8b22d3c7 --- /dev/null +++ b/google/auth/_agent_identity_utils.py @@ -0,0 +1,188 @@ +# Copyright 2025 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. + +"""Helpers for Agent Identity credentials.""" + +import base64 +import hashlib +import os +import re +import time + +from google.auth import environment_vars +from google.auth.transport import _mtls_helper +from google.auth import exceptions + +# SPIFFE trust domain patterns for Agent Identities. +_AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS = [ + r"^agents\.global\.org-\d+\.system\.id\.goog$", + r"^agents\.global\.proj-\d+\.system\.id\.goog$", +] + + +def get_agent_identity_certificate_path(): + """Gets the certificate path from the certificate config file. + + The path to the certificate config file is read from the + GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function + implements a retry mechanism to handle cases where the environment + variable is set before the file is available on the filesystem. + + Returns: + str: The path to the leaf certificate file. + + Raises: + google.auth.exceptions.RefreshError: If the certificate config file + or the certificate file cannot be found after retries. + """ + import json + + cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) + if not cert_config_path: + return None + + # Poll for the config file and the certificate file to be available. + # Phase 1: Poll rapidly for 5 seconds (50 * 0.1s). + # Phase 2: Slow down polling for the next 25 seconds (50 * 0.5s). + for i in range(100): + if os.path.exists(cert_config_path): + with open(cert_config_path, "r") as f: + cert_config = json.load(f) + cert_path = ( + cert_config.get("cert_configs", {}) + .get("workload", {}) + .get("cert_path") + ) + if cert_path and os.path.exists(cert_path): + return cert_path + if i < 50: + time.sleep(0.1) + else: + time.sleep(0.5) + + raise exceptions.RefreshError( + "Certificate config or certificate file not found after multiple retries. " + f"If you are using Agent Engine, you can export " + f"{environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES} to false to " + "disable cert bound tokens to fall back to unbound tokens." + ) + + +def parse_certificate(cert_bytes): + """Parses a PEM-encoded certificate. + + Args: + cert_bytes (bytes): The PEM-encoded certificate bytes. + + Returns: + cryptography.x509.Certificate: The parsed certificate object. + """ + from cryptography import x509 + + return x509.load_pem_x509_certificate(cert_bytes) + + +def _is_agent_identity_certificate(cert): + """Checks if a certificate is an Agent Identity certificate. + + This is determined by checking the Subject Alternative Name (SAN) for a + SPIFFE ID with a trust domain matching Agent Identity patterns. + + Args: + cert (cryptography.x509.Certificate): The parsed certificate object. + + Returns: + bool: True if the certificate is an Agent Identity certificate, + False otherwise. + """ + from cryptography import x509 + from cryptography.x509.oid import ExtensionOID + + try: + ext = cert.extensions.get_extension_for_oid( + ExtensionOID.SUBJECT_ALTERNATIVE_NAME + ) + except x509.ExtensionNotFound: + return False + uris = ext.value.get_values_for_type(x509.UniformResourceIdentifier) + + for uri in uris: + if uri.startswith("spiffe://"): + spiffe_id = uri[len("spiffe://") :] + trust_domain = spiffe_id.split("/", 1)[0] + for pattern in _AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS: + if re.match(pattern, trust_domain): + return True + return False + + +def calculate_certificate_fingerprint(cert): + """Calculates the base64-encoded SHA256 hash of a DER-encoded certificate. + + Args: + cert (cryptography.x509.Certificate): The parsed certificate object. + + Returns: + str: The base64-encoded SHA256 fingerprint. + """ + from cryptography.hazmat.primitives import serialization + + der_cert = cert.public_bytes(serialization.Encoding.DER) + fingerprint = hashlib.sha256(der_cert).digest() + return base64.urlsafe_b64encode(fingerprint).rstrip(b"=").decode("utf-8") + + +def should_request_bound_token(cert): + """Determines if a bound token should be requested. + + This is based on the GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES + environment variable and whether the certificate is an agent identity cert. + + Args: + cert (cryptography.x509.Certificate): The parsed certificate object. + + Returns: + bool: True if a bound token should be requested, False otherwise. + """ + is_agent_cert = _is_agent_identity_certificate(cert) + is_opted_in = ( + os.environ.get( + environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + "true", + ).lower() + == "true" + ) + return is_agent_cert and is_opted_in + +def call_client_cert_callback(): + """Calls the client cert callback and returns the certificate and key.""" + _, cert_bytes, key_bytes, passphrase = ( + _mtls_helper.get_client_ssl_credentials(generate_encrypted_key=True) + ) + return cert_bytes, key_bytes + +def get_cached_cert_fingerprint(cached_cert): + """Returns the fingerprint of the cached certificate.""" + if cached_cert: + cert_obj = parse_certificate(cached_cert) + cached_cert_fingerprint = ( + calculate_certificate_fingerprint( + cert_obj + ) + ) + else: + raise ValueError("mTLS connection is not configured.") + return cached_cert_fingerprint + + diff --git a/google/auth/transport/requests.py b/google/auth/transport/requests.py index 2753912c6..26230f170 100644 --- a/google/auth/transport/requests.py +++ b/google/auth/transport/requests.py @@ -38,6 +38,8 @@ from google.auth import environment_vars from google.auth import exceptions from google.auth import transport +from google.auth.transport import _mtls_helper +from google.auth import _agent_identity_utils import google.auth.transport._mtls_helper from google.oauth2 import service_account @@ -468,6 +470,7 @@ def configure_mtls_channel(self, client_cert_callback=None): if self._is_mtls: mtls_adapter = _MutualTlsAdapter(cert, key) + self._cached_cert = cert self.mount("https://", mtls_adapter) except ( exceptions.ClientCertError, @@ -556,7 +559,41 @@ def request( response.status_code in self._refresh_status_codes and _credential_refresh_attempt < self._max_refresh_attempts ): - + if response.status_code == 401: + if self.is_mtls: + call_cert_callback_result = ( + _agent_identity_utils.call_client_cert_callback() + ) + cert_obj = _agent_identity_utils.parse_certificate( + call_cert_callback_result[0] + ) + current_cert_fingerprint = ( + _agent_identity_utils.calculate_certificate_fingerprint( + cert_obj + ) + ) + cached_fingerprint = ( + _agent_identity_utils.get_cached_cert_fingerprint( + self._cached_cert + ) + ) + if cached_fingerprint != current_cert_fingerprint: + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS " + "channel." + ) + self.configure_mtls_channel( + lambda: call_cert_callback_result + ) + except Exception as e: + _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) + raise e + else: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because the client" + " certificate has not changed." + ) _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", response.status_code, @@ -596,7 +633,13 @@ def is_mtls(self): """Indicates if the created SSL channel is mutual TLS.""" return self._is_mtls + @property + def cached_cert(self): + """Returns the cached client certificate.""" + return self._cached_cert + def close(self): if self._auth_request_session is not None: self._auth_request_session.close() super(AuthorizedSession, self).close() + diff --git a/google/auth/transport/urllib3.py b/google/auth/transport/urllib3.py index 03ed75aa2..04d21a672 100644 --- a/google/auth/transport/urllib3.py +++ b/google/auth/transport/urllib3.py @@ -54,6 +54,8 @@ from google.auth import environment_vars from google.auth import exceptions from google.auth import transport +from google.auth.transport import _mtls_helper +from google.auth import _agent_identity_utils from google.oauth2 import service_account if version.parse(urllib3.__version__) >= version.parse("2.0.0"): # pragma: NO COVER @@ -301,6 +303,7 @@ def __init__( # Request instance used by internal methods (for example, # credentials.refresh). self._request = Request(self.http) + self._is_mtls = False # https://google.aip.dev/auth/4111 # Attempt to use self-signed JWTs when a service account is used. @@ -339,7 +342,10 @@ def configure_mtls_channel(self, client_cert_callback=None): environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false" ) if use_client_cert != "true": + self._is_mtls = False return False + else: + self._is_mtls = True try: import OpenSSL @@ -354,6 +360,7 @@ def configure_mtls_channel(self, client_cert_callback=None): if found_cert_key: self.http = _make_mutual_tls_http(cert, key) + self._cached_cert = cert else: self.http = _make_default_http() except ( @@ -386,6 +393,11 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): if headers is None: headers = self.headers + use_mtls = False + if self._is_mtls == True: + if "mtls.googleapis.com" in url or "mtls.sandbox.googleapis.com" in url: + use_mtls = True + # Make a copy of the headers. They will be modified by the credentials # and we want to pass the original headers if we recurse. request_headers = headers.copy() @@ -407,28 +419,65 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): response.status in self._refresh_status_codes and _credential_refresh_attempt < self._max_refresh_attempts ): - - _LOGGER.info( + if response.status == 401: + if use_mtls: + call_cert_callback_result = ( + _agent_identity_utils.call_client_cert_callback() + ) + cert_obj = _agent_identity_utils.parse_certificate( + call_cert_callback_result[0] + ) + current_cert_fingerprint = ( + _agent_identity_utils.calculate_certificate_fingerprint( + cert_obj + ) + ) + if self._cached_cert: + cached_fingerprint = ( + _agent_identity_utils.get_cached_cert_fingerprint( + self._cached_cert + ) + ) + if cached_fingerprint != current_cert_fingerprint: + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS " + "channel." + ) + self.configure_mtls_channel( + lambda: call_cert_callback_result + ) + except Exception as e: + _LOGGER.error( + "Failed to reconfigure mTLS channel: %s", e + ) + raise e + else: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because the " + "client certificate has not changed." + ) + + _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", response.status, _credential_refresh_attempt + 1, self._max_refresh_attempts, ) - self.credentials.refresh(self._request) + self.credentials.refresh(self._request) - # Recurse. Pass in the original headers, not our modified set. - return self.urlopen( - method, - url, - body=body, - headers=headers, - _credential_refresh_attempt=_credential_refresh_attempt + 1, - **kwargs, - ) + # Recurse. Pass in the original headers, not our modified set. + return self.urlopen( + method, + url, + body=body, + headers=headers, + _credential_refresh_attempt=_credential_refresh_attempt + 1, + **kwargs, + ) return response - # Proxy methods for compliance with the urllib3.PoolManager interface def __enter__(self): @@ -443,6 +492,11 @@ def __del__(self): if hasattr(self, "http") and self.http is not None: self.http.clear() + @property + def is_mtls(self): + """Indicates if the created SSL channel is mutual TLS.""" + return self._is_mtls + @property def headers(self): """Proxy to ``self.http``.""" @@ -452,3 +506,5 @@ def headers(self): def headers(self, value): """Proxy to ``self.http``.""" self.http.headers = value + + diff --git a/tests/transport/test_requests.py b/tests/transport/test_requests.py index 0da3e36d9..c6e151c99 100644 --- a/tests/transport/test_requests.py +++ b/tests/transport/test_requests.py @@ -537,6 +537,119 @@ def test_close_w_passed_in_auth_request(self): authed_session.close() # no raise + def test_cert_rotation_when_cert_mismatch_and_mtls_enabled(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = make_response(status=http_client.OK) + # First request will 401, second request will succeed. + adapter = AdapterStub( + [make_response(status=http_client.UNAUTHORIZED), final_response] + ) + + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.TEST_URL, adapter) + + old_cert = b'-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n' + + # New certificate and key to simulate rotation. + new_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + + # Set _cached_cert to a callable that returns the old certificate. + authed_session._cached_cert = old_cert + authed_session._is_mtls = True + + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth.transport.requests._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + result = authed_session.request("GET", self.TEST_URL) + + # Asserts to verify the behavior. + assert mock_callback.called + assert credentials.refresh.called + assert credentials.refresh.call_count == 1 + assert result.status_code == final_response.status_code + + def test_no_cert_rotation_when_cert_match_and_mTLS_enabled(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = make_response(status=http_client.UNAUTHORIZED) + adapter = AdapterStub([make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.UNAUTHORIZED)]) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.TEST_URL, adapter) + authed_session._is_mtls = True + + old_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + + # New certificate and key to simulate rotation. + new_cert = old_cert + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + # Set _cached_cert to a callable that returns the old certificate. + authed_session._cached_cert = old_cert + + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth.transport.requests._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + result = authed_session.request("GET", self.TEST_URL) + + # Asserts to verify the behavior. + assert credentials.refresh.call_count == 2 + assert result.status_code == final_response.status_code + + def test_no_cert_match_check_when_mtls_disabled(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = make_response(status=http_client.UNAUTHORIZED) + adapter = AdapterStub([make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.UNAUTHORIZED), make_response(status=http_client.UNAUTHORIZED)]) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.TEST_URL, adapter) + authed_session._is_mtls = False + + new_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + + # New certificate and key to simulate rotation. + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth.transport.requests._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + result = authed_session.request("GET", self.TEST_URL) + + # Asserts to verify the behavior. + assert not mock_callback.called + assert result.status_code == final_response.status_code + + def test_no_cert_rotation_when_no_unauthorized_response(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = make_response(status=http_client.UPGRADE_REQUIRED) + + # Response is set to code other than 401(Unauthorized). + adapter = AdapterStub([make_response(status=http_client.UPGRADE_REQUIRED)]) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.TEST_URL, adapter) + + authed_session._is_mtls = True + + result = authed_session.request("GET", self.TEST_URL) + assert result.status_code == final_response.status_code + + # Asserts to verify the behavior. + assert not credentials.refresh.called + assert credentials.refresh.call_count == 0 class TestMutualTlsOffloadAdapter(object): @mock.patch.object(requests.adapters.HTTPAdapter, "init_poolmanager") diff --git a/tests/transport/test_urllib3.py b/tests/transport/test_urllib3.py index e83230032..2b83b1fac 100644 --- a/tests/transport/test_urllib3.py +++ b/tests/transport/test_urllib3.py @@ -320,3 +320,106 @@ def test_clear_pool_on_del(self): authed_http.http = None authed_http.__del__() # Expect it to not crash + + def test_cert_rotation_when_cert_mismatch_and_mtls_endpoint_used(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.OK) + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), final_response]) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + + old_cert = b'-----BEGIN CERTIFICATE-----\nMIIBdTCCARqgAwIBAgIJAOYVvu/axMxvMAoGCCqGSM49BAMCMCcxJTAjBgNVBAMM\nHEdvb2dsZSBFbmRwb2ludCBWZXJpZmljYXRpb24wHhcNMjUwNzMwMjMwNjA4WhcN\nMjYwNzMxMjMwNjA4WjAnMSUwIwYDVQQDDBxHb29nbGUgRW5kcG9pbnQgVmVyaWZp\nY2F0aW9uMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEbtr18gkEtwPow2oqyZsU\n4KLwFaLFlRlYv55UATS3QTDykDnIufC42TJCnqFRYhwicwpE2jnUV+l9g3Voias8\nraMvMC0wCQYDVR0TBAIwADALBgNVHQ8EBAMCB4AwEwYDVR0lBAwwCgYIKwYBBQUH\nAwIwCgYIKoZIzj0EAwIDSQAwRgIhAKcjW6dmF1YCksXPgDPlPu/nSnOjb3qCcivz\n/Jxq2zoeAiEA7/aNxcEoCGS3hwMIXoaaD/vPcZOOopKSyqXCvxRooKQ=\n-----END CERTIFICATE-----\n' + + # New certificate and key to simulate rotation. + new_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + # Set _cached_cert to a callable that returns the old certificate. + authed_http._cached_cert = old_cert + authed_http._is_mtls = True + # Mock call_client_cert_callback to return the new certificate. + with mock.patch.object( + google.auth._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + # mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + + # Asserts to verify the behavior. + assert result == final_response + assert credentials.refresh.called + assert credentials.refresh.call_count == 1 + assert mock_callback.called + + def test_no_cert_rotation_when_cert_match_and_mtls_endpoint_used(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UNAUTHORIZED) + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), ResponseStub(status=http_client.UNAUTHORIZED), ResponseStub(status=http_client.UNAUTHORIZED)]) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + old_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + + new_cert = old_cert + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + # Set _cached_cert to a callable that returns the same certificate. + authed_http._cached_cert = old_cert + authed_http._is_mtls = True + # Mock call_client_cert_callback to return the certificate. + with mock.patch.object( + google.auth._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + # mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + + # Asserts to verify the behavior. + assert credentials.refresh.call_count == 2 + assert result.status == final_response.status + + def test_no_cert_match_check_when_mtls_endpoint_not_used(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UNAUTHORIZED) + http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED), ResponseStub(status=http_client.UNAUTHORIZED), ResponseStub(status=http_client.UNAUTHORIZED)]) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = False + new_cert = b'-----BEGIN CERTIFICATE-----\nMIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV\nBAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV\nMRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\nCgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM\n7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer\nuQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp\ngyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4\n+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3\nZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O\ngN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh\nGaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD\nAQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr\nodJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk\n+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9\novNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql\nybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT\ncDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB\n-----END CERTIFICATE-----\n' + new_key = b'-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIHeMEkGCSqGSIb3DQEFDTA8MBsGCSqGSIb3DQEFDDAOBAj9XnJ2h78QVAICCAAw\nHQYJYIZIAWUDBAECBBBeiiOF2LnLzq/wjb/viwMwBIGQk28Zkfj2EIk42bgc7UzC\nSf98qssCVhsIYz0Xa3eSATg8Cpn83YieaBeyxdk/tXTnrOhxMV/vt7T98kWhaGbH\n5Z9CdGVLfes0UFvVJqrlk6vcf2sOnLCGbrn78HS+ayrGOCRSCd/7+dnEiB/7Um1B\nMk6BBJHsLEnZZSHyfrw8jvYgVmcSBy/WdY0pqldD/+4D\n-----END ENCRYPTED PRIVATE KEY-----\n' + + # Mock call_client_cert_callback to return the certificate. + with mock.patch.object( + google.auth._agent_identity_utils, + 'call_client_cert_callback', + return_value=(new_cert, new_key) + ) as mock_callback: + # non-mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.googleapis.com") + + # Asserts to verify the behavior. + assert not mock_callback.called + assert result.status == final_response.status + + def test_no_cert_rotation_when_no_unauthorized_response(self): + credentials = mock.Mock(wraps=CredentialsStub()) + final_response = ResponseStub(status=http_client.UPGRADE_REQUIRED) + + # Response is set to code other than 401(Unauthorized). + http = HttpStub([ResponseStub(status=http_client.UPGRADE_REQUIRED)]) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = True + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + # mTLS endpoint is used + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + assert result.status == final_response.status + assert not credentials.refresh.called + assert credentials.refresh.call_count == 0 +