Skip to content

chore(generator): gapic generator centralization mtls #17749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
hebaalazzeh wants to merge 6 commits into
base: feat/gapic-centralization-api-core-mtls
Choose a base branch
from
Draft

chore(generator): gapic generator centralization mtls #17749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bbcc340
refactor(generator): use centralized mtls and client config helpers
hebaalazzeh Jul 16, 2026
bd510e5
Merge branch 'feat/gapic-centralization-api-core-mtls' into feat/gapi…
hebaalazzeh Jul 16, 2026
5cc1bec
Merge branch 'feat/gapic-centralization-api-core-mtls' into feat/gapi…
hebaalazzeh Jul 16, 2026
3c02de0
Merge branch 'feat/gapic-centralization-api-core-mtls' into feat/gapi…
hebaalazzeh Jul 16, 2026
07ff22c
Merge branch 'feat/gapic-centralization-api-core-mtls' into feat/gapi…
hebaalazzeh Jul 17, 2026
d662472
refactor(generator): update templates and goldens to use client_utils…
hebaalazzeh Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 13 additions & 60 deletions ...c-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ from google.api_core import exceptions as core_exceptions
from google.api_core import extended_operation
{% endif %}
from google.api_core import gapic_v1
from google.api_core.gapic_v1 import client_utils
from google.api_core import retry as retries
from google.auth import credentials as ga_credentials # type: ignore
from google.auth.transport import mtls # type: ignore
Expand Down Expand Up @@ -189,30 +190,9 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
_DEFAULT_UNIVERSE = "googleapis.com"

@staticmethod
def _use_client_cert_effective():
"""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 _use_client_cert_effective() -> bool:
"""Returns whether client certificate should be used for mTLS."""
return client_utils.use_client_cert_effective()

@classmethod
def from_service_account_info(cls, info: dict, *args, **kwargs):
Expand Down Expand Up @@ -352,44 +332,17 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
return api_endpoint, client_cert_source

@staticmethod
def _read_environment_variables():
"""Returns the environment variables used by the client.

Returns:
Tuple[bool, str, 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 = {{ service.client_name }}._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
def _read_environment_variables() -> Tuple[bool, str, Optional[str]]:
"""Returns the environment variables used by the client."""
return client_utils.read_environment_variables()

@staticmethod
def _get_client_cert_source(provided_cert_source, use_cert_flag):
"""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 mtls.has_default_client_cert_source():
client_cert_source = mtls.default_client_cert_source()
return client_cert_source
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."""
return client_utils.get_client_cert_source(provided_cert_source, use_cert_flag)

@staticmethod
def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtls_endpoint) -> str:
Expand Down
140 changes: 0 additions & 140 deletions .../gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,146 +172,6 @@ def test__get_default_mtls_endpoint():
assert {{ service.client_name }}._get_default_mtls_endpoint(non_googleapi) == non_googleapi
assert {{ service.client_name }}._get_default_mtls_endpoint(custom_endpoint) == custom_endpoint

def test__read_environment_variables():
assert {{ service.client_name }}._read_environment_variables() == (False, "auto", None)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
assert {{ service.client_name }}._read_environment_variables() == (True, "auto", None)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}):
assert {{ service.client_name }}._read_environment_variables() == (False, "auto", None)

with mock.patch.dict(
os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported"}
):
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with pytest.raises(ValueError) as excinfo:
{{ service.client_name }}._read_environment_variables()
assert (
str(excinfo.value)
== "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`"
)
else:
assert {{ service.client_name }}._read_environment_variables() == (
False,
"auto",
None,
)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}):
assert {{ service.client_name }}._read_environment_variables() == (False, "never", None)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
assert {{ service.client_name }}._read_environment_variables() == (False, "always", None)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}):
assert {{ service.client_name }}._read_environment_variables() == (False, "auto", None)

with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "Unsupported"}):
with pytest.raises(MutualTLSChannelError) as excinfo:
{{ service.client_name }}._read_environment_variables()
assert str(excinfo.value) == "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"

with mock.patch.dict(os.environ, {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": "foo.com"}):
assert {{ service.client_name }}._read_environment_variables() == (False, "auto", "foo.com")


def test_use_client_cert_effective():
# Test case 1: Test when `should_use_client_cert` returns True.
# We mock the `should_use_client_cert` function to simulate a scenario where
# the google-auth library supports automatic mTLS and determines that a
# client certificate should be used.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=True):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 2: Test when `should_use_client_cert` returns False.
# We mock the `should_use_client_cert` function to simulate a scenario where
# the google-auth library supports automatic mTLS and determines that a
# client certificate should NOT be used.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 3: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "true".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 4: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "false".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 5: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "True".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "True"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 6: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "False".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "False"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 7: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "TRUE".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "TRUE"}):
assert {{ service.client_name }}._use_client_cert_effective() is True

# Test case 8: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to "FALSE".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "FALSE"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 9: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is not set.
# In this case, the method should return False, which is the default value.
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, clear=True):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 10: Test when `should_use_client_cert` is unavailable and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to an invalid value.
# The method should raise a ValueError as the environment variable must be either
# "true" or "false".
if not hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "unsupported"}):
with pytest.raises(ValueError):
{{ service.client_name }}._use_client_cert_effective()

# Test case 11: Test when `should_use_client_cert` is available and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is set to an invalid value.
# The method should return False as the environment variable is set to an invalid value.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "unsupported"}):
assert {{ service.client_name }}._use_client_cert_effective() is False

# Test case 12: Test when `should_use_client_cert` is available and the
# `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is unset. Also,
# the GOOGLE_API_CONFIG environment variable is unset.
if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": ""}):
with mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}):
assert {{ service.client_name }}._use_client_cert_effective() is False

def test__get_client_cert_source():
mock_provided_cert_source = mock.Mock()
mock_default_cert_source = mock.Mock()

assert {{ service.client_name }}._get_client_cert_source(None, False) is None
assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, False) is None
assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, True) == mock_provided_cert_source

with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=True):
with mock.patch('google.auth.transport.mtls.default_client_cert_source', return_value=mock_default_cert_source):
assert {{ service.client_name }}._get_client_cert_source(None, True) is mock_default_cert_source
assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source

@mock.patch.object({{ service.client_name }}, "_DEFAULT_ENDPOINT_TEMPLATE", modify_default_endpoint_template({{ service.client_name }}))
{% if 'grpc' in opts.transport %}
Expand Down
73 changes: 13 additions & 60 deletions ...or/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from google.api_core import client_options as client_options_lib
from google.api_core import exceptions as core_exceptions
from google.api_core import gapic_v1
from google.api_core.gapic_v1 import client_utils
from google.api_core import retry as retries
from google.auth import credentials as ga_credentials # type: ignore
from google.auth.transport import mtls # type: ignore
Expand Down Expand Up @@ -144,30 +145,9 @@ def _get_default_mtls_endpoint(api_endpoint) -> Optional[str]:
_DEFAULT_UNIVERSE = "googleapis.com"

@staticmethod
def _use_client_cert_effective():
"""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 _use_client_cert_effective() -> bool:
"""Returns whether client certificate should be used for mTLS."""
return client_utils.use_client_cert_effective()

@classmethod
def from_service_account_info(cls, info: dict, *args, **kwargs):
Expand Down Expand Up @@ -410,44 +390,17 @@ def get_mtls_endpoint_and_cert_source(cls, client_options: Optional[client_optio
return api_endpoint, client_cert_source

@staticmethod
def _read_environment_variables():
"""Returns the environment variables used by the client.

Returns:
Tuple[bool, str, 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 = AssetServiceClient._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
def _read_environment_variables() -> Tuple[bool, str, Optional[str]]:
"""Returns the environment variables used by the client."""
return client_utils.read_environment_variables()

@staticmethod
def _get_client_cert_source(provided_cert_source, use_cert_flag):
"""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 mtls.has_default_client_cert_source():
client_cert_source = mtls.default_client_cert_source()
return client_cert_source
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."""
return client_utils.get_client_cert_source(provided_cert_source, use_cert_flag)

@staticmethod
def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtls_endpoint) -> str:
Expand Down
Loading
Loading