Merge branch 'main' into remove-rsa-dependency

Author: daniel-sanche

google/auth/transport/_mtls_helper.py

@@ -47,6 +47,20 @@
     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"
+)
+
 
 def _check_config_path(config_path):
     """Checks for config file path. If it exists, returns the absolute path with user expansion;
@@ -183,6 +197,25 @@ def _get_workload_cert_and_key_paths(config_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
 
 
@@ -279,7 +312,7 @@ def _run_cert_provider_command(command, expect_encrypted_key=False):
 def get_client_ssl_credentials(
     generate_encrypted_key=False,
     context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
-    certificate_config_path=CERTIFICATE_CONFIGURATION_DEFAULT_PATH,
+    certificate_config_path=None,
 ):
     """Returns the client side certificate, private key and passphrase.
 
@@ -306,13 +339,10 @@ def get_client_ssl_credentials(
             the cert, key and passphrase.
     """
 
-    # 1. Check for certificate config json.
-    cert_config_path = _check_config_path(certificate_config_path)
-    if cert_config_path:
-        # Attempt to retrieve X.509 Workload cert and key.
-        cert, key = _get_workload_cert_and_key(cert_config_path)
-        if cert and key:
-            return True, cert, key, None
+    # 1.  Attempt to retrieve X.509 Workload cert and key.
+    cert, key = _get_workload_cert_and_key(certificate_config_path)
+    if cert and key:
+        return True, cert, key, None
 
     # 2. Check for context aware metadata json
     metadata_path = _check_config_path(context_aware_metadata_path)

system_tests/system_tests_async/test_default.py

@@ -16,8 +16,11 @@
 import pytest
 
 from google.auth import _default_async
+from google.auth.exceptions import RefreshError
+
+EXPECT_PROJECT_ID = os.getenv("EXPECT_PROJECT_ID")
+CREDENTIALS = os.getenv("GOOGLE_APPLICATION_CREDENTIALS", "")
 
-EXPECT_PROJECT_ID = os.environ.get("EXPECT_PROJECT_ID")
 
 @pytest.mark.asyncio
 async def test_application_default_credentials(verify_refresh):
@@ -26,4 +29,10 @@ async def test_application_default_credentials(verify_refresh):
     if EXPECT_PROJECT_ID is not None:
         assert project_id is not None
 
-    await verify_refresh(credentials)
+    try:
+        await verify_refresh(credentials)
+    except RefreshError as e:
+        # allow expired credentials for explicit_authorized_user tests
+        # TODO: https://github.com/googleapis/google-auth-library-python/issues/1882
+        if not CREDENTIALS.endswith("authorized_user.json") or "Token has been expired or revoked" not in str(e):
+            raise

system_tests/system_tests_sync/test_default.py

@@ -15,8 +15,10 @@
 import os
 
 import google.auth
+from google.auth.exceptions import RefreshError
 
-EXPECT_PROJECT_ID = os.environ.get("EXPECT_PROJECT_ID")
+EXPECT_PROJECT_ID = os.getenv("EXPECT_PROJECT_ID")
+CREDENTIALS = os.getenv("GOOGLE_APPLICATION_CREDENTIALS", "")
 
 
 def test_application_default_credentials(verify_refresh):
@@ -25,4 +27,10 @@ def test_application_default_credentials(verify_refresh):
     if EXPECT_PROJECT_ID is not None:
         assert project_id is not None
 
-    verify_refresh(credentials)
+    try:
+        verify_refresh(credentials)
+    except RefreshError as e:
+        # allow expired credentials for explicit_authorized_user tests
+        # TODO: https://github.com/googleapis/google-auth-library-python/issues/1882
+        if not CREDENTIALS.endswith("authorized_user.json") or "Token has been expired or revoked" not in str(e):
+            raise

tests/transport/test__mtls_helper.py

@@ -334,9 +334,102 @@ 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_without_metadata(self, mock_check_config_path):
+    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
+    )
+    @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True)
+    def test_success_without_metadata(
+        self, mock_check_config_path, mock_get_workload_cert_and_key
+    ):
         mock_check_config_path.return_value = False
+        mock_get_workload_cert_and_key.return_value = (None, None)
         has_cert, cert, key, passphrase = _mtls_helper.get_client_ssl_credentials()
         assert not has_cert
         assert cert is None
@@ -395,12 +488,17 @@ def test_missing_cert_command(
     )
     @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(
+        "google.auth.transport._mtls_helper._get_workload_cert_and_key", autospec=True
+    )
     def test_customize_context_aware_metadata_path(
         self,
+        mock_get_workload_cert_and_key,
         mock_check_config_path,
         mock_load_json_file,
         mock_run_cert_provider_command,
     ):
+        mock_get_workload_cert_and_key.return_value = (None, None)
         context_aware_metadata_path = "/path/to/metata/data"
         mock_check_config_path.return_value = context_aware_metadata_path
         mock_load_json_file.return_value = {"cert_provider_command": ["command"]}