feat: auto-enable mTLS when supported certificates are detected

Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>

Author: agrawalradhika-cell

googleapiclient/discovery.py

@@ -649,16 +649,23 @@ def build_from_document(
 
         # Obtain client cert and create mTLS http channel if cert exists.
         client_cert_to_use = None
-        use_client_cert = os.getenv(GOOGLE_API_USE_CLIENT_CERTIFICATE, "false")
-        if not use_client_cert in ("true", "false"):
-            raise MutualTLSChannelError(
-                "Unsupported GOOGLE_API_USE_CLIENT_CERTIFICATE value. Accepted values: true, false"
+        if hasattr(mtls, "should_use_client_cert"):
+            use_client_cert = mtls.should_use_client_cert()
+        else:
+            # if unsupported, fallback to reading from env var
+            use_client_cert_str = os.getenv(
+                "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
+            ).lower()
+            use_client_cert = use_client_cert_str == "true"
+            if use_client_cert_str not in ("true", "false"):
+                raise MutualTLSChannelError(
+                    "Unsupported GOOGLE_API_USE_CLIENT_CERTIFICATE value. Accepted values: true, false"
             )
         if client_options and client_options.client_cert_source:
             raise MutualTLSChannelError(
                 "ClientOptions.client_cert_source is not supported, please use ClientOptions.client_encrypted_cert_source."
             )
-        if use_client_cert == "true":
+        if use_client_cert:
             if (
                 client_options
                 and hasattr(client_options, "client_encrypted_cert_source")

tests/test_discovery.py

@@ -778,7 +778,19 @@ def test_self_signed_jwt_disabled(self):
 
 REGULAR_ENDPOINT = "https://www.googleapis.com/plus/v1/"
 MTLS_ENDPOINT = "https://www.mtls.googleapis.com/plus/v1/"
-
+CONFIG_DATA_WITH_WORKLOAD = {
+    "version": 1,
+    "cert_configs": {
+        "workload": {
+            "cert_path": "path/to/cert/file",
+            "key_path": "path/to/key/file",
+        }
+    },
+}
+CONFIG_DATA_WITHOUT_WORKLOAD = {
+    "version": 1,
+    "cert_configs": {},
+}
 
 class DiscoveryFromDocumentMutualTLS(unittest.TestCase):
     MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.Credentials)
@@ -884,6 +896,47 @@ def test_mtls_with_provided_client_cert(
                 self.check_http_client_cert(plus, has_client_cert=use_client_cert)
                 self.assertEqual(plus._baseUrl, base_url)
 
+    @parameterized.expand(
+        [
+            ("never", "", CONFIG_DATA_WITH_WORKLOAD , REGULAR_ENDPOINT),
+            ("auto", "", CONFIG_DATA_WITH_WORKLOAD, MTLS_ENDPOINT),
+            ("always", "", CONFIG_DATA_WITH_WORKLOAD, MTLS_ENDPOINT),
+            ("never", "", CONFIG_DATA_WITHOUT_WORKLOAD, REGULAR_ENDPOINT),
+            ("auto", "", CONFIG_DATA_WITHOUT_WORKLOAD, REGULAR_ENDPOINT),
+            ("always", "", CONFIG_DATA_WITHOUT_WORKLOAD, MTLS_ENDPOINT),
+        ]
+    )
+    def test_mtls_with_provided_client_cert_unset_environment_variable(
+        self, use_mtls_env, use_client_cert, config_data, base_url
+    ):
+        if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
+            self.skipTest(
+                "The should_use_client_cert function is not available in this "
+                "version of google-auth."
+            )
+        discovery = read_datafile("plus.json")
+        config_filename = "mock_certificate_config.json"
+        config_file_content = json.dumps(config_data)
+        m = mock.mock_open(read_data=config_file_content)
+
+        with mock.patch.dict(
+            "os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}
+        ):
+            with mock.patch.dict(
+                "os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert}
+            ):
+                with mock.patch("builtins.open", m):
+                    with mock.patch.dict("os.environ", {"GOOGLE_API_CERTIFICATE_CONFIG": config_filename}):
+                        plus = build_from_document(
+                            discovery,
+                            credentials=self.MOCK_CREDENTIALS,
+                            client_options={
+                                "client_encrypted_cert_source": self.client_encrypted_cert_source
+                            },
+                        )
+                        self.assertIsNotNone(plus)
+                        self.assertEqual(plus._baseUrl, base_url)
+
     @parameterized.expand(
         [
             ("never", "true"),
@@ -961,6 +1014,61 @@ def test_mtls_with_default_client_cert(
                 self.assertIsNotNone(plus)
                 self.check_http_client_cert(plus, has_client_cert=use_client_cert)
                 self.assertEqual(plus._baseUrl, base_url)
+    @parameterized.expand(
+        [
+            ("never", "", CONFIG_DATA_WITH_WORKLOAD, REGULAR_ENDPOINT),
+            ("auto", "", CONFIG_DATA_WITH_WORKLOAD, MTLS_ENDPOINT),
+            ("always", "", CONFIG_DATA_WITH_WORKLOAD, MTLS_ENDPOINT),
+            ("never", "", CONFIG_DATA_WITHOUT_WORKLOAD, REGULAR_ENDPOINT),
+            ("auto", "", CONFIG_DATA_WITHOUT_WORKLOAD, REGULAR_ENDPOINT),
+            ("always", "", CONFIG_DATA_WITHOUT_WORKLOAD, MTLS_ENDPOINT),
+        ]
+    )
+    @mock.patch(
+        "google.auth.transport.mtls.has_default_client_cert_source", autospec=True
+    )
+    @mock.patch(
+        "google.auth.transport.mtls.default_client_encrypted_cert_source", autospec=True
+    )
+    def test_mtls_with_default_client_cert_with_unset_environment_variable(
+        self,
+        use_mtls_env,
+        use_client_cert,
+        config_data,
+        base_url,
+        default_client_encrypted_cert_source,
+        has_default_client_cert_source,
+    ):
+        if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
+            self.skipTest(
+                "The should_use_client_cert function is not available in this "
+                "version of google-auth."
+            )
+        has_default_client_cert_source.return_value = True
+        default_client_encrypted_cert_source.return_value = (
+            self.client_encrypted_cert_source
+        )
+        discovery = read_datafile("plus.json")
+        config_filename = "mock_certificate_config.json"
+        config_file_content = json.dumps(config_data)
+        m = mock.mock_open(read_data=config_file_content)
+
+        with mock.patch.dict(
+            "os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}
+        ):
+            with mock.patch.dict(
+                "os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert}
+            ):
+                with mock.patch("builtins.open", m):
+                    with mock.patch.dict("os.environ", {"GOOGLE_API_CERTIFICATE_CONFIG": config_filename}):
+                        plus = build_from_document(
+                        discovery,
+                        credentials=self.MOCK_CREDENTIALS,
+                        adc_cert_path=self.ADC_CERT_PATH,
+                        adc_key_path=self.ADC_KEY_PATH,
+                        )
+                        self.assertIsNotNone(plus)
+                        self.assertEqual(plus._baseUrl, base_url)
 
     @parameterized.expand(
         [