fix(auth): centralize cert discovery logic and steps

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:

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
 
 

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"
 

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

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,71 +47,51 @@ 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)
 
     # 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)