refactor(api-core): consolidate client_cert and config_helpers into client_utils.py

Author: hebaalazzeh

packages/google-api-core/google/api_core/gapic_v1/client_cert.py

@@ -13,10 +13,12 @@
 # limitations under the License.
 #
 
-from typing import Optional
+import os
+from typing import Any, Callable, Optional, Tuple
 from urllib.parse import urlparse, urlunparse
 
 from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+from google.auth.transport import mtls  # type: ignore
 
 
 def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
@@ -132,3 +134,90 @@ def get_universe_domain(
     if not resolved:
         raise ValueError("Universe Domain cannot be an empty string.")
     return resolved
+
+
+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
+