diff --git a/packages/google-api-core/google/api_core/gapic_v1/__init__.py b/packages/google-api-core/google/api_core/gapic_v1/__init__.py index e5b7ad352ee6..e8cae980d58a 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/__init__.py +++ b/packages/google-api-core/google/api_core/gapic_v1/__init__.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from google.api_core.gapic_v1 import client_helpers from google.api_core.gapic_v1 import client_info from google.api_core.gapic_v1 import config from google.api_core.gapic_v1 import config_async @@ -20,6 +21,7 @@ from google.api_core.gapic_v1 import routing_header __all__ = [ + "client_helpers", "client_info", "config", "config_async", diff --git a/packages/google-api-core/google/api_core/gapic_v1/client_helpers.py b/packages/google-api-core/google/api_core/gapic_v1/client_helpers.py new file mode 100644 index 000000000000..feaa44c2752a --- /dev/null +++ b/packages/google-api-core/google/api_core/gapic_v1/client_helpers.py @@ -0,0 +1,257 @@ +# -*- 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 GAPIC client initialization.""" + +import os +import re +import uuid +from typing import Any, Optional, Tuple + +from google.auth.exceptions import MutualTLSChannelError # type: ignore +from google.auth.transport import mtls # type: ignore + +_MTLS_ENDPOINT_RE = re.compile( + r"(?P[^.]+)(?P\.mtls)?(?P\.sandbox)?" + r"(?P\.googleapis\.com)?" +) + + +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. + Args: + api_endpoint (Optional[str]): the api endpoint to convert. + Returns: + Optional[str]: converted mTLS api endpoint. + """ + if not api_endpoint: + return api_endpoint + + m = _MTLS_ENDPOINT_RE.match(api_endpoint) + if m is None: + # Could not parse api_endpoint; return as-is. + return api_endpoint + + name, mtls_group, sandbox, googledomain = m.groups() + if mtls_group or not googledomain: + return api_endpoint + + if sandbox: + return api_endpoint.replace( + "sandbox.googleapis.com", "mtls.sandbox.googleapis.com" + ) + + return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com") + + +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_api_endpoint( + api_override: Optional[str], + client_cert_source: Optional[Any], + universe_domain: str, + use_mtls_endpoint: str, + default_universe: str, + default_mtls_endpoint: Optional[str], + default_endpoint_template: str, +) -> Optional[str]: + """Return the API endpoint used by the client. + + Args: + api_override (Optional[str]): The API endpoint override. + client_cert_source (Optional[Any]): The client certificate source. + universe_domain (str): The universe domain. + use_mtls_endpoint (str): How to use the mTLS endpoint. + default_universe (str): The default universe. + default_mtls_endpoint (Optional[str]): The default mTLS endpoint. + default_endpoint_template (str): The default endpoint template. + + Returns: + Optional[str]: The API endpoint to be used by the client. + """ + if api_override is not None: + return api_override + elif use_mtls_endpoint == "always" or ( + use_mtls_endpoint == "auto" and client_cert_source + ): + if universe_domain != default_universe: + raise MutualTLSChannelError( + f"mTLS is not supported in any universe other than " + f"{default_universe}." + ) + return default_mtls_endpoint + else: + return default_endpoint_template.format( + UNIVERSE_DOMAIN=universe_domain + ) # noqa: E501 + + +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() # noqa: E501 + 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 + + +def get_client_cert_source( + provided_cert_source: Optional[Any], use_cert_flag: bool +) -> Optional[Any]: + """Return the client cert source to be used by the client. + + Args: + provided_cert_source (bytes): The client certificate source provided. + use_cert_flag (bool): A flag indicating whether to use the + client certificate. + + Returns: + bytes or None: The client cert source to be used by the client. + """ + client_cert_source = None + if use_cert_flag: + if provided_cert_source: + client_cert_source = provided_cert_source + elif ( + hasattr(mtls, "has_default_client_cert_source") + and mtls.has_default_client_cert_source() + ): + client_cert_source = mtls.default_client_cert_source() + return client_cert_source + + +def get_universe_domain( + client_universe_domain: Optional[str], + universe_domain_env: Optional[str], + default_universe: str, +) -> str: + """Return the universe domain used by the client. + + Args: + client_universe_domain (Optional[str]): The universe domain + configured via the client options. + universe_domain_env (Optional[str]): The universe domain + configured via the "GOOGLE_CLOUD_UNIVERSE_DOMAIN" env var. + default_universe (str): The default universe domain. + + Returns: + str: The universe domain to be used by the client. + + Raises: + ValueError: If the universe domain is an empty string. + """ + 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 ValueError("Universe Domain cannot be an empty string.") + return universe_domain + + +def setup_request_id( + request: Any, field_name: str, is_proto3_optional: bool +) -> None: # noqa: E501 + """Populate a UUID4 field in the request if it is not already set. + + Args: + request (Union[google.protobuf.message.Message, dict]): The + request object. + field_name (str): The name of the field to populate. + is_proto3_optional (bool): Whether the field is proto3 optional. + """ + if isinstance(request, dict): + if is_proto3_optional: + if field_name not in request: + request[field_name] = str(uuid.uuid4()) + elif not request.get(field_name): + request[field_name] = str(uuid.uuid4()) + return + + if is_proto3_optional: + try: + # Pure protobuf messages + if not request.HasField(field_name): + setattr(request, field_name, str(uuid.uuid4())) + except (AttributeError, ValueError): + # Proto-plus messages or other objects + if field_name not in request: + setattr(request, field_name, str(uuid.uuid4())) + else: + if not getattr(request, field_name): + setattr(request, field_name, str(uuid.uuid4())) + + +# Backward compatibility aliases for private methods +# Previously, gapic-generator-python generated clients used these methods +_use_client_cert_effective = use_client_cert_effective +_get_default_mtls_endpoint = get_default_mtls_endpoint +_read_environment_variables = read_environment_variables +_get_client_cert_source = get_client_cert_source +_get_universe_domain = get_universe_domain +_setup_request_id = setup_request_id diff --git a/packages/google-api-core/tests/unit/gapic/test_client_helpers.py b/packages/google-api-core/tests/unit/gapic/test_client_helpers.py new file mode 100644 index 000000000000..41826cbd1e5f --- /dev/null +++ b/packages/google-api-core/tests/unit/gapic/test_client_helpers.py @@ -0,0 +1,266 @@ +# -*- 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 + +try: + import grpc # noqa: F401 +except ImportError: + pytest.skip("No GRPC", allow_module_level=True) + +from google.api_core.gapic_v1 import client_helpers +from google.auth.exceptions import MutualTLSChannelError + + +def test_get_default_mtls_endpoint(): + # Test valid API endpoints + assert ( + client_helpers.get_default_mtls_endpoint("foo.googleapis.com") + == "foo.mtls.googleapis.com" + ) + assert ( + client_helpers.get_default_mtls_endpoint("foo.sandbox.googleapis.com") + == "foo.mtls.sandbox.googleapis.com" + ) + + # Test endpoints that shouldn't be converted + assert ( + client_helpers.get_default_mtls_endpoint("foo.mtls.googleapis.com") + == "foo.mtls.googleapis.com" + ) + assert client_helpers.get_default_mtls_endpoint("foo.com") == "foo.com" + + # Test empty/None endpoints + assert client_helpers.get_default_mtls_endpoint("") == "" + assert client_helpers.get_default_mtls_endpoint(None) is None + + +@mock.patch("google.auth.transport.mtls.should_use_client_cert", create=True) +def test_use_client_cert_effective_with_google_auth(mock_method): + # Test when google-auth supports the method + mock_method.return_value = True + assert client_helpers.use_client_cert_effective() is True + + mock_method.return_value = False + assert client_helpers.use_client_cert_effective() is False + + +@mock.patch.dict(os.environ, {}, clear=True) +def test_use_client_cert_effective_fallback(): + # We must patch hasattr to simulate google-auth lacking the method + with mock.patch( + "google.api_core.gapic_v1.client_helpers.hasattr", return_value=False + ): + # Default is false + assert client_helpers.use_client_cert_effective() is False + + env_true = {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"} + with mock.patch.dict(os.environ, env_true): + assert client_helpers.use_client_cert_effective() is True + + with mock.patch.dict( + os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"} + ): + assert client_helpers.use_client_cert_effective() is False + + with mock.patch.dict( + os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "invalid"} + ): + match_str = "must be either `true` or `false`" + with pytest.raises(ValueError, match=match_str): + client_helpers.use_client_cert_effective() + + +def test_get_api_endpoint_override(): + # If api_override is provided, it should be returned + # regardless of other args + endpoint = client_helpers.get_api_endpoint( + api_override="custom.endpoint.com", + client_cert_source=None, + universe_domain="googleapis.com", + use_mtls_endpoint="auto", + default_universe="googleapis.com", + default_mtls_endpoint="foo.mtls.googleapis.com", + default_endpoint_template="foo.{UNIVERSE_DOMAIN}", + ) + assert endpoint == "custom.endpoint.com" + + +def test_get_api_endpoint_mtls_always(): + # use_mtls_endpoint == "always" should use the default mtls endpoint + endpoint = client_helpers.get_api_endpoint( + api_override=None, + client_cert_source=None, + universe_domain="googleapis.com", + use_mtls_endpoint="always", + default_universe="googleapis.com", + default_mtls_endpoint="foo.mtls.googleapis.com", + default_endpoint_template="foo.{UNIVERSE_DOMAIN}", + ) + assert endpoint == "foo.mtls.googleapis.com" + + +def test_get_api_endpoint_mtls_auto_with_cert(): + # "auto" with client_cert_source should use mtls + endpoint = client_helpers.get_api_endpoint( + api_override=None, + client_cert_source=mock.Mock(), + universe_domain="googleapis.com", + use_mtls_endpoint="auto", + default_universe="googleapis.com", + default_mtls_endpoint="foo.mtls.googleapis.com", + default_endpoint_template="foo.{UNIVERSE_DOMAIN}", + ) + assert endpoint == "foo.mtls.googleapis.com" + + +def test_get_api_endpoint_mtls_auto_no_cert(): + # "auto" without client_cert_source should use the default template + endpoint = client_helpers.get_api_endpoint( + api_override=None, + client_cert_source=None, + universe_domain="googleapis.com", + use_mtls_endpoint="auto", + default_universe="googleapis.com", + default_mtls_endpoint="foo.mtls.googleapis.com", + default_endpoint_template="foo.{UNIVERSE_DOMAIN}", + ) + assert endpoint == "foo.googleapis.com" + + +def test_get_api_endpoint_mtls_universe_mismatch(): + # mTLS is only supported in the default universe + with pytest.raises(MutualTLSChannelError, match="mTLS is not supported"): + client_helpers.get_api_endpoint( + api_override=None, + client_cert_source=mock.Mock(), + universe_domain="custom-universe.com", + use_mtls_endpoint="auto", + default_universe="googleapis.com", + default_mtls_endpoint="foo.mtls.googleapis.com", + default_endpoint_template="foo.{UNIVERSE_DOMAIN}", + ) + + +@mock.patch( + "google.api_core.gapic_v1.client_helpers.use_client_cert_effective" +) # noqa: E501 +@mock.patch.dict(os.environ, clear=True) +def test_read_environment_variables(mock_effective): + mock_effective.return_value = True + os.environ["GOOGLE_API_USE_MTLS_ENDPOINT"] = "always" + os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] = "custom.com" + + cert, mtls, domain = client_helpers.read_environment_variables() + assert cert is True + assert mtls == "always" + assert domain == "custom.com" + + +@mock.patch.dict(os.environ, clear=True) +def test_read_environment_variables_invalid_mtls(): + os.environ["GOOGLE_API_USE_MTLS_ENDPOINT"] = "invalid" + with pytest.raises( + MutualTLSChannelError, match="must be `never`, `auto` or `always`" + ): + client_helpers.read_environment_variables() + + +@mock.patch( + "google.auth.transport.mtls.has_default_client_cert_source", create=True +) # noqa: E501 +@mock.patch( + "google.auth.transport.mtls.default_client_cert_source", create=True +) # noqa: E501 +def test_get_client_cert_source(mock_default, mock_has_default): + mock_default.return_value = b"default_cert" + mock_has_default.return_value = True + + # When use_cert_flag is False, return None + assert client_helpers.get_client_cert_source(b"provided", False) is None + + # When provided_cert_source is given, return provided + assert ( + client_helpers.get_client_cert_source(b"provided", True) == b"provided" + ) # noqa: E501 + + # When no provided cert but default is available + assert client_helpers.get_client_cert_source(None, True) == b"default_cert" + + +def test_get_universe_domain(): + # client_universe_domain takes precedence + assert ( + client_helpers.get_universe_domain( + "client.com", "env.com", "default.com" + ) # noqa: E501 + == "client.com" + ) + + # env takes precedence over default + assert ( + client_helpers.get_universe_domain(None, "env.com", "default.com") + == "env.com" # noqa: E501 + ) + + # fallback to default + assert ( + client_helpers.get_universe_domain(None, None, "default.com") + == "default.com" # noqa: E501 + ) + + +def test_get_universe_domain_empty(): + with pytest.raises(ValueError, match="cannot be an empty string"): + client_helpers.get_universe_domain("", None, "default.com") + + +def test_setup_request_id(): + import uuid + + # test dict request + req = {} + client_helpers.setup_request_id(req, "request_id", True) + assert "request_id" in req + uuid_str = req["request_id"] + uuid.UUID(uuid_str) # verify it is a valid UUID + + # test dict request when already set + req = {"request_id": "existing"} + client_helpers.setup_request_id(req, "request_id", True) + assert req["request_id"] == "existing" + + class DummyRequest: + def __init__(self): + self.request_id = "" + + def HasField(self, field_name): + if not hasattr(self, field_name): + raise ValueError() + return bool(getattr(self, field_name)) + + # test object request proto3 optional true + req_obj = DummyRequest() + client_helpers.setup_request_id(req_obj, "request_id", True) + assert req_obj.request_id != "" + uuid.UUID(req_obj.request_id) + + # test object request proto3 optional false + req_obj2 = DummyRequest() + client_helpers.setup_request_id(req_obj2, "request_id", False) + assert req_obj2.request_id != "" + uuid.UUID(req_obj2.request_id)