diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 9497368070dd..6540a2ae25f7 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -61,20 +61,6 @@ 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.""" @@ -489,25 +475,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 @@ -768,10 +735,8 @@ def check_use_client_cert(): # 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: diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py index 96b84e8e587f..8105bc9e8015 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__) @@ -45,24 +44,17 @@ def has_default_client_cert_source(include_context_aware=True): bool: indicating if the default client cert source exists. """ if ( - include_context_aware - and _mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH) + _mtls_helper._get_cert_config_path(include_context_aware=include_context_aware) is not None ): return True if ( - _mtls_helper._check_config_path( - _mtls_helper.CERTIFICATE_CONFIGURATION_DEFAULT_PATH - ) + include_context_aware + and _mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_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 diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index e7b80e92d0ab..0329590e6abe 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -341,93 +341,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 ) @@ -848,7 +761,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 +780,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 +796,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 +812,16 @@ 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.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_exists.return_value = False assert _mtls_helper.check_use_client_cert() is False def test_use_client_cert_precedence(self): @@ -941,7 +864,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..1e1ccc780d34 100644 --- a/packages/google-auth/tests/transport/test_grpc.py +++ b/packages/google-auth/tests/transport/test_grpc.py @@ -633,9 +633,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..1105b7f56420 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,20 @@ 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._check_config_path") -def test_has_default_client_cert_source_falls_back(mock_check): +@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 skips CONTEXT_AWARE_METADATA_PATH if None, and checks the next path. + Tests that it checks X.509 WIF first, and if found, returns True without checking context aware metadata. """ - # 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 - - mock_check.side_effect = side_effect + # 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 +68,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)