diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py index 0fa6a92f57ff..ed4264293443 100644 --- a/packages/google-auth/google/auth/_agent_identity_utils.py +++ b/packages/google-auth/google/auth/_agent_identity_utils.py @@ -16,6 +16,7 @@ import base64 import hashlib +import json import os import re import stat @@ -73,13 +74,32 @@ def _is_certificate_file_ready(path): return False +def _is_in_well_known_dir(path): + """Checks if the given path is inside the well-known Agent Identity directory.""" + if not path: + return False + well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH) + try: + real_path = os.path.realpath(path) + real_well_known_dir = os.path.realpath(well_known_dir) + return ( + os.path.commonpath([real_well_known_dir, real_path]) == real_well_known_dir + ) + except ValueError: + return False + + def get_agent_identity_certificate_path(): - """Gets the agent certificate path from the certificate config file. + """Gets the agent certificate path from the certificate config file or GKE bundle. The path to the certificate config file is read from the GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function can optionally trigger polling to handle cases where the environment variable is set before the files are available on the filesystem. + When GOOGLE_API_CERTIFICATE_CONFIG and + CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH are unset and no + implicit certificate_config.json is present, it falls back to checking + the GKE workload credential bundle path without polling. Returns: Optional[str]: The path to the agent's certificate file, or None if unavailable. @@ -88,9 +108,18 @@ def get_agent_identity_certificate_path(): google.auth.exceptions.RefreshError: If the certificate config file or the certificate file cannot be found after retries. """ + from google.auth.transport import _mtls_helper + cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) if not cert_config_path: + if ( + not _mtls_helper._has_explicit_cert_config_env() + and _mtls_helper._should_use_gke_credential_bundle( + _mtls_helper._get_cert_config_path() + ) + ): + return _mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH return None # We trigger polling only if the config path points to the well-known directory. @@ -98,16 +127,7 @@ def get_agent_identity_certificate_path(): # config file and the certificate file may experience a brief startup latency. # For all other paths, we return early to avoid introducing unnecessary startup # delays. - well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH) - try: - abs_cert_path = os.path.abspath(cert_config_path) - abs_well_known_dir = os.path.abspath(well_known_dir) - should_poll = ( - os.path.commonpath([abs_well_known_dir, abs_cert_path]) - == abs_well_known_dir - ) - except ValueError: - should_poll = False + should_poll = _is_in_well_known_dir(cert_config_path) return _get_cert_path_with_optional_polling(cert_config_path, should_poll) @@ -141,9 +161,9 @@ def _get_cert_path_with_optional_polling(cert_config_path, should_poll): if _is_certificate_file_ready(cert_path): return cert_path - # The config was parsed, but the cert file is not ready yet - if not should_poll: - # If polling is disabled, return early. + # The config was parsed, but the cert file is not ready yet. + # Only poll if both the config path and cert path are in the well-known directory. + if not (should_poll and _is_in_well_known_dir(cert_path)): return None if not has_logged_cert_warning: @@ -182,7 +202,7 @@ def _get_cert_path_with_optional_polling(cert_config_path, should_poll): raise exceptions.RefreshError( "Certificate config or certificate file not found after multiple retries. " f"Token binding protection is failing. You can turn off this protection by setting " - f"{environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES} to false " + f"{environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN} to false " "to fall back to unbound tokens." ) @@ -203,8 +223,6 @@ def _parse_cert_path_from_config(cert_config_path): KeyError: If the certificate config file does not contain the expected structure. """ - import json - with open(cert_config_path, "r", encoding="utf-8") as f: cert_config = json.load(f) @@ -221,64 +239,105 @@ def _parse_cert_path_from_config(cert_config_path): return workload_config["cert_path"] -def get_and_parse_agent_identity_certificate(): +def _is_bound_token_opted_out(): + """Returns True only if bound tokens are explicitly disabled via env vars.""" + val = os.environ.get( + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, "" + ).strip() + if not val: + # Fall back to the deprecated env var for backward compatibility + val = os.environ.get( + environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + "true", + ).strip() + return val.lower() == "false" + + +def get_agent_identity_certificate_and_bytes(): """Gets and parses the agent identity certificate if not opted out. Checks if the user has opted out of certificate-bound tokens. If not, it gets the certificate path, reads the file, and parses it. Returns: - The parsed certificate object if found and not opted out, otherwise None. + Tuple[Optional[cryptography.x509.Certificate], Optional[bytes]]: A tuple + of (parsed certificate object, certificate bytes) if found and not + opted out, otherwise (None, None). """ + from google.auth.transport import _mtls_helper + # If the user has opted out of cert bound tokens, there is no need to # look up the certificate. - is_opted_out = ( - os.environ.get( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - "true", - ).lower() - == "false" - ) - if is_opted_out: - return None + if _is_bound_token_opted_out(): + return None, None # Respect explicit opt-out of mTLS / client certs - from google.auth.transport import _mtls_helper - env_override = _mtls_helper._check_use_client_cert_env() if env_override is False: - return None + return None, None cert_path = get_agent_identity_certificate_path() if not cert_path: - return None + return None, None try: with open(cert_path, "rb") as cert_file: - cert_bytes = cert_file.read() - except PermissionError as e: + raw_bytes = cert_file.read() + except OSError as e: warnings.warn( f"Failed to read agent identity certificate file at {cert_path}: {e}. " "Token binding protection cannot be enabled. Falling back to unbound tokens." ) - return None + return None, None - return parse_certificate(cert_bytes) + cert_blocks = _mtls_helper._CERT_REGEX.findall(raw_bytes) + if not cert_blocks: + warnings.warn( + f"No PEM certificate blocks found in {cert_path}. " + "Token binding protection cannot be enabled. Falling back to unbound tokens." + ) + return None, None + + cert_bytes = b"\n".join(block.strip() for block in cert_blocks) + b"\n" + try: + return parse_certificate(raw_bytes), cert_bytes + except (ValueError, ImportError) as e: + warnings.warn( + f"Failed to parse agent identity certificate at {cert_path}: {e}. " + "Token binding protection cannot be enabled. Falling back to unbound tokens." + ) + return None, None def parse_certificate(cert_bytes): - """Parses a PEM-encoded certificate. + """Parses a PEM-encoded certificate or certificate chain and returns the leaf certificate. Args: cert_bytes (bytes): The PEM-encoded certificate bytes. Returns: - cryptography.x509.Certificate: The parsed certificate object. + cryptography.x509.Certificate: The leaf (first) parsed certificate object. + + Raises: + ValueError: If no certificates are found or any certificate in the chain + is malformed. + ImportError: If the cryptography library is not installed. """ + from google.auth.transport import _mtls_helper + + if not cert_bytes: + raise ValueError("Certificate bytes cannot be empty or None.") + try: from cryptography import x509 - return x509.load_pem_x509_certificate(cert_bytes) + cert_blocks = _mtls_helper._CERT_REGEX.findall(cert_bytes) + if not cert_blocks: + return x509.load_pem_x509_certificate(cert_bytes) + if _mtls_helper._has_unmatched_pem_markers(cert_bytes, cert_blocks): + raise ValueError("Malformed or truncated PEM certificate chain.") + certs = [x509.load_pem_x509_certificate(block) for block in cert_blocks] + return certs[0] except ImportError as e: raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e @@ -350,8 +409,9 @@ def calculate_certificate_fingerprint(cert): 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. + This is based on the GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN env var + (falls back to the deprecated GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES + if unset or empty) and whether the certificate is an agent identity cert. Args: cert (cryptography.x509.Certificate): The parsed certificate object. @@ -359,25 +419,17 @@ def should_request_bound_token(cert): 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" - ) - if not (is_agent_cert and is_opted_in): + from google.auth.transport import _mtls_helper + + if _is_bound_token_opted_out(): return False # Respect explicit opt-out of mTLS / client certs - from google.auth.transport import _mtls_helper - env_override = _mtls_helper._check_use_client_cert_env() if env_override is False: return False - return True + return _is_agent_identity_certificate(cert) def get_cached_cert_fingerprint(cached_cert): diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py index 14e950427b5f..980c86f5ea19 100644 --- a/packages/google-auth/google/auth/aio/transport/sessions.py +++ b/packages/google-auth/google/auth/aio/transport/sessions.py @@ -19,7 +19,6 @@ import inspect import logging import time -import urllib.parse import warnings from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Mapping, Optional, Union @@ -44,11 +43,6 @@ ClientTimeout = None _LOGGER = logging.getLogger(__name__) -_MTLS_URL_PREFIXES = [ - "mtls.googleapis.com", - "mtls.sandbox.googleapis.com", - "p.googleapis.com", -] # Tracks the internal aiohttp installation and usage try: @@ -482,13 +476,9 @@ async def request( async def _recover_auth_state(): is_mtls_endpoint = False if self._is_mtls: - hostname = urllib.parse.urlsplit(url).hostname - if hostname: - is_mtls_endpoint = any( - hostname == prefix - or hostname.endswith("." + prefix) - for prefix in _MTLS_URL_PREFIXES - ) + is_mtls_endpoint = ( + google.auth.transport._mtls_helper.is_mtls_endpoint(url) + ) # Snapshot the stale certificate state BEFORE acquiring the lock. # This represents the cert that caused the 401 rejection. if is_mtls_endpoint: diff --git a/packages/google-auth/google/auth/compute_engine/_metadata.py b/packages/google-auth/google/auth/compute_engine/_metadata.py index 58c509dfa768..7404cf2896e1 100644 --- a/packages/google-auth/google/auth/compute_engine/_metadata.py +++ b/packages/google-auth/google/auth/compute_engine/_metadata.py @@ -27,7 +27,14 @@ import requests -from google.auth import _helpers, environment_vars, exceptions, metrics, transport +from google.auth import ( + _agent_identity_utils, + _helpers, + environment_vars, + exceptions, + metrics, + transport, +) from google.auth._exponential_backoff import ExponentialBackoff from google.auth.compute_engine import _mtls @@ -250,6 +257,8 @@ def get( headers=None, return_none_for_not_found_error=False, timeout=_METADATA_DEFAULT_TIMEOUT, + method="GET", + body=None, ): """Fetch a resource from the metadata server. @@ -271,6 +280,8 @@ def get( return_none_for_not_found_error (Optional[bool]): If True, returns None for 404 error instead of throwing an exception. timeout (int): How long to wait, in seconds for the metadata server to respond. + method (str): The HTTP method to use for the request. Defaults to "GET". + body (Optional[bytes]): The HTTP request body payload to send. Defaults to None. Returns: Union[Mapping, str]: If the metadata server returns JSON, a mapping of @@ -283,8 +294,12 @@ def get( google.auth.exceptions.MutualTLSChannelError: if using mtls and the environment configuration is invalid for mTLS (for example, the metadata host has been overridden in strict mTLS mode). + ValueError: if a request body is specified with the GET method. """ + if body is not None and method.upper() == "GET": + raise ValueError("Request body cannot be specified with GET method.") + use_mtls = _mtls.should_use_mds_mtls() # Prepare the request object for mTLS if needed. # This will create a new request object with the mTLS session. @@ -314,9 +329,15 @@ def get( last_exception = None for attempt in backoff: try: - response = request( - url=url, method="GET", headers=headers_to_use, timeout=timeout - ) + kwargs = { + "url": url, + "method": method, + "headers": headers_to_use, + "timeout": timeout, + } + if body is not None: + kwargs["body"] = body + response = request(**kwargs) if response.status in transport.DEFAULT_RETRYABLE_STATUS_CODES: _LOGGER.warning( "Compute Engine Metadata server unavailable on " @@ -460,6 +481,40 @@ def get_service_account_info(request, service_account="default"): return get(request, path, params={"recursive": "true"}) +def _build_token_request_options(metrics_header_value, bind_token=True): + """Returns (method, body, headers) for a metadata server token request. + + Defaults to a standard GET request with the x-goog-api-client metrics header. + If ``bind_token`` is ``True``, upgrades to a POST request with a JSON + certificate_chain body and Content-Type header if an Agent Identity + certificate is present and bound tokens are enabled. + + Args: + metrics_header_value (str): Value for the x-goog-api-client header. + bind_token (bool): Whether to bind the token to the Agent Identity + certificate. Defaults to ``True``. If ``False``, returns a ``GET`` + request without reading the certificate. If ``True``, requests a + bound token via ``POST`` when a valid Agent Identity certificate is + present and token binding is enabled. + + Returns: + Tuple[str, Optional[bytes], Mapping[str, str]]: A tuple of + (HTTP method, request body bytes, request headers). + """ + headers = {metrics.API_CLIENT_HEADER: metrics_header_value} + if not bind_token: + return "GET", None, headers + + cert, cert_bytes = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + if cert and _agent_identity_utils.should_request_bound_token(cert): + headers["Content-Type"] = "application/json" + body = json.dumps({"certificate_chain": cert_bytes.decode("utf-8")}).encode( + "utf-8" + ) + return "POST", body, headers + return "GET", None, headers + + def get_service_account_token(request, service_account="default", scopes=None): """Get the OAuth 2.0 access token for a service account. @@ -478,26 +533,20 @@ def get_service_account_token(request, service_account="default", scopes=None): google.auth.exceptions.TransportError: if an error occurred while retrieving metadata. """ - from google.auth import _agent_identity_utils - params = {} if scopes: if not isinstance(scopes, str): scopes = ",".join(scopes) params["scopes"] = scopes - cert = _agent_identity_utils.get_and_parse_agent_identity_certificate() - if cert: - if _agent_identity_utils.should_request_bound_token(cert): - fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(cert) - params["bindCertificateFingerprint"] = fingerprint - - metrics_header = { - metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds() - } + method, body, headers = _build_token_request_options( + metrics.token_request_access_token_mds() + ) path = "instance/service-accounts/{0}/token".format(service_account) - token_json = get(request, path, params=params, headers=metrics_header) + token_json = get( + request, path, params=params, headers=headers, method=method, body=body + ) token_expiry = _helpers.utcnow() + datetime.timedelta( seconds=token_json["expires_in"] ) diff --git a/packages/google-auth/google/auth/compute_engine/credentials.py b/packages/google-auth/google/auth/compute_engine/credentials.py index d0d24b41f4e4..4a71b36f9308 100644 --- a/packages/google-auth/google/auth/compute_engine/credentials.py +++ b/packages/google-auth/google/auth/compute_engine/credentials.py @@ -21,12 +21,14 @@ import datetime import logging +import os from typing import TYPE_CHECKING, Optional from google.auth import ( _helpers, _regional_access_boundary_utils, credentials, + environment_vars, exceptions, iam, jwt, @@ -340,6 +342,7 @@ def __init__( signer=None, use_metadata_identity_endpoint=False, quota_project_id=None, + bind_id_token=None, ): """ Args: @@ -364,18 +367,34 @@ def __init__( otherwise ValueError will be raised. quota_project_id (Optional[str]): The project ID used for quota and billing. + bind_id_token (Optional[bool]): Controls whether to request a + certificate-bound ID token. Can only be set when + ``use_metadata_identity_endpoint`` is ``True``. + If ``True``, requests a bound token whenever a valid Agent + Identity certificate is available and token binding is not + disabled via ``GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN``, falling + back to an unbound token otherwise (or raising + :class:`~google.auth.exceptions.RefreshError` if a configured + certificate is not found after retries). If ``False``, always + requests an unbound token. If ``None`` (default), token binding + is determined automatically by the library. Set ``True`` or + ``False`` explicitly if your application requires a specific + behavior. Raises: ValueError: If ``use_metadata_identity_endpoint`` is set to True, and one of ``token_uri``, ``additional_claims``, ``service_account_email``, - ``signer`` arguments is set. + ``signer`` arguments is set, or if + ``use_metadata_identity_endpoint`` is False and + ``bind_id_token`` is not None. """ super(IDTokenCredentials, self).__init__() self._quota_project_id = quota_project_id self._use_metadata_identity_endpoint = use_metadata_identity_endpoint self._target_audience = target_audience + self._bind_id_token = bind_id_token if use_metadata_identity_endpoint: if token_uri or additional_claims or service_account_email or signer: @@ -387,8 +406,13 @@ def __init__( self._token_uri = None self._additional_claims = None self._signer = None + elif bind_id_token is not None: + raise ValueError( + "If use_metadata_identity_endpoint is False, bind_id_token " + "must not be set" + ) - if service_account_email is None: + if service_account_email is None and request is not None: sa_info = _metadata.get_service_account_info(request) self._service_account_email = sa_info["email"] else: @@ -409,79 +433,60 @@ def __init__( else: self._additional_claims = {} - def with_target_audience(self, target_audience): - """Create a copy of these credentials with the specified target - audience. - Args: - target_audience (str): The intended audience for these credentials, - used when requesting the ID Token. - Returns: - google.auth.service_account.IDTokenCredentials: A new credentials - instance. - """ + def _make_copy(self): # since the signer is already instantiated, # the request is not needed if self._use_metadata_identity_endpoint: - return self.__class__( + cred = self.__class__( None, - target_audience=target_audience, + target_audience=self._target_audience, use_metadata_identity_endpoint=True, quota_project_id=self._quota_project_id, ) + cred._service_account_email = self._service_account_email + cred._bind_id_token = getattr(self, "_bind_id_token", None) + return cred else: return self.__class__( None, service_account_email=self._service_account_email, token_uri=self._token_uri, - target_audience=target_audience, + target_audience=self._target_audience, additional_claims=self._additional_claims.copy(), signer=self.signer, use_metadata_identity_endpoint=False, quota_project_id=self._quota_project_id, ) + def with_target_audience(self, target_audience): + """Create a copy of these credentials with the specified target + audience. + Args: + target_audience (str): The intended audience for these credentials, + used when requesting the ID Token. + Returns: + google.auth.service_account.IDTokenCredentials: A new credentials + instance. + """ + cred = self._make_copy() + cred._target_audience = target_audience + return cred + @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): - # since the signer is already instantiated, - # the request is not needed - if self._use_metadata_identity_endpoint: - return self.__class__( - None, - target_audience=self._target_audience, - use_metadata_identity_endpoint=True, - quota_project_id=quota_project_id, - ) - else: - return self.__class__( - None, - service_account_email=self._service_account_email, - token_uri=self._token_uri, - target_audience=self._target_audience, - additional_claims=self._additional_claims.copy(), - signer=self.signer, - use_metadata_identity_endpoint=False, - quota_project_id=quota_project_id, - ) + cred = self._make_copy() + cred._quota_project_id = quota_project_id + return cred @_helpers.copy_docstring(credentials.CredentialsWithTokenUri) def with_token_uri(self, token_uri): - # since the signer is already instantiated, - # the request is not needed if self._use_metadata_identity_endpoint: raise ValueError( "If use_metadata_identity_endpoint is set, token_uri must not be set" ) - else: - return self.__class__( - None, - service_account_email=self._service_account_email, - token_uri=token_uri, - target_audience=self._target_audience, - additional_claims=self._additional_claims.copy(), - signer=self.signer, - use_metadata_identity_endpoint=False, - quota_project_id=self.quota_project_id, - ) + cred = self._make_copy() + cred._token_uri = token_uri + return cred def _make_authorization_grant_assertion(self): """Create the OAuth 2.0 assertion. @@ -530,11 +535,25 @@ def _call_metadata_identity_endpoint(self, request): try: path = "instance/service-accounts/default/identity" params = {"audience": self._target_audience, "format": "full"} - metrics_header = { - metrics.API_CLIENT_HEADER: metrics.token_request_id_token_mds() - } + bind_token = getattr(self, "_bind_id_token", None) + if bind_token is None: + # Temporary gate: keep default ID tokens unbound when + # GOOGLE_API_CERTIFICATE_CONFIG is set (Cloud Run). + bind_token = not os.environ.get( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG + ) + method, body, headers = _metadata._build_token_request_options( + metrics.token_request_id_token_mds(), + bind_token=bind_token, + ) + id_token = _metadata.get( - request, path, params=params, headers=metrics_header + request, + path, + params=params, + headers=headers, + method=method, + body=body, ) except exceptions.TransportError as caught_exc: new_exc = exceptions.RefreshError(caught_exc) diff --git a/packages/google-auth/google/auth/environment_vars.py b/packages/google-auth/google/auth/environment_vars.py index 54550945f994..e8f3fb6e0670 100644 --- a/packages/google-auth/google/auth/environment_vars.py +++ b/packages/google-auth/google/auth/environment_vars.py @@ -128,10 +128,22 @@ """Environment variable defining the location of Google API certificate config file. This variable is the fallback of GOOGLE_API_CERTIFICATE_CONFIG.""" +GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN = "GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN" +"""Environment variable controlling whether to enable runtime bound tokens. + +Defaults to enabled; only a case-insensitive ``"false"`` disables it. When set +to a non-empty value, this variable takes precedence over +:data:`GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES`. +""" + GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES = ( "GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES" ) -"""Environment variable to prevent agent token sharing for GCP services.""" +"""Environment variable to prevent agent token sharing for GCP services. + +.. deprecated:: v2.59.0 + Use :data:`GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN` instead. +""" GOOGLE_API_USE_MTLS_ENDPOINT = "GOOGLE_API_USE_MTLS_ENDPOINT" """Environment variable controlling whether to use mTLS endpoint or not.""" diff --git a/packages/google-auth/google/auth/identity_pool.py b/packages/google-auth/google/auth/identity_pool.py index 41b312e4f841..6607bb3aecd1 100644 --- a/packages/google-auth/google/auth/identity_pool.py +++ b/packages/google-auth/google/auth/identity_pool.py @@ -404,9 +404,10 @@ def _get_mtls_cert_and_key_paths(self): 'The credential is not configured to use mtls requests. The credential should include a "certificate" section in the credential source.' ) else: - return _mtls_helper._get_workload_cert_and_key_paths( + cert_path, key_path, _ = _mtls_helper._get_workload_cert_and_key_paths( self._certificate_config_location ) + return cert_path, key_path def _get_cert_bytes(self): cert_path, _ = self._get_mtls_cert_and_key_paths() @@ -572,12 +573,12 @@ def refresh(self, request): if self._credential_source_certificate is not None: try: cert_bytes = self._get_cert_bytes() - except (exceptions.ClientCertError, OSError) as e: + cert = _agent_identity_utils.parse_certificate(cert_bytes) + except (exceptions.ClientCertError, OSError, ValueError) as e: raise exceptions.RefreshError( - "Failed to retrieve certificate bytes for external" + "Failed to retrieve or parse certificate for external" " account credentials" ) from e - cert = _agent_identity_utils.parse_certificate(cert_bytes) if _agent_identity_utils.should_request_bound_token(cert): cert_fingerprint = ( _agent_identity_utils.calculate_certificate_fingerprint(cert) diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 469932be058c..65d54e3b762f 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -32,9 +32,10 @@ # Default gcloud config path, to be used with path.expanduser for cross-platform compatibility. CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json" +_GKE_CREDENTIAL_BUNDLE_PATH = "/var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem" _CERT_PROVIDER_COMMAND = "cert_provider_command" _CERT_REGEX = re.compile( - b"-----BEGIN CERTIFICATE-----.+-----END CERTIFICATE-----\r?\n?", re.DOTALL + b"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----\r?\n?", re.DOTALL ) # support various format of key files, e.g. @@ -43,7 +44,7 @@ # "-----BEGIN RSA PRIVATE KEY-----..." # "-----BEGIN ENCRYPTED PRIVATE KEY-----" _KEY_REGEX = re.compile( - b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+-----END [A-Z ]*PRIVATE KEY-----\r?\n?", + b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+?-----END [A-Z ]*PRIVATE KEY-----\r?\n?", re.DOTALL, ) @@ -367,12 +368,37 @@ def _load_json_file(path): return json_data +def _has_explicit_cert_config_env(): + """Returns True if an explicit certificate config environment variable is set.""" + return bool( + environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) + or environ.get( + environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH + ) + ) + + +def _should_use_gke_credential_bundle(config_file_path=None): + """Returns True if no certificate config is present and the GKE credential bundle file is present.""" + if config_file_path is not None or _has_explicit_cert_config_env(): + return False + try: + return _agent_identity_utils._is_certificate_file_ready( + _GKE_CREDENTIAL_BUNDLE_PATH + ) + except PermissionError: + return False + + def _get_workload_cert_and_key( certificate_config_path=None, include_context_aware=True ): """Read the workload identity cert and key files specified in the certificate config provided. If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG" - first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json". + first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json", + and finally fall back to the GKE workload credential bundle at + "/var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem" + when no explicit or implicit certificate_config.json is present. Args: certificate_config_path (string): The certificate config path. If no path is provided, @@ -389,11 +415,15 @@ def _get_workload_cert_and_key( the certificate or key information. """ - cert_path, key_path = _get_workload_cert_and_key_paths( + cert_path, key_path, config_file_path = _get_workload_cert_and_key_paths( certificate_config_path, include_context_aware ) if cert_path is None and key_path is None: + if certificate_config_path is None and _should_use_gke_credential_bundle( + config_file_path + ): + return _read_credential_bundle_file(_GKE_CREDENTIAL_BUNDLE_PATH) return None, None return _read_cert_and_key_files(cert_path, key_path) @@ -453,10 +483,10 @@ def _get_cert_config_path(certificate_config_path=None, include_context_aware=Tr return certificate_config_path -def _get_workload_cert_and_key_paths(config_path, include_context_aware=True): +def _get_workload_cert_and_key_paths(config_path=None, include_context_aware=True): absolute_path = _get_cert_config_path(config_path, include_context_aware) if absolute_path is None: - return None, None + return None, None, None data = _load_json_file(absolute_path) @@ -503,13 +533,15 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True): pass if not isinstance(cert_configs, dict) or "workload" not in cert_configs: - return None, None + return None, None, absolute_path workload = cert_configs["workload"] if ( not isinstance(workload, dict) - or "cert_path" not in workload - or "key_path" not in workload + or not isinstance(workload.get("cert_path"), str) + or not workload.get("cert_path") + or not isinstance(workload.get("key_path"), str) + or not workload.get("key_path") ): raise exceptions.ClientCertError( 'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format( @@ -519,7 +551,20 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True): cert_path = workload["cert_path"] key_path = workload["key_path"] - return cert_path, key_path + return cert_path, key_path, absolute_path + + +def _read_credential_bundle_file(bundle_path): + """Reads a combined PEM credential bundle containing certificate(s) and a private key.""" + # Read the bundle once so a certificate rotation on disk cannot pair an + # old certificate chain with a new private key. + with open(bundle_path, "rb") as bundle_file: + bundle_data = bundle_file.read() + + return ( + _extract_cert_chain(bundle_data, bundle_path), + _extract_private_key(bundle_data, bundle_path), + ) def _read_cert_and_key_files(cert_path, key_path): @@ -529,24 +574,32 @@ def _read_cert_and_key_files(cert_path, key_path): return cert_data, key_data -def _read_cert_file(cert_path): - with open(cert_path, "rb") as cert_file: - cert_data = cert_file.read() +def _join_cert_chain(cert_match): + return b"".join( + m if m.endswith(b"\n") or i == len(cert_match) - 1 else m + b"\n" + for i, m in enumerate(cert_match) + ) + +def _has_unmatched_pem_markers(pem_bytes, cert_blocks): + """Checks that every BEGIN/END CERTIFICATE marker belongs to a complete cert block.""" + return len(cert_blocks) != pem_bytes.count(b"-----BEGIN CERTIFICATE-----") or len( + cert_blocks + ) != pem_bytes.count(b"-----END CERTIFICATE-----") + + +def _extract_cert_chain(cert_data, cert_path): cert_match = re.findall(_CERT_REGEX, cert_data) - if len(cert_match) != 1: + if not cert_match or _has_unmatched_pem_markers(cert_data, cert_match): raise exceptions.ClientCertError( - "Certificate file {} is in an invalid format, a single PEM formatted certificate is expected".format( + "Certificate file {} is in an invalid format, at least one PEM formatted certificate is expected".format( cert_path ) ) - return cert_match[0] - + return _join_cert_chain(cert_match) -def _read_key_file(key_path): - with open(key_path, "rb") as key_file: - key_data = key_file.read() +def _extract_private_key(key_data, key_path): key_match = re.findall(_KEY_REGEX, key_data) if len(key_match) != 1: raise exceptions.ClientCertError( @@ -558,6 +611,20 @@ def _read_key_file(key_path): return key_match[0] +def _read_cert_file(cert_path): + with open(cert_path, "rb") as cert_file: + cert_data = cert_file.read() + + return _extract_cert_chain(cert_data, cert_path) + + +def _read_key_file(key_path): + with open(key_path, "rb") as key_file: + key_data = key_file.read() + + return _extract_private_key(key_data, key_path) + + def _run_cert_provider_command(command, expect_encrypted_key=False): """Run the provided command, and return client side mTLS cert, key and passphrase. @@ -591,8 +658,9 @@ def _run_cert_provider_command(command, expect_encrypted_key=False): # Extract certificate (chain), key and passphrase. cert_match = re.findall(_CERT_REGEX, stdout) - if len(cert_match) != 1: + if not cert_match or _has_unmatched_pem_markers(stdout, cert_match): raise exceptions.ClientCertError("Client SSL certificate is missing or invalid") + cert_chain = _join_cert_chain(cert_match) key_match = re.findall(_KEY_REGEX, stdout) if len(key_match) != 1: raise exceptions.ClientCertError("Client SSL key is missing or invalid") @@ -603,13 +671,13 @@ def _run_cert_provider_command(command, expect_encrypted_key=False): raise exceptions.ClientCertError("Passphrase is missing or invalid") if b"ENCRYPTED" not in key_match[0]: raise exceptions.ClientCertError("Encrypted private key is expected") - return cert_match[0], key_match[0], passphrase_match[0].strip() + return cert_chain, key_match[0], passphrase_match[0].strip() if b"ENCRYPTED" in key_match[0]: raise exceptions.ClientCertError("Encrypted private key is not expected") if len(passphrase_match) > 0: raise exceptions.ClientCertError("Passphrase is not expected") - return cert_match[0], key_match[0], None + return cert_chain, key_match[0], None def get_client_ssl_credentials( @@ -768,7 +836,9 @@ def check_use_client_cert(): as True (auto-enabled) if a workload config file exists (pointed at by GOOGLE_API_CERTIFICATE_CONFIG or CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, or the default path like ~/.config/gcloud/certificate_config.json) - containing a "workload" section. + containing a "workload" section, or if no certificate config file is + configured/present and the GKE credential bundle exists at + /var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem. Otherwise, it returns False. Returns: @@ -806,22 +876,26 @@ def check_use_client_cert(): "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.", cert_path, ) - return False + return False + + return _should_use_gke_credential_bundle(cert_path) def check_parameters_for_unauthorized_response(cached_cert): """Returns the cached and current cert fingerprint for reconfiguring mTLS. Args: - cached_cert(bytes): The cached client certificate. + cached_cert (Optional[bytes]): The cached client certificate. Returns: - bytes: The client callback cert bytes. - bytes: The client callback key bytes. - str: The base64-encoded SHA256 cached fingerprint. - str: The base64-encoded SHA256 current cert fingerprint. + Tuple[Optional[bytes], Optional[bytes], Optional[str], Optional[str]]: + The client callback cert bytes, client callback key bytes, + base64-encoded SHA256 cached fingerprint, and base64-encoded SHA256 + current cert fingerprint. """ call_cert_bytes, call_key_bytes = call_client_cert_callback() + if not call_cert_bytes: + return None, None, None, None cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes) current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint( cert_obj @@ -847,16 +921,19 @@ def call_client_cert_callback(): ".mtls.googleapis.com", ".mtls.sandbox.googleapis.com", ".p.googleapis.com", + ".mtls.run.app", ) _MTLS_EXACT_HOSTS = ( "mtls.googleapis.com", "mtls.sandbox.googleapis.com", "p.googleapis.com", + "mtls.run.app", ) def is_mtls_endpoint(url: Optional[Union[str, bytes, object]]) -> bool: - """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint. + """Checks if the given URL corresponds to an mTLS (Google APIs or Cloud Run) + or Private Service Connect (PSC) endpoint. Args: url (Optional[Union[str, bytes, object]]): The request URL. diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py index 54b675b3a1fa..db6d47daf688 100644 --- a/packages/google-auth/google/auth/transport/mtls.py +++ b/packages/google-auth/google/auth/transport/mtls.py @@ -42,10 +42,21 @@ def has_default_client_cert_source(include_context_aware=True): Returns: bool: indicating if the default client cert source exists. """ - cert_path = _mtls_helper._get_cert_config_path( - include_context_aware=include_context_aware - ) - if cert_path is not None: + try: + ( + cert_path, + key_path, + config_file_path, + ) = _mtls_helper._get_workload_cert_and_key_paths( + None, include_context_aware=include_context_aware + ) + except (exceptions.ClientCertError, OSError): + # Config exists, but is malformed or unreadable. Let caller surface error. + return True + + if cert_path is not None and key_path is not None: + return True + if _mtls_helper._should_use_gke_credential_bundle(config_file_path): return True if ( include_context_aware diff --git a/packages/google-auth/google/oauth2/_id_token_async.py b/packages/google-auth/google/oauth2/_id_token_async.py index 38d79ef5d612..024a305c0808 100644 --- a/packages/google-auth/google/oauth2/_id_token_async.py +++ b/packages/google-auth/google/oauth2/_id_token_async.py @@ -194,7 +194,7 @@ async def verify_firebase_token( ) -async def fetch_id_token(request, audience): +async def fetch_id_token(request, audience, bind_id_token=None): """Fetch the ID Token from the current environment. This function acquires ID token from the environment in the following order. @@ -223,6 +223,19 @@ async def fetch_id_token(request, audience): request (google.auth.transport.aiohttp_requests.Request): A callable used to make HTTP requests. audience (str): The audience that this ID token is intended for. + bind_id_token (Optional[bool]): Controls whether to request a + certificate-bound ID token from the metadata server identity + endpoint. If ``True``, requests a bound token whenever a valid + Agent Identity certificate is available and token binding is not + disabled via ``GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN``, falling back + to an unbound token otherwise (or raising + :class:`~google.auth.exceptions.RefreshError` if a configured + certificate is not found after retries). If ``False``, always + requests an unbound token. If ``None`` (default), token binding is + determined automatically by the library. Set ``True`` or ``False`` + explicitly if your application requires a specific behavior. Has no + effect when credentials are loaded from + ``GOOGLE_APPLICATION_CREDENTIALS``. Returns: str: The ID token. @@ -231,6 +244,9 @@ async def fetch_id_token(request, audience): ~google.auth.exceptions.DefaultCredentialsError: If metadata server doesn't exist and no valid service account credentials are found. + ~google.auth.exceptions.RefreshError: + If an error occurred while fetching the ID token or if a required + certificate file is not found after retries. """ # 1. Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment # variable. @@ -273,7 +289,10 @@ async def fetch_id_token(request, audience): request_new = requests.Request() if _metadata.ping(request_new): credentials = compute_engine.IDTokenCredentials( - request_new, audience, use_metadata_identity_endpoint=True + request_new, + audience, + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, ) credentials.refresh(request_new) return credentials.token diff --git a/packages/google-auth/google/oauth2/id_token.py b/packages/google-auth/google/oauth2/id_token.py index 1cab9440a61a..c73d11b50f94 100644 --- a/packages/google-auth/google/oauth2/id_token.py +++ b/packages/google-auth/google/oauth2/id_token.py @@ -236,7 +236,7 @@ def verify_firebase_token(id_token, request, audience=None, clock_skew_in_second ) -def fetch_id_token_credentials(audience, request=None): +def fetch_id_token_credentials(audience, request=None, bind_id_token=None): """Create the ID Token credentials from the current environment. This function acquires ID token from the environment in the following order. @@ -272,6 +272,19 @@ def fetch_id_token_credentials(audience, request=None): audience (str): The audience that this ID token is intended for. request (Optional[google.auth.transport.Request]): A callable used to make HTTP requests. A request object will be created if not provided. + bind_id_token (Optional[bool]): Controls whether to request a + certificate-bound ID token from the metadata server identity + endpoint. If ``True``, requests a bound token whenever a valid + Agent Identity certificate is available and token binding is not + disabled via ``GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN``, falling back + to an unbound token otherwise (or raising + :class:`~google.auth.exceptions.RefreshError` if a configured + certificate is not found after retries). If ``False``, always + requests an unbound token. If ``None`` (default), token binding is + determined automatically by the library. Set ``True`` or ``False`` + explicitly if your application requires a specific behavior. Has no + effect when credentials are loaded from + ``GOOGLE_APPLICATION_CREDENTIALS``. Returns: google.auth.credentials.Credentials: The ID token credentials. @@ -335,7 +348,10 @@ def fetch_id_token_credentials(audience, request=None): if _metadata.ping(request): return compute_engine.IDTokenCredentials( - request, audience, use_metadata_identity_endpoint=True + request, + audience, + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, ) except (ImportError, exceptions.TransportError): pass @@ -345,7 +361,7 @@ def fetch_id_token_credentials(audience, request=None): ) -def fetch_id_token(request, audience): +def fetch_id_token(request, audience, bind_id_token=None): """Fetch the ID Token from the current environment. This function acquires ID token from the environment in the following order. @@ -374,6 +390,19 @@ def fetch_id_token(request, audience): request (google.auth.transport.Request): A callable used to make HTTP requests. audience (str): The audience that this ID token is intended for. + bind_id_token (Optional[bool]): Controls whether to request a + certificate-bound ID token from the metadata server identity + endpoint. If ``True``, requests a bound token whenever a valid + Agent Identity certificate is available and token binding is not + disabled via ``GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN``, falling back + to an unbound token otherwise (or raising + :class:`~google.auth.exceptions.RefreshError` if a configured + certificate is not found after retries). If ``False``, always + requests an unbound token. If ``None`` (default), token binding is + determined automatically by the library. Set ``True`` or ``False`` + explicitly if your application requires a specific behavior. Has no + effect when credentials are loaded from + ``GOOGLE_APPLICATION_CREDENTIALS``. Returns: str: The ID token. @@ -382,7 +411,12 @@ def fetch_id_token(request, audience): ~google.auth.exceptions.DefaultCredentialsError: If metadata server doesn't exist and no valid service account credentials are found. + ~google.auth.exceptions.RefreshError: + If an error occurred while fetching the ID token or if a required + certificate file is not found after retries. """ - id_token_credentials = fetch_id_token_credentials(audience, request=request) + id_token_credentials = fetch_id_token_credentials( + audience, request=request, bind_id_token=bind_id_token + ) id_token_credentials.refresh(request) return id_token_credentials.token diff --git a/packages/google-auth/tests/compute_engine/test__metadata.py b/packages/google-auth/tests/compute_engine/test__metadata.py index 0b151722af07..af8573a80c66 100644 --- a/packages/google-auth/tests/compute_engine/test__metadata.py +++ b/packages/google-auth/tests/compute_engine/test__metadata.py @@ -484,6 +484,57 @@ def test_get_failure_bad_json(): ) +def test_get_body_with_get_method_raises_value_error(): + request = make_request("{}") + + with pytest.raises( + ValueError, match="Request body cannot be specified with GET method." + ): + _metadata.get(request, PATH, method="GET", body=b"some_body") + + request.assert_not_called() + + +@mock.patch("time.sleep", return_value=None) +def test_get_post_retry_preserves_method_body_and_headers(mock_sleep): + response_503 = mock.create_autospec(transport.Response, instance=True) + response_503.status = http_client.SERVICE_UNAVAILABLE + response_503.data = _helpers.to_bytes("Service Unavailable") + response_503.headers = {} + + response_ok = mock.create_autospec(transport.Response, instance=True) + response_ok.status = http_client.OK + response_ok.data = _helpers.to_bytes( + json.dumps({"access_token": "bound_token", "expires_in": 3600}) + ) + response_ok.headers = {"content-type": "application/json"} + + request = mock.create_autospec(transport.Request) + request.side_effect = [ + response_503, + exceptions.TransportError("transient transport error"), + response_ok, + ] + + expected_body = json.dumps({"certificate_chain": "fake_pem_chain"}).encode("utf-8") + result = _metadata.get( + request, + PATH, + method="POST", + body=expected_body, + headers={"Content-Type": "application/json"}, + ) + + assert result == {"access_token": "bound_token", "expires_in": 3600} + assert request.call_count == 3 + for call_args in request.call_args_list: + _, kwargs = call_args + assert kwargs["method"] == "POST" + assert kwargs["body"] == expected_body + assert kwargs["headers"]["Content-Type"] == "application/json" + assert kwargs["headers"][_metadata._METADATA_FLAVOR_HEADER] == "Google" + + def test_get_project_id(): project = "example-project" request = make_request(project, headers={"content-type": "text/plain"}) @@ -635,18 +686,12 @@ def test_get_universe_domain_other_error(): ) -@mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate", - return_value=None, -) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) -def test_get_service_account_token( - utcnow, mock_metrics_header_value, mock_get_agent_cert -): +def test_get_service_account_token(utcnow, mock_metrics_header_value): ttl = 500 request = make_request( json.dumps({"access_token": "token", "expires_in": ttl}), @@ -668,18 +713,12 @@ def test_get_service_account_token( assert expiry == utcnow() + datetime.timedelta(seconds=ttl) -@mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate", - return_value=None, -) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) -def test_get_service_account_token_with_scopes_list( - utcnow, mock_metrics_header_value, mock_get_agent_cert -): +def test_get_service_account_token_with_scopes_list(utcnow, mock_metrics_header_value): ttl = 500 request = make_request( json.dumps({"access_token": "token", "expires_in": ttl}), @@ -704,17 +743,13 @@ def test_get_service_account_token_with_scopes_list( assert expiry == utcnow() + datetime.timedelta(seconds=ttl) -@mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate", - return_value=None, -) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) def test_get_service_account_token_with_scopes_string( - utcnow, mock_metrics_header_value, mock_get_agent_cert + utcnow, mock_metrics_header_value ): ttl = 500 request = make_request( @@ -740,10 +775,9 @@ def test_get_service_account_token_with_scopes_string( assert expiry == utcnow() + datetime.timedelta(seconds=ttl) -@mock.patch("google.auth._agent_identity_utils.calculate_certificate_fingerprint") @mock.patch("google.auth._agent_identity_utils.should_request_bound_token") @mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate" + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" ) @mock.patch( "google.auth.metrics.token_request_access_token_mds", @@ -753,38 +787,39 @@ def test_get_service_account_token_with_scopes_string( def test_get_service_account_token_with_bound_token( utcnow, mock_metrics_header_value, - mock_get_and_parse, + mock_get_cert_and_bytes, mock_should_request, - mock_calculate_fingerprint, ): # Test the successful path where a certificate is found and a bound token # is requested. mock_cert = mock.sentinel.cert - mock_get_and_parse.return_value = mock_cert + mock_cert_bytes = b"fake_cert_bytes" + mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes) mock_should_request.return_value = True - mock_calculate_fingerprint.return_value = "fake_fingerprint" token_response = json.dumps({"access_token": "token", "expires_in": 3600}) request = make_request(token_response, headers={"content-type": "application/json"}) _metadata.get_service_account_token(request) - mock_get_and_parse.assert_called_once() + mock_get_cert_and_bytes.assert_called_once() mock_should_request.assert_called_once_with(mock_cert) - mock_calculate_fingerprint.assert_called_once_with(mock_cert) request.assert_called_once() _, kwargs = request.call_args - url = kwargs["url"] - assert "bindCertificateFingerprint=fake_fingerprint" in url + assert kwargs["method"] == "POST" + assert kwargs["body"] == json.dumps( + {"certificate_chain": mock_cert_bytes.decode("utf-8")} + ).encode("utf-8") + assert kwargs["headers"]["Content-Type"] == "application/json" @mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate" + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" ) -def test_get_service_account_token_no_cert(mock_get_and_parse): - # Test that no fingerprint is added when no certificate is found. - mock_get_and_parse.return_value = None +def test_get_service_account_token_no_cert(mock_get_cert_and_bytes): + # Test that a standard GET request without a body is sent when no certificate is found. + mock_get_cert_and_bytes.return_value = (None, None) token_response = json.dumps({"access_token": "token", "expires_in": 3600}) request = make_request(token_response, headers={"content-type": "application/json"}) @@ -792,19 +827,20 @@ def test_get_service_account_token_no_cert(mock_get_and_parse): request.assert_called_once() _, kwargs = request.call_args - url = kwargs["url"] - assert "bindCertificateFingerprint" not in url + assert kwargs["method"] == "GET" + assert "body" not in kwargs + assert "Content-Type" not in kwargs["headers"] @mock.patch("google.auth._agent_identity_utils.should_request_bound_token") @mock.patch( - "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate" + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" ) def test_get_service_account_token_should_not_bind( - mock_get_and_parse, mock_should_request + mock_get_cert_and_bytes, mock_should_request ): - # Test that no fingerprint is added when a cert is found but should not be used. - mock_get_and_parse.return_value = mock.sentinel.cert + # Test that a standard GET request without a body is sent when a cert is found but should not be used. + mock_get_cert_and_bytes.return_value = (mock.sentinel.cert, b"fake_cert_bytes") mock_should_request.return_value = False token_response = json.dumps({"access_token": "token", "expires_in": 3600}) request = make_request(token_response, headers={"content-type": "application/json"}) @@ -813,8 +849,23 @@ def test_get_service_account_token_should_not_bind( request.assert_called_once() _, kwargs = request.call_args - url = kwargs["url"] - assert "bindCertificateFingerprint" not in url + assert kwargs["method"] == "GET" + assert "body" not in kwargs + assert "Content-Type" not in kwargs["headers"] + + +@mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" +) +def test__build_token_request_options_unbound(mock_get_cert_and_bytes): + method, body, headers = _metadata._build_token_request_options( + ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, bind_token=False + ) + + mock_get_cert_and_bytes.assert_not_called() + assert method == "GET" + assert body is None + assert headers == {"x-goog-api-client": ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE} def test_get_service_account_info(): diff --git a/packages/google-auth/tests/compute_engine/test_credentials.py b/packages/google-auth/tests/compute_engine/test_credentials.py index 6e30d8807bf3..9490ca3e010a 100644 --- a/packages/google-auth/tests/compute_engine/test_credentials.py +++ b/packages/google-auth/tests/compute_engine/test_credentials.py @@ -13,13 +13,21 @@ # limitations under the License. import base64 import datetime +import json import re from unittest import mock import pytest # type: ignore import responses # type: ignore -from google.auth import _helpers, exceptions, jwt, transport +from google.auth import ( + _helpers, + environment_vars, + exceptions, + jwt, + metrics, + transport, +) from google.auth.compute_engine import credentials from google.auth.transport import requests @@ -471,29 +479,23 @@ def test_regional_access_boundary_disabled_state_transitions( # Subsequent check calls should return False early assert creds._is_regional_access_boundary_lookup_required() is False - @mock.patch("google.auth.compute_engine._metadata.get") - @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - @mock.patch("google.auth._agent_identity_utils.parse_certificate") @mock.patch( - "google.auth._agent_identity_utils.should_request_bound_token", - return_value=True, + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" ) @mock.patch( - "google.auth._agent_identity_utils.calculate_certificate_fingerprint", - return_value="fingerprint", + "google.auth._agent_identity_utils.should_request_bound_token", + return_value=True, ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_refresh_with_agent_identity( self, - mock_calculate_fingerprint, - mock_should_request, - mock_parse_certificate, - mock_get_path, mock_metadata_get, - tmpdir, + mock_should_request, + mock_get_cert_and_bytes, ): - cert_path = tmpdir.join("cert.pem") - cert_path.write(b"cert_content") - mock_get_path.return_value = str(cert_path) + mock_cert = mock.sentinel.cert + mock_cert_bytes = b"cert_content" + mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes) mock_metadata_get.side_effect = [ { @@ -506,33 +508,32 @@ def test_refresh_with_agent_identity( self.credentials.refresh(None) assert self.credentials.token == "token" - mock_parse_certificate.assert_called_once_with(b"cert_content") - mock_should_request.assert_called_once_with(mock_parse_certificate.return_value) + mock_get_cert_and_bytes.assert_called_once() + mock_should_request.assert_called_once_with(mock_cert) kwargs = mock_metadata_get.call_args[1] assert kwargs["params"] == { "scopes": "one,two", - "bindCertificateFingerprint": "fingerprint", } + assert kwargs["method"] == "POST" + assert kwargs["body"] == json.dumps( + {"certificate_chain": mock_cert_bytes.decode("utf-8")} + ).encode("utf-8") + assert kwargs["headers"]["Content-Type"] == "application/json" - @mock.patch("google.auth.compute_engine._metadata.get") - @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - @mock.patch("google.auth._agent_identity_utils.parse_certificate") + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) @mock.patch( "google.auth._agent_identity_utils.should_request_bound_token", return_value=False, ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_refresh_with_agent_identity_opt_out_or_not_agent( self, - mock_should_request, - mock_parse_certificate, - mock_get_path, mock_metadata_get, - tmpdir, + mock_should_request, + mock_get_cert_and_bytes, ): - cert_path = tmpdir.join("cert.pem") - cert_path.write(b"cert_content") - mock_get_path.return_value = str(cert_path) - mock_metadata_get.side_effect = [ { "email": "service-account@project.iam.gserviceaccount.com", @@ -541,13 +542,44 @@ def test_refresh_with_agent_identity_opt_out_or_not_agent( {"access_token": "token", "expires_in": 500}, ] + mock_cert = mock.sentinel.cert + mock_cert_bytes = b"cert_content" + mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes) + + self.credentials.refresh(None) + + assert self.credentials.token == "token" + mock_get_cert_and_bytes.assert_called_once() + mock_should_request.assert_called_once_with(mock_cert) + kwargs = mock_metadata_get.call_args[1] + assert kwargs["method"] == "GET" + assert kwargs["body"] is None + assert "Content-Type" not in kwargs["headers"] + + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) + def test_refresh_without_agent_identity_certificate( + self, + mock_metadata_get, + mock_get_cert_and_bytes, + ): + mock_metadata_get.side_effect = [ + {"email": "service-account@example.com", "scopes": ["one", "two"]}, + {"access_token": "token", "expires_in": 500}, + ] + + mock_get_cert_and_bytes.return_value = (None, None) + self.credentials.refresh(None) assert self.credentials.token == "token" - mock_parse_certificate.assert_called_once_with(b"cert_content") - mock_should_request.assert_called_once_with(mock_parse_certificate.return_value) + mock_get_cert_and_bytes.assert_called_once() kwargs = mock_metadata_get.call_args[1] - assert "bindCertificateFingerprint" not in kwargs.get("params", {}) + assert kwargs["method"] == "GET" + assert kwargs["body"] is None + assert "Content-Type" not in kwargs["headers"] def test_set_blocking_regional_access_boundary_lookup(self): creds = self.credentials @@ -855,6 +887,204 @@ def test_with_target_audience_integration(self): assert self.credentials.token is not None + @pytest.mark.parametrize( + "cert_config_env,bind_id_token", + [(None, None), (None, True), ("/path/to/config.json", True)], + ) + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) + @mock.patch( + "google.auth._agent_identity_utils.should_request_bound_token", + return_value=True, + ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) + def test_refresh_with_agent_identity( + self, + mock_metadata_get, + mock_should_request, + mock_get_cert_and_bytes, + cert_config_env, + bind_id_token, + monkeypatch, + ): + if cert_config_env: + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, cert_config_env + ) + mock_metadata_get.side_effect = [ + {"email": "service-account@example.com", "scopes": ["one", "two"]}, + SAMPLE_ID_TOKEN, + ] + + mock_cert = mock.sentinel.cert + mock_cert_bytes = b"cert_content" + mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes) + + request = mock.create_autospec(transport.Request, instance=True) + self.credentials = credentials.IDTokenCredentials( + request=request, + target_audience="https://audience.com", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, + ) + + self.credentials.refresh(None) + + assert self.credentials.token == SAMPLE_ID_TOKEN + mock_get_cert_and_bytes.assert_called_once() + mock_should_request.assert_called_once_with(mock_cert) + + kwargs = mock_metadata_get.call_args[1] + assert kwargs["method"] == "POST" + assert kwargs["body"] == json.dumps( + {"certificate_chain": mock_cert_bytes.decode("utf-8")} + ).encode("utf-8") + assert kwargs["headers"]["Content-Type"] == "application/json" + assert ( + kwargs["headers"][metrics.API_CLIENT_HEADER] + == metrics.token_request_id_token_mds() + ) + + @pytest.mark.parametrize( + "cert_config_env,bind_id_token", + [ + (None, False), + ("/path/to/config.json", False), + ("/path/to/config.json", None), + ], + ) + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) + def test_refresh_with_agent_identity_unbound( + self, + mock_metadata_get, + mock_get_cert_and_bytes, + cert_config_env, + bind_id_token, + monkeypatch, + ): + if cert_config_env: + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, cert_config_env + ) + mock_metadata_get.side_effect = [ + {"email": "service-account@example.com", "scopes": ["one", "two"]}, + SAMPLE_ID_TOKEN, + ] + + request = mock.create_autospec(transport.Request, instance=True) + self.credentials = credentials.IDTokenCredentials( + request=request, + target_audience="https://audience.com", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, + ) + + self.credentials.refresh(None) + + assert self.credentials.token == SAMPLE_ID_TOKEN + mock_get_cert_and_bytes.assert_not_called() + + kwargs = mock_metadata_get.call_args[1] + assert kwargs["method"] == "GET" + assert kwargs["body"] is None + assert "Content-Type" not in kwargs["headers"] + assert ( + kwargs["headers"][metrics.API_CLIENT_HEADER] + == metrics.token_request_id_token_mds() + ) + + @pytest.mark.parametrize("bind_id_token", [None, True]) + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) + @mock.patch( + "google.auth._agent_identity_utils.should_request_bound_token", + return_value=False, + ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) + def test_refresh_with_agent_identity_opt_out_or_not_agent( + self, + mock_metadata_get, + mock_should_request, + mock_get_cert_and_bytes, + bind_id_token, + ): + mock_metadata_get.side_effect = [ + {"email": "service-account@example.com", "scopes": ["one", "two"]}, + SAMPLE_ID_TOKEN, + ] + + mock_cert = mock.sentinel.cert + mock_cert_bytes = b"cert_content" + mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes) + + request = mock.create_autospec(transport.Request, instance=True) + self.credentials = credentials.IDTokenCredentials( + request=request, + target_audience="https://audience.com", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, + ) + + self.credentials.refresh(None) + + assert self.credentials.token == SAMPLE_ID_TOKEN + mock_get_cert_and_bytes.assert_called_once() + mock_should_request.assert_called_once_with(mock_cert) + + kwargs = mock_metadata_get.call_args[1] + assert kwargs["method"] == "GET" + assert kwargs["body"] is None + assert "Content-Type" not in kwargs["headers"] + assert ( + kwargs["headers"][metrics.API_CLIENT_HEADER] + == metrics.token_request_id_token_mds() + ) + + @pytest.mark.parametrize("bind_id_token", [None, True]) + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes" + ) + @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) + def test_refresh_without_agent_identity_certificate( + self, + mock_metadata_get, + mock_get_cert_and_bytes, + bind_id_token, + ): + mock_metadata_get.side_effect = [ + {"email": "service-account@example.com", "scopes": ["one", "two"]}, + SAMPLE_ID_TOKEN, + ] + + mock_get_cert_and_bytes.return_value = (None, None) + + request = mock.create_autospec(transport.Request, instance=True) + self.credentials = credentials.IDTokenCredentials( + request=request, + target_audience="https://audience.com", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, + ) + + self.credentials.refresh(None) + + assert self.credentials.token == SAMPLE_ID_TOKEN + mock_get_cert_and_bytes.assert_called_once() + + kwargs = mock_metadata_get.call_args[1] + assert kwargs["method"] == "GET" + assert kwargs["body"] is None + assert "Content-Type" not in kwargs["headers"] + assert ( + kwargs["headers"][metrics.API_CLIENT_HEADER] + == metrics.token_request_id_token_mds() + ) + @mock.patch( "google.auth._helpers.utcnow", return_value=_helpers.utcfromtimestamp(0), @@ -1220,6 +1450,7 @@ def test_get_id_token_from_metadata( assert cred.token == SAMPLE_ID_TOKEN assert cred.expiry == _helpers.utcfromtimestamp(SAMPLE_ID_TOKEN_EXP) assert cred._use_metadata_identity_endpoint + assert cred._bind_id_token is None assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@project.iam.gserviceaccount.com" @@ -1227,43 +1458,120 @@ def test_get_id_token_from_metadata( with pytest.raises(ValueError): cred.sign_bytes(b"bytes") - @mock.patch( - "google.auth.compute_engine._metadata.get_service_account_info", autospec=True - ) - def test_with_target_audience_for_metadata(self, get_service_account_info): - get_service_account_info.return_value = { - "email": "foo@project.iam.gserviceaccount.com" - } + @pytest.mark.parametrize("bind_id_token", [None, True, False]) + @responses.activate + def test_with_target_audience_for_metadata(self, bind_id_token): + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/?recursive=true", + status=200, + content_type="application/json", + json={"email": "foo@project.iam.gserviceaccount.com"}, + ) + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/identity?audience=new_audience&format=full", + status=200, + content_type="text/plain", + body=SAMPLE_ID_TOKEN, + ) cred = credentials.IDTokenCredentials( - mock.Mock(), "audience", use_metadata_identity_endpoint=True + requests.Request(), + "audience", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, ) cred = cred.with_target_audience("new_audience") + cred.refresh(requests.Request()) assert cred._target_audience == "new_audience" assert cred._use_metadata_identity_endpoint + assert cred._bind_id_token is bind_id_token assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@project.iam.gserviceaccount.com" + assert cred.token == SAMPLE_ID_TOKEN.decode("utf-8") - @mock.patch( - "google.auth.compute_engine._metadata.get_service_account_info", autospec=True - ) - def test_id_token_with_quota_project(self, get_service_account_info): - get_service_account_info.return_value = { - "email": "foo@project.iam.gserviceaccount.com" - } + @pytest.mark.parametrize("bind_id_token", [None, True, False]) + @responses.activate + def test_id_token_with_quota_project(self, bind_id_token): + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/?recursive=true", + status=200, + content_type="application/json", + json={"email": "foo@project.iam.gserviceaccount.com"}, + ) + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/identity?audience=audience&format=full", + status=200, + content_type="text/plain", + body=SAMPLE_ID_TOKEN, + ) cred = credentials.IDTokenCredentials( - mock.Mock(), "audience", use_metadata_identity_endpoint=True + requests.Request(), + "audience", + use_metadata_identity_endpoint=True, + bind_id_token=bind_id_token, ) cred = cred.with_quota_project("project-foo") + cred.refresh(requests.Request()) assert cred._quota_project_id == "project-foo" assert cred._use_metadata_identity_endpoint + assert cred._bind_id_token is bind_id_token assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@project.iam.gserviceaccount.com" + assert cred.token == SAMPLE_ID_TOKEN.decode("utf-8") + + @mock.patch( + "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes", + return_value=(None, None), + ) + @responses.activate + def test_id_token_missing_bind_id_token_attribute(self, mock_get_agent_cert): + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/?recursive=true", + status=200, + content_type="application/json", + json={"email": "foo@project.iam.gserviceaccount.com"}, + ) + responses.add( + responses.GET, + "http://metadata.google.internal/computeMetadata/v1/instance/" + "service-accounts/default/identity?audience=audience&format=full", + status=200, + content_type="text/plain", + body=SAMPLE_ID_TOKEN, + ) + + cred = credentials.IDTokenCredentials( + requests.Request(), + "audience", + use_metadata_identity_endpoint=True, + ) + # Simulate credentials pickled with an older version without _bind_id_token. + del cred._bind_id_token + + cred_with_audience = cred.with_target_audience("new_audience") + assert cred_with_audience._bind_id_token is None + + cred_with_quota = cred.with_quota_project("project-foo") + assert cred_with_quota._bind_id_token is None + + cred.refresh(requests.Request()) + assert cred.token == SAMPLE_ID_TOKEN.decode("utf-8") + mock_get_agent_cert.assert_called_once() @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True @@ -1329,3 +1637,14 @@ def test_get_id_token_from_metadata_constructor(self): use_metadata_identity_endpoint=True, service_account_email="foo@project.iam.gserviceaccount.com", ) + for bind_id_token in (True, False): + with pytest.raises( + ValueError, + match="If use_metadata_identity_endpoint is False, bind_id_token must not be set", + ): + credentials.IDTokenCredentials( + mock.Mock(), + "audience", + use_metadata_identity_endpoint=False, + bind_id_token=bind_id_token, + ) diff --git a/packages/google-auth/tests/conftest.py b/packages/google-auth/tests/conftest.py index fbce2575e929..3be7b4ad1b05 100644 --- a/packages/google-auth/tests/conftest.py +++ b/packages/google-auth/tests/conftest.py @@ -18,6 +18,9 @@ import pytest # type: ignore +from google.auth import environment_vars +from google.auth.transport import _mtls_helper + def pytest_configure(): """Load public certificate and private key.""" @@ -29,6 +32,35 @@ def pytest_configure(): with open(os.path.join(pytest.data_dir, "public_cert.pem"), "rb") as fh: pytest.public_cert_bytes = fh.read() + _mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH = os.path.join( + pytest.data_dir, "nonexistent_gke_credential_bundle.pem" + ) + + +@pytest.fixture(autouse=True) +def clean_cert_config_env(monkeypatch, tmp_path): + monkeypatch.delenv( + environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, + raising=False, + ) + monkeypatch.delenv( + environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE, + raising=False, + ) + monkeypatch.delenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, + raising=False, + ) + monkeypatch.delenv( + environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, + raising=False, + ) + monkeypatch.setattr( + _mtls_helper, + "_GKE_CREDENTIAL_BUNDLE_PATH", + str(tmp_path / "gke" / "credential" / "bundle" / "path"), + ) + @pytest.fixture def mock_non_existent_module(monkeypatch): diff --git a/packages/google-auth/tests/oauth2/test_id_token.py b/packages/google-auth/tests/oauth2/test_id_token.py index 456ca58b89f6..07fd2c3f1935 100644 --- a/packages/google-auth/tests/oauth2/test_id_token.py +++ b/packages/google-auth/tests/oauth2/test_id_token.py @@ -285,7 +285,18 @@ def test_fetch_id_token_credentials_optional_request(monkeypatch): mock_request.assert_called() -def test_fetch_id_token_credentials_from_metadata_server(monkeypatch): +@pytest.mark.parametrize( + "call_kwargs,expected_bind_id_token", + [ + ({}, None), + ({"bind_id_token": None}, None), + ({"bind_id_token": True}, True), + ({"bind_id_token": False}, False), + ], +) +def test_fetch_id_token_credentials_from_metadata_server( + monkeypatch, call_kwargs, expected_bind_id_token +): monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False) mock_req = mock.Mock() @@ -294,24 +305,39 @@ def test_fetch_id_token_credentials_from_metadata_server(monkeypatch): with mock.patch( "google.auth.compute_engine.IDTokenCredentials.__init__", return_value=None ) as mock_init: - id_token.fetch_id_token_credentials(ID_TOKEN_AUDIENCE, request=mock_req) + id_token.fetch_id_token_credentials( + ID_TOKEN_AUDIENCE, request=mock_req, **call_kwargs + ) mock_init.assert_called_once_with( - mock_req, ID_TOKEN_AUDIENCE, use_metadata_identity_endpoint=True + mock_req, + ID_TOKEN_AUDIENCE, + use_metadata_identity_endpoint=True, + bind_id_token=expected_bind_id_token, ) -def test_fetch_id_token_credentials_from_explicit_cred_json_file(monkeypatch): +@pytest.mark.parametrize("bind_id_token", [None, True, False]) +def test_fetch_id_token_credentials_from_explicit_cred_json_file( + monkeypatch, bind_id_token +): monkeypatch.setenv(environment_vars.CREDENTIALS, SERVICE_ACCOUNT_FILE) - cred = id_token.fetch_id_token_credentials(ID_TOKEN_AUDIENCE) + cred = id_token.fetch_id_token_credentials( + ID_TOKEN_AUDIENCE, bind_id_token=bind_id_token + ) assert isinstance(cred, service_account.IDTokenCredentials) assert cred._target_audience == ID_TOKEN_AUDIENCE -def test_fetch_id_token_credentials_from_impersonated_cred_json_file(monkeypatch): +@pytest.mark.parametrize("bind_id_token", [None, True, False]) +def test_fetch_id_token_credentials_from_impersonated_cred_json_file( + monkeypatch, bind_id_token +): monkeypatch.setenv(environment_vars.CREDENTIALS, IMPERSONATED_SERVICE_ACCOUNT_FILE) - cred = id_token.fetch_id_token_credentials(ID_TOKEN_AUDIENCE) + cred = id_token.fetch_id_token_credentials( + ID_TOKEN_AUDIENCE, bind_id_token=bind_id_token + ) assert isinstance(cred, impersonated_credentials.IDTokenCredentials) assert cred._target_audience == ID_TOKEN_AUDIENCE @@ -373,7 +399,16 @@ def test_fetch_id_token_credentials_invalid_cred_path(monkeypatch): ) -def test_fetch_id_token(monkeypatch): +@pytest.mark.parametrize( + "call_kwargs,expected_bind_id_token", + [ + ({}, None), + ({"bind_id_token": None}, None), + ({"bind_id_token": True}, True), + ({"bind_id_token": False}, False), + ], +) +def test_fetch_id_token(call_kwargs, expected_bind_id_token): mock_cred = mock.MagicMock() mock_cred.token = "token" @@ -382,7 +417,9 @@ def test_fetch_id_token(monkeypatch): with mock.patch( "google.oauth2.id_token.fetch_id_token_credentials", return_value=mock_cred ) as mock_fetch: - token = id_token.fetch_id_token(mock_req, ID_TOKEN_AUDIENCE) - mock_fetch.assert_called_once_with(ID_TOKEN_AUDIENCE, request=mock_req) + token = id_token.fetch_id_token(mock_req, ID_TOKEN_AUDIENCE, **call_kwargs) + mock_fetch.assert_called_once_with( + ID_TOKEN_AUDIENCE, request=mock_req, bind_id_token=expected_bind_id_token + ) mock_cred.refresh.assert_called_once_with(mock_req) assert token == "token" diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py index 7448dfa7aa0b..9ab375f840c2 100644 --- a/packages/google-auth/tests/test_agent_identity_utils.py +++ b/packages/google-auth/tests/test_agent_identity_utils.py @@ -16,6 +16,7 @@ import hashlib import json import os +import stat import urllib.parse from unittest import mock @@ -23,6 +24,7 @@ from cryptography import x509 from google.auth import _agent_identity_utils, environment_vars, exceptions +from google.auth.transport import _mtls_helper # A mock PEM-encoded certificate without an Agent Identity SPIFFE ID. NON_AGENT_IDENTITY_CERT_BYTES = ( @@ -48,34 +50,202 @@ ) -class TestAgentIdentityUtils: - @pytest.fixture(autouse=True) - def clean_env(self, monkeypatch): - monkeypatch.delenv( - environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, - raising=False, - ) - monkeypatch.delenv( - environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE, - raising=False, - ) +# Synthetic self-signed test certificate (CN=agent-identity-test) with SAN URI +# "spiffe://agents.global.proj-12345.system.id.goog/workload". +# To regenerate safely, use cryptography.x509.CertificateBuilder with a +# throwaway key (discard the private key; never commit it) and a dummy project +# ID in the SPIFFE URI. +AGENT_IDENTITY_CERT_BYTES = ( + b"-----BEGIN CERTIFICATE-----\n" + b"MIIDEjCCAfqgAwIBAgIUKZAXnXnxf8hsn+ojS1N8bN3hXrUwDQYJKoZIhvcNAQEL\n" + b"BQAwHjEcMBoGA1UEAwwTYWdlbnQtaWRlbnRpdHktdGVzdDAeFw0yNDAxMDEwMDAw\n" + b"MDBaFw0zNDAxMDEwMDAwMDBaMB4xHDAaBgNVBAMME2FnZW50LWlkZW50aXR5LXRl\n" + b"c3QwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQC+3eSHkp1oBj7rFehL\n" + b"5VJmBF9KLZ5PXQuZNYrGSpxGJ0Dx1T5ancrl8e66AfAepw9O4zdA+8Afub39PQLh\n" + b"wMTEY3O9Uqetch+2apkwXQ+yYpnorgMwqykY77ptApA8WPHzEOj58FPtyC4UqXJ7\n" + b"YKVpN92lVi1l73XBn6axo/q72KjeEdssR6UMtAd3dbGqY3af/AZNppJWRmCMWs8Z\n" + b"oAuxTH5LqYuxwvCDfYLpmQSbv4IJ/UBjkvjRIlzPo2zHg9PMdf/j6Bg9n3kkFaVH\n" + b"Oep+Zm+DtdT8JvwnG3sQ8Qn/ZCqU0z3DkT//XaElAikLwr/1MFrVHhfrYEWnDvuy\n" + b"9PHxAgMBAAGjSDBGMEQGA1UdEQQ9MDuGOXNwaWZmZTovL2FnZW50cy5nbG9iYWwu\n" + b"cHJvai0xMjM0NS5zeXN0ZW0uaWQuZ29vZy93b3JrbG9hZDANBgkqhkiG9w0BAQsF\n" + b"AAOCAQEAWEyBk7TbetWeQTYEdJwH/pNmiqoCzDYcqCSuNqJhrItHuLmSAlKBGCz6\n" + b"I6ptzY6vT7ARXoW07ivf9Ffl3TMUDLjd5Tkfn1q8JjyM1Ugbfuq7rdF2g9+5h6wg\n" + b"tjeV10LqAimr+fFaNvRiGsMfokuwPyUKYe/9d6x5NhcTTNgMQDG5SWnRe1JqPy94\n" + b"GKilWCyzDl4qzHAU5gc7lZ/6WKbYPwjJDDT4/d3AvNx1O/cQCG7Mz4veDuG2Jqh+\n" + b"FPUqQ4G9RL4zdPuXlbKfSknkmZWld1+adyitai6BzDCG9zkEEVJmLE2/e3XvNC93\n" + b"fa2asspu5y/ViCmPS0J2rzWEk7zI5w==\n" + b"-----END CERTIFICATE-----\n" +) + +@pytest.fixture(autouse=True) +def clean_env(monkeypatch): + monkeypatch.delenv( + environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, + raising=False, + ) + monkeypatch.delenv( + environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE, + raising=False, + ) + monkeypatch.delenv( + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, + raising=False, + ) + monkeypatch.delenv( + environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + raising=False, + ) + + +class TestAgentIdentityUtils: @mock.patch("cryptography.x509.load_pem_x509_certificate") def test_parse_certificate(self, mock_load_cert): + mock_load_cert.return_value = mock.sentinel.cert result = _agent_identity_utils.parse_certificate(b"cert_bytes") mock_load_cert.assert_called_once_with(b"cert_bytes") - assert result == mock_load_cert.return_value + assert result == mock.sentinel.cert + + @pytest.mark.parametrize("invalid_input", [b"", None]) + def test_parse_certificate_empty_or_none_raises_value_error(self, invalid_input): + with pytest.raises( + ValueError, match="Certificate bytes cannot be empty or None" + ): + _agent_identity_utils.parse_certificate(invalid_input) + + @pytest.mark.parametrize( + "second_cert_block", + [ + # Valid Base64, invalid ASN.1 DER + b"-----BEGIN CERTIFICATE-----\n" + + base64.b64encode(b"not valid asn1 der payload") + + b"\n-----END CERTIFICATE-----\n", + # Corrupted Base64 + b"-----BEGIN CERTIFICATE-----\n!!!not_base64!!!\n-----END CERTIFICATE-----\n", + # Non-UTF-8 bytes + b"-----BEGIN CERTIFICATE-----\n\xff\xfe\xfd\n-----END CERTIFICATE-----\n", + # Truncated block missing END CERTIFICATE + b"-----BEGIN CERTIFICATE-----\nMIIB\n", + # Truncated block missing BEGIN CERTIFICATE + b"MIIB\n-----END CERTIFICATE-----\n", + ], + ) + def test_parse_certificate_full_chain_rejects_malformed_intermediate( + self, second_cert_block + ): + chain_bytes = NON_AGENT_IDENTITY_CERT_BYTES + second_cert_block + with pytest.raises(ValueError): + _agent_identity_utils.parse_certificate(chain_bytes) def test_is_certificate_file_ready_empty_path(self): result = _agent_identity_utils._is_certificate_file_ready("") assert result is False - def test_get_agent_identity_certificate_path_empty_env(self, monkeypatch): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False + def test_is_in_well_known_dir_empty_path(self): + assert _agent_identity_utils._is_in_well_known_dir("") is False + + def test_is_in_well_known_dir_resolves_symlinks(self, tmpdir, monkeypatch): + real_run_dir = tmpdir.mkdir("run") + var_dir = tmpdir.mkdir("var") + symlink_var_run = var_dir.join("run") + os.symlink(str(real_run_dir), str(symlink_var_run)) + + well_known_via_symlink = os.path.join( + str(symlink_var_run), + "secrets", + "workload-spiffe-credentials", + "certificates.pem", ) + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + well_known_via_symlink, + ) + + resolved_cert_path = os.path.join( + str(real_run_dir), + "secrets", + "workload-spiffe-credentials", + "certificates.pem", + ) + assert _agent_identity_utils._is_in_well_known_dir(resolved_cert_path) is True + + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=False, + ) + def test_get_agent_identity_certificate_path_empty_env( + self, mock_is_ready, mock_get_config + ): result = _agent_identity_utils.get_agent_identity_certificate_path() assert result is None + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + + @mock.patch("google.auth._agent_identity_utils.time.sleep") + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + ) + def test_get_agent_identity_certificate_path_gke_bundle_fallback( + self, mock_is_ready, mock_get_config, mock_sleep + ): + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result == _mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + mock_sleep.assert_not_called() + + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + side_effect=PermissionError("Access denied"), + ) + def test_get_agent_identity_certificate_path_gke_bundle_permission_error( + self, mock_is_ready, mock_get_config + ): + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result is None + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value="/home/user/.config/gcloud/certificate_config.json", + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + ) + def test_get_agent_identity_certificate_path_no_gke_fallback_when_implicit_config_exists( + self, mock_is_ready, mock_get_config + ): + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result is None + mock_get_config.assert_called_once() + mock_is_ready.assert_not_called() + + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + ) + def test_get_agent_identity_certificate_path_no_gke_fallback_when_context_aware_env_set( + self, mock_is_ready, monkeypatch + ): + monkeypatch.setenv( + environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, + "/missing/context_aware.json", + ) + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result is None + mock_is_ready.assert_not_called() @mock.patch("google.auth._agent_identity_utils.os.path.commonpath") @mock.patch( @@ -137,8 +307,6 @@ def test_is_certificate_file_ready_os_error(self, mock_stat): @mock.patch("google.auth._agent_identity_utils.os.stat") def test_is_certificate_file_ready_not_a_file(self, mock_stat): - import stat - mock_stat.return_value = mock.MagicMock(st_mode=stat.S_IFDIR, st_size=4096) result = _agent_identity_utils._is_certificate_file_ready("/path/to/cert") assert result is False @@ -207,37 +375,73 @@ def test_calculate_certificate_fingerprint(self): assert fingerprint == expected_fingerprint + @pytest.mark.parametrize( + "primary,fallback,expected", + [ + # Default (both unset) -> not opted out + (None, None, False), + # Explicit opt-in / non-false values on primary -> not opted out + ("true", None, False), + ("TRUE", None, False), + ("1", None, False), + ("invalid", None, False), + ("", None, False), + # Explicit opt-out via primary (including surrounding whitespace) -> opted out + ("false", None, True), + ("FALSE", None, True), + (" false ", None, True), + (" FALSE\n", None, True), + # Primary overrides fallback + ("false", "true", True), + ("true", "false", False), + # Empty or whitespace-only string on primary falls back to secondary + ("", " false ", True), + (" ", " false ", True), + # Fallback when primary is unset + (None, "false", True), + (None, "true", False), + ], + ) + def test_is_bound_token_opted_out(self, monkeypatch, primary, fallback, expected): + if primary is not None: + monkeypatch.setenv( + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, + primary, + ) + if fallback is not None: + monkeypatch.setenv( + environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + fallback, + ) + assert _agent_identity_utils._is_bound_token_opted_out() is expected + @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate") def test_should_request_bound_token(self, mock_is_agent, monkeypatch): - # Agent cert, default env var (opt-in) + # Agent cert, opted in mock_is_agent.return_value = True - monkeypatch.delenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - raising=False, - ) - assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) - - # Agent cert, explicit opt-in monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, "true", ) assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) - # Agent cert, explicit opt-out + # Agent cert, opted out + mock_is_agent.reset_mock() monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, "false", ) assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) + mock_is_agent.assert_not_called() - # Non-agent cert, opt-in + # Non-agent cert, opted in mock_is_agent.return_value = False monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, "true", ) assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) + mock_is_agent.assert_called_once_with(mock.sentinel.cert) @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate") def test_should_request_bound_token_explicit_use_client_cert_false( @@ -249,6 +453,7 @@ def test_should_request_bound_token_explicit_use_client_cert_false( "false", ) assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) + mock_is_agent.assert_not_called() @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate") def test_should_request_bound_token_explicit_use_client_cert_invalid( @@ -260,6 +465,7 @@ def test_should_request_bound_token_explicit_use_client_cert_invalid( "foo", ) assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert) + mock_is_agent.assert_not_called() @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate") def test_should_request_bound_token_auto_enablement(self, mock_is_agent): @@ -381,9 +587,8 @@ def test_get_agent_identity_certificate_path_failure( _agent_identity_utils.get_agent_identity_certificate_path() assert "not found after multiple retries" in str(excinfo.value) - assert ( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES - in str(excinfo.value) + assert environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN in str( + excinfo.value ) assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS) @@ -412,9 +617,8 @@ def test_get_agent_identity_certificate_path_fail_fast_config_missing( mock_sleep.assert_not_called() @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") def test_get_agent_identity_certificate_path_fail_fast_cert_missing( - self, mock_exists, mock_sleep, tmpdir, monkeypatch + self, mock_sleep, tmpdir, monkeypatch ): # Simulate config path outside well-known dir where config is valid but cert is missing. well_known_path = tmpdir.mkdir("well_known_cert").join("certificates.pem") @@ -434,10 +638,36 @@ def test_get_agent_identity_certificate_path_fail_fast_cert_missing( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - def exists_side_effect(path): - return path == str(config_path) + result = _agent_identity_utils.get_agent_identity_certificate_path() - mock_exists.side_effect = exists_side_effect + assert result is None + mock_sleep.assert_not_called() + + @mock.patch("time.sleep") + def test_get_agent_identity_certificate_path_well_known_config_external_missing_cert_no_poll( + self, mock_sleep, tmpdir, monkeypatch + ): + well_known_dir = tmpdir.mkdir("workload-spiffe-credentials") + external_dir = tmpdir.mkdir("external_certs") + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(well_known_dir.join("certificates.pem")), + ) + config_path = well_known_dir.join("config.json") + config_path.write( + json.dumps( + { + "cert_configs": { + "workload": { + "cert_path": str(external_dir.join("missing_cert.pem")) + } + } + } + ) + ) + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) + ) result = _agent_identity_utils.get_agent_identity_certificate_path() @@ -445,9 +675,8 @@ def exists_side_effect(path): mock_sleep.assert_not_called() @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") def test_get_agent_identity_certificate_path_cert_not_found( - self, mock_exists, mock_sleep, tmpdir, monkeypatch + self, mock_sleep, tmpdir, monkeypatch ): monkeypatch.setattr( "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", @@ -462,11 +691,6 @@ def test_get_agent_identity_certificate_path_cert_not_found( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - def exists_side_effect(path): - return path == str(config_path) - - mock_exists.side_effect = exists_side_effect - with pytest.raises(exceptions.RefreshError): _agent_identity_utils.get_agent_identity_certificate_path() @@ -589,89 +813,196 @@ def test_get_agent_identity_certificate_path_permission_error_cert_file( mock_sleep.assert_not_called() @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_opted_out( - self, mock_get_path, monkeypatch + def test_get_agent_identity_certificate_and_bytes_success( + self, mock_get_path, tmpdir ): - monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - "false", + cert_file = tmpdir.join("cert.pem") + cert_file.write_binary(NON_AGENT_IDENTITY_CERT_BYTES) + mock_get_path.return_value = str(cert_file) + + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + assert isinstance(cert, x509.Certificate) + assert cert_bytes == NON_AGENT_IDENTITY_CERT_BYTES + + @pytest.mark.parametrize("key_index", [0, 1, 2]) + @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") + def test_get_agent_identity_certificate_and_bytes_combined_bundle( + self, mock_get_path, tmpdir, key_index + ): + non_utf8_bag_attrs = b"Bag Attributes\n friendlyName: \xff\xfe\n" + private_key_pem = ( + b"-----BEGIN PRIVATE KEY-----\n" + b"MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC3\n" + b"-----END PRIVATE KEY-----\n" ) - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() - assert result is None - mock_get_path.assert_not_called() + blocks = [ + AGENT_IDENTITY_CERT_BYTES.rstrip(b"\n") + b" \n", + NON_AGENT_IDENTITY_CERT_BYTES, + ] + blocks.insert(key_index, private_key_pem) + cert_file = tmpdir.join("credentialbundle.pem") + cert_file.write_binary(non_utf8_bag_attrs + b"".join(blocks)) + mock_get_path.return_value = str(cert_file) + + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + expected_certs = AGENT_IDENTITY_CERT_BYTES + NON_AGENT_IDENTITY_CERT_BYTES + assert isinstance(cert, x509.Certificate) + assert _agent_identity_utils._is_agent_identity_certificate(cert) + assert cert_bytes == expected_certs + assert cert_bytes.decode("utf-8") == expected_certs.decode("utf-8") @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_no_path( - self, mock_get_path, monkeypatch + def test_get_agent_identity_certificate_and_bytes_no_cert_blocks( + self, mock_get_path, tmpdir ): - monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - "true", + cert_file = tmpdir.join("empty_or_key_only.pem") + cert_file.write_binary( + b"-----BEGIN PRIVATE KEY-----\nMIIB\n-----END PRIVATE KEY-----\n" ) - mock_get_path.return_value = None - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() - assert result is None - mock_get_path.assert_called_once() + mock_get_path.return_value = str(cert_file) + + with pytest.warns(UserWarning, match="No PEM certificate blocks found"): + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + assert cert is None + assert cert_bytes is None - @mock.patch("google.auth._agent_identity_utils.parse_certificate") @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_success( - self, mock_get_path, mock_parse_certificate, monkeypatch + def test_get_agent_identity_certificate_and_bytes_os_error( + self, mock_get_path, tmpdir ): - monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - "true", + missing_cert_file = tmpdir.join("deleted_during_rotation.pem") + mock_get_path.return_value = str(missing_cert_file) + + with pytest.warns( + UserWarning, match="Failed to read agent identity certificate file" + ): + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + assert cert is None + assert cert_bytes is None + + @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") + def test_get_agent_identity_certificate_and_bytes_corrupt_cert_value_error( + self, mock_get_path, tmpdir + ): + cert_file = tmpdir.join("corrupt_cert.pem") + cert_file.write_binary( + b"-----BEGIN CERTIFICATE-----\nnot_valid_base64_or_der\n-----END CERTIFICATE-----\n" ) - mock_get_path.return_value = "/fake/cert.pem" - mock_open = mock.mock_open(read_data=b"cert_bytes") + mock_get_path.return_value = str(cert_file) - with mock.patch("builtins.open", mock_open): - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() + with pytest.warns( + UserWarning, match="Failed to parse agent identity certificate" + ): + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() - mock_open.assert_called_once_with("/fake/cert.pem", "rb") - mock_parse_certificate.assert_called_once_with(b"cert_bytes") - assert result == mock_parse_certificate.return_value + assert cert is None + assert cert_bytes is None + + @pytest.mark.parametrize( + "corrupt_intermediate", + [ + b"-----BEGIN CERTIFICATE-----\n" + + base64.b64encode(b"invalid_asn1_der_intermediate") + + b"\n-----END CERTIFICATE-----\n", + b"-----BEGIN CERTIFICATE-----\ntruncated_without_end_marker\n", + ], + ) + @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") + def test_get_agent_identity_certificate_and_bytes_corrupt_intermediate_warns( + self, mock_get_path, corrupt_intermediate, tmpdir + ): + cert_file = tmpdir.join("chain_with_corrupt_intermediate.pem") + cert_file.write_binary(AGENT_IDENTITY_CERT_BYTES + corrupt_intermediate) + mock_get_path.return_value = str(cert_file) + + with pytest.warns( + UserWarning, match="Failed to parse agent identity certificate" + ): + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + assert cert is None + assert cert_bytes is None @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_use_client_cert_false( + def test_get_agent_identity_certificate_and_bytes_opted_out( self, mock_get_path, monkeypatch ): monkeypatch.setenv( - environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, + environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, "false", ) - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() - assert result is None + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + assert cert is None + assert cert_bytes is None mock_get_path.assert_not_called() @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_use_client_cert_invalid( + def test_get_agent_identity_certificate_and_bytes_no_path(self, mock_get_path): + mock_get_path.return_value = None + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + assert cert is None + assert cert_bytes is None + mock_get_path.assert_called_once() + + @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") + def test_get_agent_identity_certificate_and_bytes_use_client_cert_false( self, mock_get_path, monkeypatch ): monkeypatch.setenv( environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, - "foo", + "false", ) - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() - assert result is None + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + assert cert is None + assert cert_bytes is None mock_get_path.assert_not_called() @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") - def test_get_and_parse_agent_identity_certificate_file_read_error( + def test_get_agent_identity_certificate_and_bytes_use_client_cert_invalid( self, mock_get_path, monkeypatch ): monkeypatch.setenv( - environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, - "true", + environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, + "foo", ) - mock_get_path.return_value = "/fake/cert.pem" - mock_open = mock.mock_open() - mock_open.side_effect = PermissionError("Permission denied") - - with mock.patch("builtins.open", mock_open): - result = _agent_identity_utils.get_and_parse_agent_identity_certificate() - - assert result is None + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + assert cert is None + assert cert_bytes is None + mock_get_path.assert_not_called() def test_get_cached_cert_fingerprint_no_cert(self): with pytest.raises(ValueError, match="mTLS connection is not configured."): @@ -709,3 +1040,20 @@ def test_is_agent_identity_certificate_raises_import_error(self): def test_calculate_certificate_fingerprint_raises_import_error(self): with pytest.raises(ImportError, match="The cryptography library is required"): _agent_identity_utils.calculate_certificate_fingerprint(mock.sentinel.cert) + + @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path") + def test_get_agent_identity_certificate_and_bytes_warns_without_cryptography( + self, mock_get_path, tmpdir + ): + cert_file = tmpdir.join("cert.pem") + cert_file.write_binary(AGENT_IDENTITY_CERT_BYTES) + mock_get_path.return_value = str(cert_file) + + with pytest.warns(UserWarning, match="The cryptography library is required"): + ( + cert, + cert_bytes, + ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes() + + assert cert is None + assert cert_bytes is None diff --git a/packages/google-auth/tests/test_identity_pool.py b/packages/google-auth/tests/test_identity_pool.py index 7d6a7c1b68d8..a6b59ef4552a 100644 --- a/packages/google-auth/tests/test_identity_pool.py +++ b/packages/google-auth/tests/test_identity_pool.py @@ -1005,7 +1005,7 @@ def test_retrieve_subject_token_json_file(self): @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_default( self, mock_get_workload_cert_and_key_paths @@ -1020,7 +1020,7 @@ def test_retrieve_subject_token_certificate_default( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_non_default_path( self, mock_get_workload_cert_and_key_paths @@ -1035,7 +1035,7 @@ def test_retrieve_subject_token_certificate_non_default_path( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_trust_chain_with_leaf( self, mock_get_workload_cert_and_key_paths @@ -1049,7 +1049,7 @@ def test_retrieve_subject_token_certificate_trust_chain_with_leaf( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_trust_chain_without_leaf( self, mock_get_workload_cert_and_key_paths @@ -1063,7 +1063,7 @@ def test_retrieve_subject_token_certificate_trust_chain_without_leaf( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_trust_chain_invalid_order( self, mock_get_workload_cert_and_key_paths @@ -1081,7 +1081,7 @@ def test_retrieve_subject_token_certificate_trust_chain_invalid_order( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_trust_chain_file_does_not_exist( self, mock_get_workload_cert_and_key_paths @@ -1102,7 +1102,7 @@ def test_retrieve_subject_token_certificate_trust_chain_file_does_not_exist( @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(CERT_FILE, KEY_FILE), + return_value=(CERT_FILE, KEY_FILE, None), ) def test_retrieve_subject_token_certificate_invalid_trust_chain_file( self, mock_get_workload_cert_and_key_paths @@ -1764,7 +1764,7 @@ def test_refresh_success_supplier_without_impersonation_url(self): @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=("cert", "key"), + return_value=("cert", "key", None), ) def test_get_mtls_certs(self, mock_get_workload_cert_and_key_paths): credentials = self.make_credentials( @@ -1789,7 +1789,7 @@ def test_get_mtls_certs_invalid(self): @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", - return_value=(None, None), + return_value=(None, None, None), ) def test_get_cert_bytes_none_raises_error( self, mock_get_workload_cert_and_key_paths @@ -1819,7 +1819,7 @@ def test_refresh_cert_error_raises_refresh_error(self, mock_get_cert_bytes): credentials.refresh(None) assert excinfo.match( - "Failed to retrieve certificate bytes for external account credentials" + "Failed to retrieve or parse certificate for external account credentials" ) @mock.patch.object( @@ -1835,9 +1835,30 @@ def test_refresh_os_error_raises_refresh_error(self, mock_get_cert_bytes): with pytest.raises(exceptions.RefreshError) as excinfo: credentials.refresh(None) - msg = "Failed to retrieve certificate bytes for external" + msg = "Failed to retrieve or parse certificate for external" assert excinfo.match(msg + " account credentials") + @mock.patch( + "google.auth._agent_identity_utils.parse_certificate", + side_effect=ValueError("malformed certificate chain"), + ) + @mock.patch.object( + identity_pool.Credentials, "_get_cert_bytes", return_value=b"bad_cert" + ) + def test_refresh_parse_certificate_value_error_raises_refresh_error( + self, mock_get_cert_bytes, mock_parse_certificate + ): + credentials = self.make_credentials( + credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy() + ) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.refresh(None) + + assert excinfo.match( + "Failed to retrieve or parse certificate for external account credentials" + ) + @mock.patch("google.auth._agent_identity_utils.parse_certificate") @mock.patch( "google.auth._agent_identity_utils.should_request_bound_token", diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py index b390a8dfe91b..a96a1e5a404e 100644 --- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py +++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py @@ -539,7 +539,14 @@ async def test_psc_endpoint_triggers_cert_rotation(self): await session.close() @pytest.mark.asyncio - async def test_non_mtls_url_bypasses_rotation(self): + @pytest.mark.parametrize( + "non_mtls_url", + [ + "https://pubsub.googleapis.com/test", + "https://my-service-xyz-uc.a.run.app/test", + ], + ) + async def test_non_mtls_url_bypasses_rotation(self, non_mtls_url): """Verifies that standard non-mTLS URLs bypass certificate rotation.""" mock_creds = mock.AsyncMock(spec=credentials.Credentials) mock_creds.before_request = mock.AsyncMock(return_value=None) @@ -570,7 +577,7 @@ async def test_non_mtls_url_bypasses_rotation(self): session, "configure_mtls_channel", new_callable=mock.AsyncMock ) as mock_conf, ): - resp = await session.request("GET", "https://pubsub.googleapis.com/test") + resp = await session.request("GET", non_mtls_url) assert resp == mock_resp_200 mock_check.assert_not_called() @@ -862,7 +869,14 @@ async def slow_refresh(*args, **kwargs): await session.close() @pytest.mark.asyncio - async def test_cert_rotation_with_completed_mtls_init_task(self): + @pytest.mark.parametrize( + "mtls_url", + [ + "https://pubsub.mtls.googleapis.com/test", + "https://my-service-123456.us-central1.mtls.run.app/test", + ], + ) + async def test_cert_rotation_with_completed_mtls_init_task(self, mtls_url): """ Verifies that when _mtls_init_task is already completed, receiving a 401 with rotated certificates properly resets _mtls_init_task and reconfigures mTLS. @@ -904,10 +918,8 @@ async def dummy_completed(): ) as mock_conf, ): mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp") - # Must use a hostname matching _MTLS_URL_PREFIXES (e.g. *.mtls.googleapis.com) - resp = await session.request( - "GET", "https://pubsub.mtls.googleapis.com/test" - ) + # Must use a hostname matching _mtls_helper.is_mtls_endpoint + resp = await session.request("GET", mtls_url) assert resp == mock_resp_200 mock_conf.assert_called_once() # Verify the previous completed task was cleared during rotation diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index 675f54b46324..f335aff2712d 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -56,7 +56,7 @@ def check_cert_and_key(content, expected_cert, expected_key): success = True cert_match = re.findall(_mtls_helper._CERT_REGEX, content) - success = success and len(cert_match) == 1 and cert_match[0] == expected_cert + success = success and len(cert_match) >= 1 and b"".join(cert_match) == expected_cert key_match = re.findall(_mtls_helper._KEY_REGEX, content) success = success and len(key_match) == 1 and key_match[0] == expected_key @@ -67,32 +67,40 @@ def check_cert_and_key(content, expected_cert, expected_key): class TestCertAndKeyRegex(object): def test_cert_and_key(self): # Test single cert and single key - check_cert_and_key( + assert check_cert_and_key( pytest.public_cert_bytes + pytest.private_key_bytes, pytest.public_cert_bytes, pytest.private_key_bytes, ) - check_cert_and_key( + assert check_cert_and_key( pytest.private_key_bytes + pytest.public_cert_bytes, pytest.public_cert_bytes, pytest.private_key_bytes, ) # Test cert chain and single key - check_cert_and_key( + assert check_cert_and_key( pytest.public_cert_bytes + pytest.public_cert_bytes + pytest.private_key_bytes, pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.private_key_bytes, ) - check_cert_and_key( + assert check_cert_and_key( pytest.private_key_bytes + pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.public_cert_bytes + pytest.public_cert_bytes, pytest.private_key_bytes, ) + # Test interleaved key between certificates in a combined bundle + assert check_cert_and_key( + pytest.public_cert_bytes + + pytest.private_key_bytes + + pytest.public_cert_bytes, + pytest.public_cert_bytes + pytest.public_cert_bytes, + pytest.private_key_bytes, + ) def test_key(self): # Create some fake keys for regex check. @@ -109,13 +117,13 @@ def test_key(self): /fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQAB -----END EC PRIVATE KEY-----""" - check_cert_and_key( + assert check_cert_and_key( pytest.public_cert_bytes + KEY, pytest.public_cert_bytes, KEY ) - check_cert_and_key( + assert check_cert_and_key( pytest.public_cert_bytes + RSA_KEY, pytest.public_cert_bytes, RSA_KEY ) - check_cert_and_key( + assert check_cert_and_key( pytest.public_cert_bytes + EC_KEY, pytest.public_cert_bytes, EC_KEY ) @@ -200,6 +208,21 @@ def test_success_with_cert_chain(self, mock_popen): assert key == ENCRYPTED_EC_PRIVATE_KEY assert passphrase == PASSPHRASE_VALUE + @pytest.mark.parametrize( + "trailing_bytes", + [ + b"-----BEGIN CERTIFICATE-----\nMIIB\n", + b"MIIB\n-----END CERTIFICATE-----\n", + ], + ) + @mock.patch("subprocess.Popen", autospec=True) + def test_truncated_cert_chain_raises_error(self, mock_popen, trailing_bytes): + mock_popen.return_value = self.create_mock_process( + pytest.public_cert_bytes + trailing_bytes + pytest.private_key_bytes, b"" + ) + with pytest.raises(exceptions.ClientCertError): + _mtls_helper._run_cert_provider_command(["command"]) + @mock.patch("subprocess.Popen", autospec=True) def test_missing_cert(self, mock_popen): mock_popen.return_value = self.create_mock_process( @@ -743,6 +766,37 @@ def test_no_key_file( with pytest.raises(exceptions.ClientCertError): _mtls_helper._get_workload_cert_and_key("") + @pytest.mark.parametrize( + "workload", + [ + {"cert_path": None, "key_path": "path/to/key"}, + {"cert_path": "path/to/cert", "key_path": None}, + {"cert_path": None, "key_path": None}, + {"cert_path": "", "key_path": "path/to/key"}, + {"cert_path": "path/to/cert", "key_path": ""}, + {"cert_path": 123, "key_path": "path/to/key"}, + {"cert_path": "path/to/cert", "key_path": 123}, + ], + ) + @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True + ) + def test_invalid_cert_or_key_path_value( + self, + mock_get_cert_config_path, + mock_load_json_file, + workload, + ): + mock_get_cert_config_path.return_value = "/path/to/cert" + mock_load_json_file.return_value = {"cert_configs": {"workload": workload}} + + with pytest.raises( + exceptions.ClientCertError, + match='Workload certificate configuration is missing "cert_path" or "key_path"', + ): + _mtls_helper._get_workload_cert_and_key("") + class TestReadCertAndKeyFile(object): def test_success(self): @@ -779,6 +833,40 @@ def test_invalid_key_file(self): with pytest.raises(exceptions.ClientCertError): _mtls_helper._read_cert_and_key_files(cert_path, key_path) + def test_combined_bundle_with_interleaved_key(self, tmp_path): + bundle_file = tmp_path / "credentialbundle.pem" + bundle_file.write_bytes( + pytest.public_cert_bytes.rstrip(b"\n") + + pytest.private_key_bytes + + pytest.public_cert_bytes + ) + actual_cert, actual_key = _mtls_helper._read_cert_and_key_files( + str(bundle_file), str(bundle_file) + ) + assert actual_cert == pytest.public_cert_bytes + pytest.public_cert_bytes + assert actual_key == pytest.private_key_bytes + + def test_multiple_keys_raises_error(self, tmp_path): + cert_path = os.path.join(pytest.data_dir, "public_cert.pem") + key_file = tmp_path / "two_keys.pem" + key_file.write_bytes(pytest.private_key_bytes + pytest.private_key_bytes) + with pytest.raises(exceptions.ClientCertError): + _mtls_helper._read_cert_and_key_files(cert_path, str(key_file)) + + @pytest.mark.parametrize( + "trailing_bytes", + [ + b"-----BEGIN CERTIFICATE-----\nMIIB\n", + b"MIIB\n-----END CERTIFICATE-----\n", + ], + ) + def test_truncated_multi_cert_raises_error(self, tmp_path, trailing_bytes): + cert_file = tmp_path / "truncated_chain.pem" + cert_file.write_bytes(pytest.public_cert_bytes + trailing_bytes) + key_path = os.path.join(pytest.data_dir, "privatekey.pem") + with pytest.raises(exceptions.ClientCertError): + _mtls_helper._read_cert_and_key_files(str(cert_file), key_path) + class TestGetCertConfigPath(object): def test_success_with_override(self): @@ -1223,6 +1311,21 @@ def test_check_parameters_for_unauthorized_response_without_cached_cert( mock_call_client_cert_callback.assert_called_once() mock_agent_identity_utils.get_cached_cert_fingerprint.assert_not_called() + @mock.patch("google.auth.transport._mtls_helper.call_client_cert_callback") + @mock.patch("google.auth.transport._mtls_helper._agent_identity_utils") + def test_check_parameters_for_unauthorized_response_no_call_cert( + self, mock_agent_identity_utils, mock_call_client_cert_callback + ): + mock_call_client_cert_callback.return_value = (None, None) + + result = _mtls_helper.check_parameters_for_unauthorized_response( + cached_cert=CERT_MOCK_VAL + ) + + assert result == (None, None, None, None) + mock_call_client_cert_callback.assert_called_once() + mock_agent_identity_utils.parse_certificate.assert_not_called() + @mock.patch("google.auth.transport._mtls_helper.get_client_ssl_credentials") def test_call_client_cert_callback(self, mock_get_client_ssl_credentials): mock_get_client_ssl_credentials.return_value = ( @@ -1913,6 +2016,12 @@ class TestIsMtlsEndpoint(object): "https://p.googleapis.com/", "https://p.googleapis.com:443/v1", "https://p.googleapis.com.", + "https://mtls.run.app", + "https://mtls.run.app/", + "https://my-service-123456.us-central1.mtls.run.app", + "https://my-service-123456.us-central1.mtls.run.app/v1/invocations", + "https://tag---my-service-123456.us-central1.mtls.run.app.", + b"https://my-service-123456.us-central1.mtls.run.app", ], ) def test_is_mtls_endpoint_true(self, url): @@ -1926,6 +2035,11 @@ def test_is_mtls_endpoint_true(self, url): "https://storage.googleapis.com:443/b/my-bucket", "https://storage.googleapis.com:443/bucket/mtls.googleapis.com?pageSize=10#frag", "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://my-service-xyz-uc.a.run.app", + "https://my-service-123456.us-central1.run.app", + "https://my-service-xyz-uc.a.run.app/mtls.run.app", + "https://fake-mtls.run.app/v1", + "https://fake-mtls.run.app.attacker.com/v1", "https://[2001:db8::1]:443/mtls.googleapis.com", "https://[::1]:8443/mtls.googleapis.com", "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com", @@ -1944,6 +2058,7 @@ def test_is_mtls_endpoint_true(self, url): "https://storage.googleapis.com/bucket/mtls.googleapis.com" ), "https://.", + "https://[::1", "", None, 123, @@ -1952,3 +2067,215 @@ def test_is_mtls_endpoint_true(self, url): ) def test_is_mtls_endpoint_false(self, url): assert _mtls_helper.is_mtls_endpoint(url) is False + + +class TestReadCredentialBundleFile(object): + @pytest.mark.parametrize("key_index", [0, 1]) + def test_single_cert_and_key(self, tmpdir, key_index): + bundle_file = tmpdir.join("x509.credential-bundle.private-key.pem") + blocks = [pytest.public_cert_bytes] + blocks.insert(key_index, pytest.private_key_bytes) + bundle_file.write_binary(b"".join(blocks)) + + with mock.patch("builtins.open", wraps=open) as mock_open: + cert_bytes, key_bytes = _mtls_helper._read_credential_bundle_file( + str(bundle_file) + ) + mock_open.assert_called_once_with(str(bundle_file), "rb") + assert cert_bytes == pytest.public_cert_bytes + assert key_bytes == pytest.private_key_bytes + + @pytest.mark.parametrize("key_index", [0, 1, 2]) + def test_multi_cert_chain_and_key(self, tmpdir, key_index): + bundle_file = tmpdir.join("x509.credential-bundle.private-key.pem") + blocks = [ + pytest.public_cert_bytes.rstrip(b"\n") + b" \n", + pytest.public_cert_bytes, + ] + blocks.insert(key_index, pytest.private_key_bytes) + bundle_file.write_binary(b"".join(blocks)) + + cert_bytes, key_bytes = _mtls_helper._read_credential_bundle_file( + str(bundle_file) + ) + assert cert_bytes == pytest.public_cert_bytes + pytest.public_cert_bytes + assert key_bytes == pytest.private_key_bytes + + def test_missing_cert_raises_error(self, tmpdir): + bundle_file = tmpdir.join("key_only.pem") + bundle_file.write_binary(pytest.private_key_bytes) + + with pytest.raises(exceptions.ClientCertError, match="at least one PEM"): + _mtls_helper._read_credential_bundle_file(str(bundle_file)) + + def test_missing_key_raises_error(self, tmpdir): + bundle_file = tmpdir.join("cert_only.pem") + bundle_file.write_binary(pytest.public_cert_bytes) + + with pytest.raises( + exceptions.ClientCertError, match="a single PEM formatted private key" + ): + _mtls_helper._read_credential_bundle_file(str(bundle_file)) + + def test_multiple_keys_raises_error(self, tmpdir): + bundle_file = tmpdir.join("multi_key.pem") + bundle_file.write_binary( + pytest.public_cert_bytes + + pytest.private_key_bytes + + pytest.private_key_bytes + ) + + with pytest.raises( + exceptions.ClientCertError, match="a single PEM formatted private key" + ): + _mtls_helper._read_credential_bundle_file(str(bundle_file)) + + def test_os_error_propagates(self, tmpdir): + missing_bundle = str(tmpdir.join("nonexistent_rotated_symlink.pem")) + with pytest.raises(OSError): + _mtls_helper._read_credential_bundle_file(missing_bundle) + + +class TestGkeCredentialBundleDiscovery(object): + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + autospec=True, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + @mock.patch( + "google.auth.transport._mtls_helper._read_credential_bundle_file", + autospec=True, + ) + def test_get_workload_cert_and_key_gke_fallback( + self, mock_read_bundle, mock_is_ready, mock_get_config_path + ): + mock_read_bundle.return_value = ( + pytest.public_cert_bytes, + pytest.private_key_bytes, + ) + + cert, key = _mtls_helper._get_workload_cert_and_key() + assert cert == pytest.public_cert_bytes + assert key == pytest.private_key_bytes + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + mock_read_bundle.assert_called_once_with( + _mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH + ) + + @mock.patch( + "google.auth.transport._mtls_helper.path.exists", + return_value=False, + autospec=True, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_get_workload_cert_and_key_no_gke_fallback_when_explicit_path_set( + self, mock_is_ready, mock_exists + ): + cert, key = _mtls_helper._get_workload_cert_and_key("/missing/config.json") + assert cert is None + assert key is None + mock_exists.assert_called_once_with("/missing/config.json") + mock_is_ready.assert_not_called() + + @mock.patch( + "google.auth.transport._mtls_helper.path.exists", + return_value=False, + autospec=True, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_get_workload_cert_and_key_no_gke_fallback_when_explicit_env_set( + self, mock_is_ready, mock_exists, monkeypatch + ): + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, "/missing/config.json" + ) + cert, key = _mtls_helper._get_workload_cert_and_key() + assert cert is None + assert key is None + mock_exists.assert_called_once_with("/missing/config.json") + mock_is_ready.assert_not_called() + + @pytest.mark.parametrize("include_context_aware", [True, False]) + @mock.patch( + "google.auth.transport._mtls_helper.path.exists", + return_value=False, + autospec=True, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_get_workload_cert_and_key_no_gke_fallback_when_context_aware_env_set( + self, mock_is_ready, mock_exists, include_context_aware, monkeypatch + ): + monkeypatch.setenv( + environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, + "/missing/context_aware_config.json", + ) + cert, key = _mtls_helper._get_workload_cert_and_key( + include_context_aware=include_context_aware + ) + assert cert is None + assert key is None + assert mock_exists.call_count == 1 + mock_is_ready.assert_not_called() + + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + autospec=True, + ) + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_check_use_client_cert_gke_auto_enabled( + self, mock_is_ready, mock_get_config_path + ): + assert _mtls_helper.check_use_client_cert() is True + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_check_use_client_cert_gke_disabled_by_env( + self, mock_is_ready, monkeypatch + ): + monkeypatch.setenv(environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false") + assert _mtls_helper.check_use_client_cert() is False + mock_is_ready.assert_not_called() + + @mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, + ) + def test_no_gke_fallback_when_implicit_config_has_no_workload( + self, mock_is_ready, tmpdir, monkeypatch + ): + config_file = tmpdir.join("certificate_config.json") + config_file.write('{"cert_configs": {}}') + monkeypatch.setattr( + _mtls_helper._cloud_sdk, "get_config_path", lambda: str(tmpdir) + ) + + assert _mtls_helper._get_workload_cert_and_key() == (None, None) + assert _mtls_helper.check_use_client_cert() is False + mock_is_ready.assert_not_called() diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py index 7caf63861751..5899a1b1c73b 100644 --- a/packages/google-auth/tests/transport/test_mtls.py +++ b/packages/google-auth/tests/transport/test_mtls.py @@ -46,7 +46,7 @@ def side_effect(path): # Assert assert result is True - mock_get_cert.assert_called_once_with(include_context_aware=True) + mock_get_cert.assert_called_once_with(None, True) mock_check.assert_any_call(_mtls_helper.CONTEXT_AWARE_METADATA_PATH) @@ -63,13 +63,21 @@ def test_has_default_client_cert_source_without_context_aware( result = mtls.has_default_client_cert_source(include_context_aware=False) assert result is False - mock_get_cert.assert_called_once_with(include_context_aware=False) + mock_get_cert.assert_called_once_with(None, False) mock_check.assert_not_called() +@mock.patch( + "google.auth.transport._mtls_helper._load_json_file", + return_value={ + "cert_configs": {"workload": {"cert_path": "cert.pem", "key_path": "key.pem"}} + }, +) @mock.patch("google.auth.transport._mtls_helper._check_config_path") @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") -def test_has_default_client_cert_source_falls_back(mock_get_cert, mock_check): +def test_has_default_client_cert_source_falls_back( + mock_get_cert, mock_check, mock_load_json +): """ Tests that it checks X.509 WIF first, and if found, returns True without checking context aware metadata. """ @@ -83,21 +91,28 @@ def test_has_default_client_cert_source_falls_back(mock_get_cert, mock_check): # Assert assert result is True # Verify the sequence of calls - mock_get_cert.assert_called_once_with(include_context_aware=True) + mock_get_cert.assert_called_once_with(None, True) mock_check.assert_not_called() +@mock.patch( + "google.auth.transport._mtls_helper._load_json_file", + return_value={ + "cert_configs": {"workload": {"cert_path": "cert.pem", "key_path": "key.pem"}} + }, + autospec=True, +) @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) def test_has_default_client_cert_source_env_var_success( - check_config_path, get_cert_config_path + check_config_path, get_cert_config_path, mock_load_json ): check_config_path.return_value = None get_cert_config_path.return_value = "/absolute/path/to/cert.json" assert mtls.has_default_client_cert_source(True) - get_cert_config_path.assert_called_with(include_context_aware=True) + get_cert_config_path.assert_called_with(None, True) @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path", autospec=True) @@ -476,3 +491,70 @@ def test_get_default_ssl_context_no_default_source(mock_has_default, mock_should result = mtls.get_default_ssl_context() assert result is None + + +@mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value=None, + autospec=True, +) +@mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, +) +def test_has_default_client_cert_source_gke_bundle(mock_is_ready, mock_get_cert): + assert mtls.has_default_client_cert_source(include_context_aware=False) is True + mock_is_ready.assert_called_once_with(_mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH) + + +@mock.patch( + "google.auth.transport._mtls_helper._load_json_file", + return_value={"cert_configs": {}}, + autospec=True, +) +@mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value="/home/user/.config/gcloud/certificate_config.json", + autospec=True, +) +@mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, +) +def test_has_default_client_cert_source_config_without_workload_no_gke_fallback( + mock_is_ready, mock_get_cert, mock_load_json +): + assert mtls.has_default_client_cert_source(include_context_aware=False) is False + mock_is_ready.assert_not_called() + + +@pytest.mark.parametrize( + "side_effect", + [ + exceptions.ClientCertError("Invalid JSON"), + OSError("Permission denied"), + ], +) +@mock.patch( + "google.auth.transport._mtls_helper._load_json_file", + autospec=True, +) +@mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value="/home/user/.config/gcloud/certificate_config.json", + autospec=True, +) +@mock.patch( + "google.auth._agent_identity_utils._is_certificate_file_ready", + return_value=True, + autospec=True, +) +def test_has_default_client_cert_source_malformed_config_no_gke_fallback( + mock_is_ready, mock_get_cert, mock_load_json, side_effect +): + mock_load_json.side_effect = side_effect + + assert mtls.has_default_client_cert_source(include_context_aware=False) is True + mock_is_ready.assert_not_called() diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index ae102bcb33e5..5ca8788ca42b 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -1280,7 +1280,14 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): # Assert mTLS check logic was SKIPPED (Inner Check was False) assert not mock_helper.check_parameters_for_unauthorized_response.called - def test_cert_rotation_skipped_on_non_mtls_url(self): + @pytest.mark.parametrize( + "non_mtls_url", + [ + "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://my-service-xyz-uc.a.run.app/mtls.run.app", + ], + ) + def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): """ Tests that mTLS cert rotation is skipped on non-mTLS URLs even if mTLS is enabled and an UNAUTHORIZED (401) response is received. @@ -1293,7 +1300,6 @@ def test_cert_rotation_skipped_on_non_mtls_url(self): make_response(status=http_client.OK), ] ) - non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com" authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) @@ -1313,10 +1319,18 @@ def test_cert_rotation_skipped_on_non_mtls_url(self): # Assert mTLS check logic was SKIPPED assert not mock_check_params.called - def test_cert_rotation_triggered_on_psc_url(self): + @pytest.mark.parametrize( + "mtls_url", + [ + "https://storage.p.googleapis.com/b/my-bucket", + "https://my-service-123456.us-central1.mtls.run.app/v1", + ], + ) + def test_cert_rotation_triggered_on_mtls_url(self, mtls_url): """ - Tests that mTLS cert rotation IS triggered on a Private Service Connect - (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received. + Tests that mTLS cert rotation IS triggered on Private Service Connect + (PSC) and Cloud Run mTLS endpoints when an UNAUTHORIZED (401) response + is received. """ credentials = mock.Mock(wraps=CredentialsStub()) adapter = AdapterStub( @@ -1325,11 +1339,10 @@ def test_cert_rotation_triggered_on_psc_url(self): make_response(status=http_client.OK), ] ) - psc_url = "https://storage.p.googleapis.com/b/my-bucket" authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 ) - authed_session.mount(psc_url, adapter) + authed_session.mount(mtls_url, adapter) authed_session._is_mtls = True authed_session._cached_cert = b"cached_cert" @@ -1338,9 +1351,9 @@ def test_cert_rotation_triggered_on_psc_url(self): "check_parameters_for_unauthorized_response", return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"), ) as mock_check_params: - authed_session.request("GET", psc_url) + authed_session.request("GET", mtls_url) - # Assert mTLS check logic was called on PSC endpoint + # Assert mTLS check logic was called on PSC / Cloud Run mTLS endpoint mock_check_params.assert_called_once() assert credentials.refresh.called diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index b9136a4f9f7c..7ffd50aeee32 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -657,7 +657,14 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): # Assert mTLS check logic was SKIPPED (Inner Check was False) assert not mock_helper.check_parameters_for_unauthorized_response.called - def test_cert_rotation_skipped_on_non_mtls_url(self): + @pytest.mark.parametrize( + "non_mtls_url", + [ + "https://storage.googleapis.com/bucket/mtls.googleapis.com", + "https://my-service-xyz-uc.a.run.app/mtls.run.app", + ], + ) + def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url): """ Tests that mTLS cert rotation is skipped on non-mTLS URLs even if mTLS is enabled and an UNAUTHORIZED (401) response is received. @@ -669,7 +676,6 @@ def test_cert_rotation_skipped_on_non_mtls_url(self): ResponseStub(status=http_client.OK), ] ) - non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com" authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http ) @@ -688,10 +694,18 @@ def test_cert_rotation_skipped_on_non_mtls_url(self): # Assert mTLS check logic was SKIPPED assert not mock_check_params.called - def test_cert_rotation_triggered_on_psc_url(self): + @pytest.mark.parametrize( + "mtls_url", + [ + "https://storage.p.googleapis.com/b/my-bucket", + "https://my-service-123456.us-central1.mtls.run.app/v1", + ], + ) + def test_cert_rotation_triggered_on_mtls_url(self, mtls_url): """ - Tests that mTLS cert rotation IS triggered on a Private Service Connect - (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received. + Tests that mTLS cert rotation IS triggered on Private Service Connect + (PSC) and Cloud Run mTLS endpoints when an UNAUTHORIZED (401) response + is received. """ credentials = mock.Mock(wraps=CredentialsStub()) http = HttpStub( @@ -700,7 +714,6 @@ def test_cert_rotation_triggered_on_psc_url(self): ResponseStub(status=http_client.OK), ] ) - psc_url = "https://storage.p.googleapis.com/b/my-bucket" authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http ) @@ -712,9 +725,9 @@ def test_cert_rotation_triggered_on_psc_url(self): "check_parameters_for_unauthorized_response", return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"), ) as mock_check_params: - authed_http.urlopen("GET", psc_url) + authed_http.urlopen("GET", mtls_url) - # Assert mTLS check logic was called on PSC endpoint + # Assert mTLS check logic was called on PSC / Cloud Run mTLS endpoint mock_check_params.assert_called_once() assert credentials.refresh.called diff --git a/packages/google-auth/tests_async/oauth2/test_id_token.py b/packages/google-auth/tests_async/oauth2/test_id_token.py index 22456c6bdace..883a9ee9c5ec 100644 --- a/packages/google-auth/tests_async/oauth2/test_id_token.py +++ b/packages/google-auth/tests_async/oauth2/test_id_token.py @@ -209,11 +209,25 @@ async def test_verify_firebase_token_clock_skew(verify_token): @pytest.mark.asyncio -async def test_fetch_id_token_from_metadata_server(monkeypatch): +@pytest.mark.parametrize( + "call_kwargs,expected_bind_id_token", + [ + ({}, None), + ({"bind_id_token": None}, None), + ({"bind_id_token": True}, True), + ({"bind_id_token": False}, False), + ], +) +async def test_fetch_id_token_from_metadata_server( + monkeypatch, call_kwargs, expected_bind_id_token +): monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False) - def mock_init(self, request, audience, use_metadata_identity_endpoint): + def mock_init( + self, request, audience, use_metadata_identity_endpoint, bind_id_token + ): assert use_metadata_identity_endpoint + assert bind_id_token is expected_bind_id_token self.token = "id_token" with mock.patch("google.auth.compute_engine._metadata.ping", return_value=True): @@ -224,13 +238,16 @@ def mock_init(self, request, audience, use_metadata_identity_endpoint): ): request = mock.AsyncMock() token = await id_token.fetch_id_token( - request, "https://pubsub.googleapis.com" + request, + "https://pubsub.googleapis.com", + **call_kwargs, ) assert token == "id_token" @pytest.mark.asyncio -async def test_fetch_id_token_from_explicit_cred_json_file(monkeypatch): +@pytest.mark.parametrize("bind_id_token", [None, True, False]) +async def test_fetch_id_token_from_explicit_cred_json_file(monkeypatch, bind_id_token): monkeypatch.setenv(environment_vars.CREDENTIALS, test_id_token.SERVICE_ACCOUNT_FILE) async def mock_refresh(self, request): @@ -240,7 +257,9 @@ async def mock_refresh(self, request): _service_account_async.IDTokenCredentials, "refresh", mock_refresh ): request = mock.AsyncMock() - token = await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") + token = await id_token.fetch_id_token( + request, "https://pubsub.googleapis.com", bind_id_token=bind_id_token + ) assert token == "id_token"