diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 9497368070dd..fc5382a04565 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -51,30 +51,10 @@ _LOGGER = logging.getLogger(__name__) -# A flag to track if we have already logged a warning about mTLS auto-enablement failures. -# This prevents log spam when client libraries create transports or session instances -# frequently within a single process. -_has_logged_mtls_warning = False - - _PASSPHRASE_REGEX = re.compile( b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL ) -# Temporary patch to accomodate incorrect cert config in Cloud Run prod environment. -_WELL_KNOWN_CLOUD_RUN_CERT_PATH = ( - "/var/run/secrets/workload-spiffe-credentials/certificates.pem" -) -_WELL_KNOWN_CLOUD_RUN_KEY_PATH = ( - "/var/run/secrets/workload-spiffe-credentials/private_key.pem" -) -_INCORRECT_CLOUD_RUN_CERT_PATH = ( - "/var/lib/volumes/certificate/workload-certificates/certificates.pem" -) -_INCORRECT_CLOUD_RUN_KEY_PATH = ( - "/var/lib/volumes/certificate/workload-certificates/private_key.pem" -) - class _MemfdCreationError(OSError): """Raised when Linux in-memory virtual file creation (memfd) fails.""" @@ -436,10 +416,15 @@ def _get_cert_config_path(certificate_config_path=None, include_context_aware=Tr The absolute path of the certificate config file, and None if the file does not exist. """ + source = "function argument" + is_explicit = True if certificate_config_path is None: env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None) if env_path is not None and env_path != "": certificate_config_path = env_path + source = ( + f"environment variable {environment_vars.GOOGLE_API_CERTIFICATE_CONFIG}" + ) else: env_path = environ.get( environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, @@ -447,11 +432,19 @@ def _get_cert_config_path(certificate_config_path=None, include_context_aware=Tr ) if include_context_aware and env_path is not None and env_path != "": certificate_config_path = env_path + source = f"environment variable {environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH}" else: certificate_config_path = CERTIFICATE_CONFIGURATION_DEFAULT_PATH + is_explicit = False certificate_config_path = path.expanduser(certificate_config_path) if not path.exists(certificate_config_path): + if is_explicit: + _LOGGER.debug( + "Certificate configuration file explicitly specified via %s at %s does not exist", + source, + certificate_config_path, + ) return None return certificate_config_path @@ -489,25 +482,6 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True): cert_path = workload["cert_path"] key_path = workload["key_path"] - # == BEGIN Temporary Cloud Run PATCH == - # See https://github.com/googleapis/google-auth-library-python/issues/1881 - if (cert_path == _INCORRECT_CLOUD_RUN_CERT_PATH) and ( - key_path == _INCORRECT_CLOUD_RUN_KEY_PATH - ): - if not path.exists(cert_path) and not path.exists(key_path): - _LOGGER.debug( - "Applying Cloud Run certificate path patch. " - "Configured paths not found: %s, %s. " - "Using well-known paths: %s, %s", - cert_path, - key_path, - _WELL_KNOWN_CLOUD_RUN_CERT_PATH, - _WELL_KNOWN_CLOUD_RUN_KEY_PATH, - ) - cert_path = _WELL_KNOWN_CLOUD_RUN_CERT_PATH - key_path = _WELL_KNOWN_CLOUD_RUN_KEY_PATH - # == END Temporary Cloud Run PATCH == - return cert_path, key_path @@ -755,36 +729,33 @@ def check_use_client_cert(): will default to False. If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred as True (auto-enabled) if a workload config file exists (pointed at by - GOOGLE_API_CERTIFICATE_CONFIG) containing a "workload" section. + 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. Otherwise, it returns False. Returns: bool: Whether the client certificate should be used for mTLS connection. """ - global _has_logged_mtls_warning env_override = _check_use_client_cert_env() if env_override is not None: return env_override # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set) - # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set. - cert_path = getenv(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) or getenv( - environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH - ) + # Check if a workload config file exists. + cert_path = _get_cert_config_path(include_context_aware=True) if cert_path: try: with open(cert_path, "r") as f: content = json.load(f) except (FileNotFoundError, OSError, json.JSONDecodeError) as e: - if not _has_logged_mtls_warning: - _LOGGER.warning( - "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s", - cert_path, - e, - ) - _has_logged_mtls_warning = True + _LOGGER.debug( + "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s", + cert_path, + e, + ) return False # Structural validation @@ -794,12 +765,10 @@ def check_use_client_cert(): return True # If we got here, the file exists but the expected structure is missing - if not _has_logged_mtls_warning: - _LOGGER.warning( - "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.", - cert_path, - ) - _has_logged_mtls_warning = True + _LOGGER.debug( + "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.", + cert_path, + ) return False diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py index 96b84e8e587f..bc92b4295eaf 100644 --- a/packages/google-auth/google/auth/transport/mtls.py +++ b/packages/google-auth/google/auth/transport/mtls.py @@ -24,7 +24,6 @@ from google.auth import exceptions from google.auth.transport import _mtls_helper - _LOGGER = logging.getLogger(__name__) @@ -44,25 +43,18 @@ 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: + return True if ( include_context_aware and _mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH) is not None ): return True - if ( - _mtls_helper._check_config_path( - _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH - ) - is not None - ): - return True - cert_config_path = getenv("GOOGLE_API_CERTIFICATE_CONFIG") - if ( - cert_config_path - and _mtls_helper._check_config_path(cert_config_path) is not None - ): - return True + return False @@ -146,7 +138,9 @@ def should_use_client_cert(): If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding bool value will be returned If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred by - reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG, and verifying it + reading a file 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, and verifying it contains a "workload" section. If so, the function will return True, otherwise False. diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index e7b80e92d0ab..259f97f4af76 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -315,13 +315,16 @@ def test_success_with_context_aware_metadata( ) @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) + @mock.patch("os.path.exists", autospec=True) def test_success_with_certificate_config( self, + mock_path_exists, mock_check_config_path, mock_load_json_file, mock_get_cert_config_path, mock_read_cert_and_key_files, ): + mock_path_exists.return_value = True cert_config_path = "/path/to/config" mock_check_config_path.return_value = cert_config_path mock_load_json_file.return_value = { @@ -341,93 +344,6 @@ def test_success_with_certificate_config( assert key == pytest.private_key_bytes assert passphrase is None - @mock.patch( - "google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True - ) - @mock.patch( - "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True - ) - @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) - @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) - def test_success_with_certificate_config_cloud_run_patch( - self, - mock_check_config_path, - mock_load_json_file, - mock_get_cert_config_path, - mock_read_cert_and_key_files, - ): - cert_config_path = "/path/to/config" - mock_check_config_path.return_value = cert_config_path - mock_load_json_file.return_value = { - "cert_configs": { - "workload": { - "cert_path": _mtls_helper._INCORRECT_CLOUD_RUN_CERT_PATH, - "key_path": _mtls_helper._INCORRECT_CLOUD_RUN_KEY_PATH, - } - } - } - mock_get_cert_config_path.return_value = cert_config_path - mock_read_cert_and_key_files.return_value = ( - pytest.public_cert_bytes, - pytest.private_key_bytes, - ) - - has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials() - assert has_cert - assert cert == pytest.public_cert_bytes - assert key == pytest.private_key_bytes - assert passphrase is None - - mock_read_cert_and_key_files.assert_called_once_with( - _mtls_helper._WELL_KNOWN_CLOUD_RUN_CERT_PATH, - _mtls_helper._WELL_KNOWN_CLOUD_RUN_KEY_PATH, - ) - - @mock.patch("os.path.exists", autospec=True) - @mock.patch( - "google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True - ) - @mock.patch( - "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True - ) - @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) - @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) - def test_success_with_certificate_config_cloud_run_patch_skipped_if_cert_exists( - self, - mock_check_config_path, - mock_load_json_file, - mock_get_cert_config_path, - mock_read_cert_and_key_files, - mock_os_path_exists, - ): - cert_config_path = "/path/to/config" - mock_check_config_path.return_value = cert_config_path - mock_os_path_exists.return_value = True - mock_load_json_file.return_value = { - "cert_configs": { - "workload": { - "cert_path": _mtls_helper._INCORRECT_CLOUD_RUN_CERT_PATH, - "key_path": _mtls_helper._INCORRECT_CLOUD_RUN_KEY_PATH, - } - } - } - mock_get_cert_config_path.return_value = cert_config_path - mock_read_cert_and_key_files.return_value = ( - pytest.public_cert_bytes, - pytest.private_key_bytes, - ) - - has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials() - assert has_cert - assert cert == pytest.public_cert_bytes - assert key == pytest.private_key_bytes - assert passphrase is None - - mock_read_cert_and_key_files.assert_called_once_with( - _mtls_helper._INCORRECT_CLOUD_RUN_CERT_PATH, - _mtls_helper._INCORRECT_CLOUD_RUN_KEY_PATH, - ) - @mock.patch( "google.auth.transport._mtls_helper._get_workload_cert_and_key", autospec=True ) @@ -531,12 +447,15 @@ class TestGetWorkloadCertAndKey(object): @mock.patch( "google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True ) + @mock.patch("os.path.exists", autospec=True) def test_success( self, + mock_path_exists, mock_read_cert_and_key_files, mock_get_cert_config_path, mock_load_json_file, ): + mock_path_exists.return_value = True cert_config_path = "/path/to/cert" mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { @@ -569,7 +488,11 @@ def test_file_not_found_returns_none(self, mock_get_cert_config_path): @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) - def test_no_cert_configs(self, mock_get_cert_config_path, mock_load_json_file): + @mock.patch("os.path.exists", autospec=True) + def test_no_cert_configs( + self, mock_path_exists, mock_get_cert_config_path, mock_load_json_file + ): + mock_path_exists.return_value = True mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = {} @@ -592,7 +515,11 @@ def test_no_workload(self, mock_get_cert_config_path, mock_load_json_file): @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) - def test_no_cert_file(self, mock_get_cert_config_path, mock_load_json_file): + @mock.patch("os.path.exists", autospec=True) + def test_no_cert_file( + self, mock_path_exists, mock_get_cert_config_path, mock_load_json_file + ): + mock_path_exists.return_value = True mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { "cert_configs": {"workload": {"key_path": "path/to/key"}} @@ -605,7 +532,11 @@ def test_no_cert_file(self, mock_get_cert_config_path, mock_load_json_file): @mock.patch( "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True ) - def test_no_key_file(self, mock_get_cert_config_path, mock_load_json_file): + @mock.patch("os.path.exists", autospec=True) + def test_no_key_file( + self, mock_path_exists, mock_get_cert_config_path, mock_load_json_file + ): + mock_path_exists.return_value = True mock_get_cert_config_path.return_value = "/path/to/cert" mock_load_json_file.return_value = { "cert_configs": {"workload": {"cert_path": "path/to/key"}} @@ -657,10 +588,16 @@ def test_success_with_override(self): returned_path = _mtls_helper._get_cert_config_path(config_path) assert returned_path == config_path - def test_override_does_not_exist(self): + @mock.patch("google.auth.transport._mtls_helper._LOGGER.debug") + def test_override_does_not_exist(self, mock_debug): config_path = "fake/file/path" returned_path = _mtls_helper._get_cert_config_path(config_path) assert returned_path is None + mock_debug.assert_called_once_with( + "Certificate configuration file explicitly specified via %s at %s does not exist", + "function argument", + "fake/file/path", + ) @mock.patch.dict( os.environ, @@ -688,9 +625,15 @@ def test_env_variable(self, mock_path_exists): expected_path = "path/to/config/file" assert returned_path == expected_path - @mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}) + @mock.patch.dict( + os.environ, + { + "GOOGLE_API_CERTIFICATE_CONFIG": "", + "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", + }, + ) @mock.patch("os.path.exists", autospec=True) - def test_env_variable_file_does_not_exist(self, mock_path_exists): + def test_default_file_does_not_exist(self, mock_path_exists): mock_path_exists.return_value = False returned_path = _mtls_helper._get_cert_config_path() assert returned_path is None @@ -730,11 +673,27 @@ def test_cert_config_path_fallback(self): os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": "path/to/config/file"} ) @mock.patch("os.path.exists", autospec=True) - def test_default_file_does_not_exist(self, mock_path_exists): + def test_env_variable_file_does_not_exist(self, mock_path_exists): mock_path_exists.return_value = False returned_path = _mtls_helper._get_cert_config_path() assert returned_path is None + @mock.patch.dict( + os.environ, + { + "GOOGLE_API_CERTIFICATE_CONFIG": "", + "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "path/to/context/aware/config", + }, + ) + @mock.patch("os.path.exists", autospec=True) + def test_cert_config_path_ignore_context_aware(self, mock_path_exists): + mock_path_exists.return_value = True + returned_path = _mtls_helper._get_cert_config_path(include_context_aware=False) + expected_path = os.path.expanduser( + _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH + ) + assert returned_path == expected_path + class TestGetClientCertAndKey(object): def test_callback_success(self): @@ -848,7 +807,9 @@ def test_env_var_explicit_garbage(self): "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", }, ) - def test_config_file_success(self, mock_file): + @mock.patch("os.path.exists", autospec=True) + def test_config_file_success(self, mock_exists, mock_file): + mock_exists.return_value = True # We manually apply mock_open here so we can keep autospec=True on the decorator mock_file.side_effect = mock.mock_open( read_data='{"cert_configs": {"workload": "exists"}}' @@ -865,7 +826,9 @@ def test_config_file_success(self, mock_file): "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", }, ) - def test_config_file_missing_keys(self, mock_file): + @mock.patch("os.path.exists", autospec=True) + def test_config_file_missing_keys(self, mock_exists, mock_file): + mock_exists.return_value = True mock_file.side_effect = mock.mock_open(read_data='{"cert_configs": {}}') assert _mtls_helper.check_use_client_cert() is False @@ -879,7 +842,9 @@ def test_config_file_missing_keys(self, mock_file): "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", }, ) - def test_config_file_bad_json(self, mock_file): + @mock.patch("os.path.exists", autospec=True) + def test_config_file_bad_json(self, mock_exists, mock_file): + mock_exists.return_value = True mock_file.side_effect = mock.mock_open(read_data="{bad_json") assert _mtls_helper.check_use_client_cert() is False @@ -893,12 +858,34 @@ def test_config_file_bad_json(self, mock_file): "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", }, ) - def test_config_file_not_found(self, mock_file): + @mock.patch("os.path.exists", autospec=True) + def test_config_file_not_found(self, mock_exists, mock_file): + mock_exists.return_value = True mock_file.side_effect = FileNotFoundError assert _mtls_helper.check_use_client_cert() is False + @mock.patch("builtins.open", autospec=True) + @mock.patch.dict( + os.environ, + { + "GOOGLE_API_USE_CLIENT_CERTIFICATE": "", + "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "", + "GOOGLE_API_CERTIFICATE_CONFIG": "/path/to/config", + "CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH": "", + }, + ) + @mock.patch("os.path.exists", autospec=True) + def test_config_file_invalid_json_type(self, mock_exists, mock_file): + mock_exists.return_value = True + mock_file.side_effect = mock.mock_open(read_data="[]") + assert _mtls_helper.check_use_client_cert() is False + + @mock.patch("builtins.open", autospec=True) @mock.patch.dict(os.environ, {}, clear=True) - def test_no_env_vars_set(self): + @mock.patch("os.path.exists", autospec=True) + def test_no_env_vars_set(self, mock_exists, mock_open): + mock_exists.return_value = False + mock_open.side_effect = FileNotFoundError() assert _mtls_helper.check_use_client_cert() is False def test_use_client_cert_precedence(self): @@ -941,7 +928,9 @@ def test_use_client_cert_fallback(self): assert _mtls_helper.check_use_client_cert() is False @mock.patch("builtins.open", autospec=True) - def test_check_use_client_cert_config_fallback(self, mock_file): + @mock.patch("os.path.exists", autospec=True) + def test_check_use_client_cert_config_fallback(self, mock_exists, mock_file): + mock_exists.return_value = True # Test fallback for config file when determining if client cert should be used cloudsdk_path = "/path/to/cloudsdk/config" diff --git a/packages/google-auth/tests/transport/test_grpc.py b/packages/google-auth/tests/transport/test_grpc.py index 9f3c117ed933..de3e882ba25a 100644 --- a/packages/google-auth/tests/transport/test_grpc.py +++ b/packages/google-auth/tests/transport/test_grpc.py @@ -404,13 +404,16 @@ def test_secure_authorized_channel_cert_callback_without_client_cert_env( @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True) @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True) class TestSslCredentials(object): + @mock.patch("os.path.exists", autospec=True) def test_no_context_aware_metadata( self, + mock_path_exists, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): + mock_path_exists.return_value = False # Mock that the metadata file doesn't exist. mock_check_config_path.return_value = None @@ -608,19 +611,14 @@ def test_get_client_ssl_credentials_transient_error_retry( certificate_chain=b"cert", private_key=b"key" ) - @mock.patch( - "google.auth.transport.mtls.has_default_client_cert_source", autospec=True - ) def test_get_client_ssl_credentials_auto_enablement( self, - mock_has_default_client_cert_source, mock_check_config_path, mock_load_json_file, mock_get_client_ssl_credentials, mock_ssl_channel_credentials, ): fake_config_content = '{"version": 1, "cert_configs": {"workload": {"cert_path": "/tmp/mock_cert.pem", "key_path": "/tmp/mock_key.pem"}}}' - mock_has_default_client_cert_source.return_value = True mock_get_client_ssl_credentials.return_value = ( True, PUBLIC_CERT_BYTES, @@ -633,9 +631,16 @@ def test_get_client_ssl_credentials_auto_enablement( { environment_vars.GOOGLE_API_CERTIFICATE_CONFIG: "fake_config_path.json", }, - ), mock.patch("builtins.open", mock.mock_open(read_data=fake_config_content)): - # Ensure GOOGLE_API_USE_CLIENT_CERTIFICATE is not present in the environment + ), mock.patch( + "builtins.open", mock.mock_open(read_data=fake_config_content) + ), mock.patch( + "os.path.exists", return_value=True + ): + # Ensure mTLS explicit flags are not present in the environment os.environ.pop(environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, None) + os.environ.pop( + environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE, None + ) ssl_credentials = google.auth.transport.grpc.SslCredentials() assert ssl_credentials.ssl_credentials is not None diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py index 70f258c3ff26..ee108448fa72 100644 --- a/packages/google-auth/tests/transport/test_mtls.py +++ b/packages/google-auth/tests/transport/test_mtls.py @@ -23,13 +23,18 @@ from google.auth.transport import mtls +@mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") @mock.patch("google.auth.transport._mtls_helper._check_config_path") -def test_has_default_client_cert_source_with_context_aware_metadata(mock_check): +def test_has_default_client_cert_source_with_context_aware_metadata( + mock_check, mock_get_cert +): """ Directly tests the logic: if CONTEXT_AWARE_METADATA_PATH is found, return True. """ - # Setup: Return a path only for the Context Aware Metadata Path + # Setup: _get_cert_config_path returns None, so it falls back to context aware metadata + mock_get_cert.return_value = None + def side_effect(path): if path == _mtls_helper.CONTEXT_AWARE_METADATA_PATH: return "/path/to/context_aware_metadata.json" @@ -42,23 +47,36 @@ def side_effect(path): # Assert assert result is True + mock_get_cert.assert_called_once_with(include_context_aware=True) mock_check.assert_any_call(_mtls_helper.CONTEXT_AWARE_METADATA_PATH) - assert side_effect("non-matching-path") is None +@mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") @mock.patch("google.auth.transport._mtls_helper._check_config_path") -def test_has_default_client_cert_source_falls_back(mock_check): +def test_has_default_client_cert_source_without_context_aware( + mock_check, mock_get_cert +): """ - Tests that it skips CONTEXT_AWARE_METADATA_PATH if None, and checks the next path. + Tests that if include_context_aware is False, it skips checking CONTEXT_AWARE_METADATA_PATH. """ + mock_get_cert.return_value = None - # Setup: First path is None, second path is valid - def side_effect(path): - if path == _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH: - return "/path/to/default_cert.json" - return None + result = mtls.has_default_client_cert_source(include_context_aware=False) - mock_check.side_effect = side_effect + assert result is False + mock_get_cert.assert_called_once_with(include_context_aware=False) + mock_check.assert_not_called() + + +@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): + """ + Tests that it checks X.509 WIF first, and if found, returns True without checking context aware metadata. + """ + + # Setup: First path is valid + mock_get_cert.return_value = "/path/to/default_cert.json" # Execute result = mtls.has_default_client_cert_source(True) @@ -66,47 +84,30 @@ def side_effect(path): # Assert assert result is True # Verify the sequence of calls - expected_calls = [ - mock.call(_mtls_helper.CONTEXT_AWARE_METADATA_PATH), - mock.call(_mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH), - ] - mock_check.assert_has_calls(expected_calls) + mock_get_cert.assert_called_once_with(include_context_aware=True) + mock_check.assert_not_called() -@mock.patch("google.auth.transport.mtls.getenv", 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, mock_getenv): - # 1. Mock getenv to return our test path - mock_getenv.side_effect = lambda var: ( - "path/to/cert.json" if var == "GOOGLE_API_CERTIFICATE_CONFIG" else None - ) - - # 2. Mock _check_config_path side effect - def side_effect(path): - # Return None for legacy paths to ensure we reach the env var logic - if path == "path/to/cert.json": - return "/absolute/path/to/cert.json" - return None - - check_config_path.side_effect = side_effect +def test_has_default_client_cert_source_env_var_success( + check_config_path, get_cert_config_path +): + check_config_path.return_value = None + get_cert_config_path.return_value = "/absolute/path/to/cert.json" - # 3. This should now return True assert mtls.has_default_client_cert_source(True) - # 4. Verify the env var path was checked - check_config_path.assert_called_with("path/to/cert.json") + get_cert_config_path.assert_called_with(include_context_aware=True) -@mock.patch("google.auth.transport.mtls.getenv", 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_invalid_config_path( - check_config_path, mock_getenv + check_config_path, get_cert_config_path ): - # Set the env var but make the check fail - mock_getenv.side_effect = lambda var: ( - "invalid/path" if var == "GOOGLE_API_CERTIFICATE_CONFIG" else None - ) check_config_path.return_value = None + get_cert_config_path.return_value = None assert not mtls.has_default_client_cert_source(True)