feat(api-core): centralize MTLS fallback functions in gapic_v1 config

Author: hebaalazzeh

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,51 @@ 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

packages/google-api-core/tests/unit/gapic/test_config.py

@@ -12,7 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+from unittest import mock
+
 import pytest
+from google.auth.exceptions import MutualTLSChannelError
 
 try:
     import grpc  # noqa: F401
@@ -91,3 +95,92 @@ 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()