chore(api_core): move mtls and client configuration logic to gapic_v1

packages/google-api-core/google/api_core/gapic_v1/__init__.py

@@ -25,9 +25,11 @@
 # Older Python versions safely ignore this variable.
 __lazy_modules__: Set[str] = {
     "google.api_core.gapic_v1.client_info",
+    "google.api_core.gapic_v1.client_cert",
+    "google.api_core.gapic_v1.config_helpers",
     "google.api_core.gapic_v1.routing_header",
 }
-__all__ = ["client_info", "routing_header"]
+__all__ = ["client_info", "client_cert", "config_helpers", "routing_header"]
 
 if _has_grpc:
     __lazy_modules__.update(
@@ -41,6 +43,8 @@
 
 from google.api_core.gapic_v1 import (  # noqa: E402
     client_info,
+    client_cert,
+    config_helpers,
     routing_header,
 )
 

packages/google-api-core/google/api_core/gapic_v1/client_cert.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for client certificate handling and mTLS authentication."""
+
+import os
+from typing import Callable, Optional, Tuple
+
+from google.auth.transport import mtls  # type: ignore
+
+
+def use_client_cert_effective() -> bool:
+    """Returns whether client certificate should be used for mTLS if the
+    google-auth version supports should_use_client_cert automatic mTLS
+    enablement.
+
+    Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.
+
+    Returns:
+        bool: whether client certificate should be used for mTLS
+    Raises:
+        ValueError: (If using a version of google-auth without
+        should_use_client_cert and GOOGLE_API_USE_CLIENT_CERTIFICATE is
+        set to an unexpected value.)
+    """
+    # check if google-auth version supports should_use_client_cert for
+    # automatic mTLS enablement
+    if hasattr(mtls, "should_use_client_cert"):  # pragma: NO COVER
+        return mtls.should_use_client_cert()
+    else:  # pragma: NO COVER
+        # if unsupported, fallback to reading from env var
+        use_client_cert_str = os.getenv(
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
+        ).lower()
+        if use_client_cert_str not in ("true", "false"):
+            raise ValueError(
+                "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` "
+                "must be either `true` or `false`"
+            )
+        return use_client_cert_str == "true"
+
+
+def get_client_cert_source(
+    provided_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+    use_cert_flag: bool,
+) -> Optional[Callable[[], Tuple[bytes, bytes]]]:
+    """Return the client cert source to be used by the client.
+
+    Args:
+        provided_cert_source (Callable[[], Tuple[bytes, bytes]]): The client certificate source provided.
+        use_cert_flag (bool): A flag indicating whether to use the
+            client certificate.
+
+    Returns:
+        Callable[[], Tuple[bytes, bytes]] or None: The client cert source to be used by the client.
+    """
+    if use_cert_flag:
+        if provided_cert_source:
+            return provided_cert_source
+        elif (
+            hasattr(mtls, "has_default_client_cert_source")
+            and mtls.has_default_client_cert_source()
+        ):
+            return mtls.default_client_cert_source()
+        else:
+            raise ValueError(
+                "Client certificate is required for mTLS, but no client certificate source was provided or found."
+            )
+    return None

packages/google-api-core/google/api_core/gapic_v1/config_helpers.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for parsing environment variables."""
+
+import os
+from typing import Optional, Tuple
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+
+from google.api_core.gapic_v1.client_cert import use_client_cert_effective
+
+
+def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
+    """Returns the environment variables used by the client.
+
+    Returns:
+        Tuple[bool, str, Optional[str]]: returns the
+        GOOGLE_API_USE_CLIENT_CERTIFICATE, GOOGLE_API_USE_MTLS_ENDPOINT,
+        and GOOGLE_CLOUD_UNIVERSE_DOMAIN environment variables.
+
+    Raises:
+        ValueError: If GOOGLE_API_USE_CLIENT_CERTIFICATE is not
+            any of ["true", "false"].
+        google.auth.exceptions.MutualTLSChannelError: If
+            GOOGLE_API_USE_MTLS_ENDPOINT is not any of
+            ["auto", "never", "always"].
+    """
+    use_client_cert = use_client_cert_effective()
+    use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
+    universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
+    if use_mtls_endpoint not in ("auto", "never", "always"):
+        raise MutualTLSChannelError(
+            "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` "
+            "must be `never`, `auto` or `always`"
+        )
+    return use_client_cert, use_mtls_endpoint, universe_domain_env

packages/google-api-core/tests/conftest.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def mock_mtls_env():
+    """Autouse session-scoped fixture to isolate unit tests from workstation mTLS environments."""
+    with mock.patch.dict(
+        os.environ,
+        {
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false",
+            "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false",
+        },
+    ):
+        yield

packages/google-api-core/tests/unit/gapic/test_client_cert.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+
+import pytest
+
+from google.auth.transport import mtls
+from google.api_core.gapic_v1.client_cert import (
+    get_client_cert_source,
+    use_client_cert_effective,
+)
+
+
+@pytest.mark.parametrize(
+    "has_method, method_val, env_val, expected",
+    [
+        # should_use_client_cert is available cases
+        (True, True, None, "true"),
+        (True, False, None, "false"),
+        (True, False, "unsupported", "false"),
+        # should_use_client_cert is unavailable cases
+        (False, None, "true", "true"),
+        (False, None, "false", "false"),
+        (False, None, "True", "true"),
+        (False, None, "False", "false"),
+        (False, None, "TRUE", "true"),
+        (False, None, "FALSE", "false"),
+        (False, None, None, "false"),
+        (False, None, "unsupported", "value_error"),
+    ],
+    ids=[
+        "google_auth_true",
+        "google_auth_false",
+        "google_auth_false_env_unsupported",
+        "fallback_env_true_lowercase",
+        "fallback_env_false_lowercase",
+        "fallback_env_true_titlecase",
+        "fallback_env_false_titlecase",
+        "fallback_env_true_uppercase",
+        "fallback_env_false_uppercase",
+        "fallback_env_unset",
+        "fallback_env_unsupported",
+    ],
+)
+def test_use_client_cert_effective(has_method, method_val, env_val, expected):
+    # Mock hasattr to control whether should_use_client_cert exists
+    original_hasattr = hasattr
+
+    def custom_hasattr(obj, name):
+        if obj is mtls and name == "should_use_client_cert":
+            return has_method
+        return original_hasattr(obj, name)
+
+    with mock.patch("google.api_core.gapic_v1.client_cert.hasattr", custom_hasattr):
+        with mock.patch(
+            "google.auth.transport.mtls.should_use_client_cert",
+            create=True,
+            return_value=method_val,
+        ):
+            env = {}
+            if env_val is not None:
+                env["GOOGLE_API_USE_CLIENT_CERTIFICATE"] = env_val
+            with mock.patch.dict(os.environ, env, clear=True):
+                if expected == "value_error":
+                    with pytest.raises(
+                        ValueError, match="must be either `true` or `false`"
+                    ):
+                        use_client_cert_effective()
+                else:
+                    assert use_client_cert_effective() is (expected == "true")
+
+
+@pytest.mark.parametrize(
+    "provided, use_cert, has_default_avail, default_val, expected",
+    [
+        (None, False, True, b"default", None),
+        (b"provided", False, True, b"default", None),
+        (b"provided", True, True, b"default", b"provided"),
+        (None, True, True, b"default", b"default"),
+        (None, True, False, b"default", "value_error"),
+    ],
+    ids=[
+        "use_cert_false_no_provided",
+        "use_cert_false_with_provided",
+        "use_cert_true_with_provided",
+        "use_cert_true_no_provided_default_avail",
+        "use_cert_true_no_provided_default_unavail",
+    ],
+)
+def test_get_client_cert_source(
+    provided, use_cert, has_default_avail, default_val, expected
+):
+    original_hasattr = hasattr
+
+    def custom_hasattr(obj, name):
+        if obj is mtls and name == "has_default_client_cert_source":
+            return has_default_avail
+        return original_hasattr(obj, name)
+
+    with mock.patch("google.api_core.gapic_v1.client_cert.hasattr", custom_hasattr):
+        with mock.patch(
+            "google.auth.transport.mtls.has_default_client_cert_source",
+            create=True,
+            return_value=has_default_avail,
+        ):
+            with mock.patch(
+                "google.auth.transport.mtls.default_client_cert_source",
+                create=True,
+                return_value=default_val,
+            ):
+                if expected == "value_error":
+                    with pytest.raises(
+                        ValueError, match="Client certificate is required for mTLS"
+                    ):
+                        get_client_cert_source(provided, use_cert)
+                else:
+                    assert get_client_cert_source(provided, use_cert) == expected

packages/google-api-core/tests/unit/gapic/test_config_helpers.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+
+import pytest
+
+from google.auth.exceptions import MutualTLSChannelError
+from google.api_core.gapic_v1.config_helpers import read_environment_variables
+
+
+@pytest.mark.parametrize(
+    "env, mock_cert_val, expected",
+    [
+        ({}, False, (False, "auto", None)),
+        ({}, True, (True, "auto", None)),
+        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}, False, (False, "never", None)),
+        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}, False, (False, "always", None)),
+        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}, False, (False, "auto", None)),
+        ({"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid"}, False, "mutual_tls_error"),
+        (
+            {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": "foo.com"},
+            False,
+            (False, "auto", "foo.com"),
+        ),
+    ],
+    ids=[
+        "default_env",
+        "client_cert_true",
+        "mtls_never",
+        "mtls_always",
+        "mtls_auto",
+        "mtls_invalid",
+        "universe_domain",
+    ],
+)
+def test_read_environment_variables(env, mock_cert_val, expected):
+    with mock.patch(
+        "google.api_core.gapic_v1.config_helpers.use_client_cert_effective",
+        return_value=mock_cert_val,
+    ):
+        with mock.patch.dict(os.environ, env, clear=True):
+            if expected == "mutual_tls_error":
+                with pytest.raises(
+                    MutualTLSChannelError, match="must be `never`, `auto` or `always`"
+                ):
+                    read_environment_variables()
+            else:
+                assert read_environment_variables() == expected