diff --git a/packages/google-api-core/google/api_core/gapic_v1/config.py b/packages/google-api-core/google/api_core/gapic_v1/config.py index 17e23e032ff9..624a0ef9890f 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/config.py +++ b/packages/google-api-core/google/api_core/gapic_v1/config.py @@ -19,10 +19,14 @@ """ import collections +import os +from typing import Callable, Optional, Tuple import grpc from google.api_core import exceptions, retry, timeout +from google.auth.exceptions import MutualTLSChannelError # type: ignore +from google.auth.transport import mtls # type: ignore _MILLIS_PER_SECOND = 1000.0 @@ -170,3 +174,53 @@ def parse_method_configs(interface_config, retry_impl=retry.Retry): method_configs[method_name] = MethodConfig(retry=retry_, timeout=timeout_) return method_configs + + +def use_client_cert_effective() -> bool: + """Returns whether client certificate should be used for mTLS.""" + if hasattr(mtls, "should_use_client_cert"): + return mtls.should_use_client_cert() + else: + 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.""" + 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() + else: + raise ValueError( + "Client certificate is required for mTLS, but no client certificate source was provided or found." + ) + return client_cert_source + + +def read_environment_variables() -> Tuple[bool, str, Optional[str]]: + """Returns the environment variables used by the client.""" + 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 diff --git a/packages/google-api-core/tests/unit/gapic/test_config.py b/packages/google-api-core/tests/unit/gapic/test_config.py index 6e3a168e79b2..5f5811e9b110 100644 --- a/packages/google-api-core/tests/unit/gapic/test_config.py +++ b/packages/google-api-core/tests/unit/gapic/test_config.py @@ -12,6 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os +from unittest import mock + import pytest try: @@ -21,6 +24,7 @@ from google.api_core import exceptions from google.api_core.gapic_v1 import config +from google.auth.exceptions import MutualTLSChannelError INTERFACE_CONFIG = { "retry_codes": { @@ -91,3 +95,107 @@ def test_create_method_configs(): retry, timeout = method_configs["Plain"] assert retry is None assert timeout._timeout == 30.0 + + +def test_use_client_cert_effective_true(): + mock_mtls = mock.Mock(spec=["should_use_client_cert"]) + mock_mtls.should_use_client_cert.return_value = True + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + assert config.use_client_cert_effective() is True + + +def test_use_client_cert_effective_false(): + mock_mtls = mock.Mock(spec=["should_use_client_cert"]) + mock_mtls.should_use_client_cert.return_value = False + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + assert config.use_client_cert_effective() is False + + +def test_use_client_cert_effective_fallback_env_true(): + mock_mtls = mock.Mock(spec=[]) + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}): + assert config.use_client_cert_effective() is True + + +def test_use_client_cert_effective_fallback_env_false(): + mock_mtls = mock.Mock(spec=[]) + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + with mock.patch.dict( + os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"} + ): + assert config.use_client_cert_effective() is False + + +def test_use_client_cert_effective_fallback_env_invalid(): + mock_mtls = mock.Mock(spec=[]) + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + with mock.patch.dict( + os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "invalid"} + ): + with pytest.raises( + ValueError, + match="Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`", + ): + config.use_client_cert_effective() + + +def test_get_client_cert_source_provided(): + source = mock.Mock() + assert config.get_client_cert_source(source, True) == source + + +def test_get_client_cert_source_default(): + mock_mtls = mock.Mock( + spec=["has_default_client_cert_source", "default_client_cert_source"] + ) + mock_mtls.has_default_client_cert_source.return_value = True + mock_source = mock.Mock() + mock_mtls.default_client_cert_source.return_value = mock_source + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + assert config.get_client_cert_source(None, True) == mock_source + + +def test_get_client_cert_source_none(): + mock_mtls = mock.Mock( + spec=["has_default_client_cert_source", "default_client_cert_source"] + ) + mock_mtls.has_default_client_cert_source.return_value = False + with mock.patch("google.api_core.gapic_v1.config.mtls", mock_mtls): + with pytest.raises( + ValueError, + match="Client certificate is required for mTLS, but no client certificate source was provided or found.", + ): + config.get_client_cert_source(None, True) + + +def test_get_client_cert_source_use_cert_flag_false(): + assert config.get_client_cert_source(None, False) is None + source = mock.Mock() + assert config.get_client_cert_source(source, False) is None + + +def test_read_environment_variables(): + with mock.patch( + "google.api_core.gapic_v1.config.use_client_cert_effective", return_value=True + ): + with mock.patch.dict( + os.environ, + { + "GOOGLE_API_USE_MTLS_ENDPOINT": "always", + "GOOGLE_CLOUD_UNIVERSE_DOMAIN": "my-universe.com", + }, + ): + use_cert, use_mtls, universe = config.read_environment_variables() + assert use_cert is True + assert use_mtls == "always" + assert universe == "my-universe.com" + + +def test_read_environment_variables_invalid_mtls(): + with mock.patch( + "google.api_core.gapic_v1.config.use_client_cert_effective", return_value=True + ): + with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid"}): + with pytest.raises(MutualTLSChannelError): + config.read_environment_variables()