feat(api-core): introduce client_utils and tests for use_client_cert_effective, get_client_cert_source, and read_environment_variables

Author: hebaalazzeh

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

@@ -25,10 +25,12 @@
 # Older Python versions safely ignore this variable.
 __lazy_modules__: Set[str] = {
     "google.api_core.gapic_v1.client_info",
+    "google.api_core.gapic_v1.client_utils",
     "google.api_core.gapic_v1.requests",
     "google.api_core.gapic_v1.routing_header",
 }
-__all__ = ["client_info", "requests", "routing_header"]
+__all__ = ["client_info", "client_utils", "requests", "routing_header"]
+
 
 if _has_grpc:
     __lazy_modules__.update(
@@ -42,6 +44,7 @@
 
 from google.api_core.gapic_v1 import (  # noqa: E402
     client_info,
+    client_utils,
     requests,
     routing_header,
 )

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

@@ -0,0 +1,108 @@
+# 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 setup and configuration."""
+
+import os
+from typing import Callable, Optional, Tuple
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+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
+
+
+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/google/api_core/universe.py

@@ -15,9 +15,6 @@
 """Helpers for universe domain."""
 
 from typing import Any, Optional
-from urllib.parse import urlparse, urlunparse
-
-from google.auth.exceptions import MutualTLSChannelError  # type: ignore
 
 DEFAULT_UNIVERSE = "googleapis.com"
 
@@ -39,32 +36,6 @@ def __init__(self, client_universe, credentials_universe):
         super().__init__(message)
 
 
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
 def determine_domain(
     client_universe_domain: Optional[str], universe_domain_env: Optional[str]
 ) -> str:
@@ -81,11 +52,14 @@ def determine_domain(
     Raises:
         ValueError: If the universe domain is an empty string.
     """
-    return get_universe_domain(
-        client_universe_domain,
-        universe_domain_env,
-        default_universe=DEFAULT_UNIVERSE,
-    )
+    universe_domain = DEFAULT_UNIVERSE
+    if client_universe_domain is not None:
+        universe_domain = client_universe_domain
+    elif universe_domain_env is not None:
+        universe_domain = universe_domain_env
+    if len(universe_domain.strip()) == 0:
+        raise EmptyUniverseError
+    return universe_domain
 
 
 def compare_domains(client_universe: str, credentials: Any) -> bool:
@@ -106,94 +80,3 @@ def compare_domains(client_universe: str, credentials: Any) -> bool:
     if client_universe != credentials_universe:
         raise UniverseMismatchError(client_universe, credentials_universe)
     return True
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)

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

@@ -0,0 +1,31 @@
+# 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