Skip to content

feat(api_core): add mtls and client configuration logic to gapic_v1 #17750

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

Open
hebaalazzeh wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(api_core): add mtls and client configuration logic to gapic_v1 #17750

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
5 changes: 4 additions & 1 deletion packages/google-api-core/google/api_core/gapic_v1/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -42,6 +44,7 @@

from google.api_core.gapic_v1 import ( # noqa: E402
client_info,
client_utils,
requests,
routing_header,
)
Expand Down
108 changes: 108 additions & 0 deletions packages/google-api-core/google/api_core/gapic_v1/client_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
133 changes: 8 additions & 125 deletions packages/google-api-core/google/api_core/universe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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)
31 changes: 31 additions & 0 deletions packages/google-api-core/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading