Merge remote-tracking branch 'origin/main' into feat/gapic-generator-centralization-transcoding

Author: hebaalazzeh

packages/google-auth/google/auth/_agent_identity_utils.py

@@ -67,93 +67,88 @@ def _is_certificate_file_ready(path):
         st = os.stat(path)
         return stat.S_ISREG(st.st_mode) and st.st_size > 0
     except PermissionError:
-        # Propagate PermissionError to let caller handle it (fail-fast or fallback)
+        # Propagate PermissionError to let caller handle it (e.g., return early or fallback)
         raise
     except OSError:
         return False
 
 
 def get_agent_identity_certificate_path():
-    """Gets the certificate path from the certificate config file.
+    """Gets the agent certificate path from the certificate config file.
 
     The path to the certificate config file is read from the
     GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function
-    implements a retry mechanism to handle cases where the environment
+    can optionally trigger polling to handle cases where the environment
     variable is set before the files are available on the filesystem.
 
     Returns:
-        str: The path to the leaf certificate file.
+        Optional[str]: The path to the agent's certificate file, or None if unavailable.
 
     Raises:
         google.auth.exceptions.RefreshError: If the certificate config file
             or the certificate file cannot be found after retries.
     """
-    import json
-
     cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
 
-    # Check if the well-known workload directory is mounted.
+    if not cert_config_path:
+        return None
+
+    # We trigger polling only if the config path points to the well-known directory.
+    # Cloud Run dynamically generates these files in this directory, and both the
+    # config file and the certificate file may experience a brief startup latency.
+    # For all other paths, we return early to avoid introducing unnecessary startup
+    # delays.
     well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH)
-    has_well_known_dir = os.path.exists(well_known_dir)
+    try:
+        abs_cert_path = os.path.abspath(cert_config_path)
+        abs_well_known_dir = os.path.abspath(well_known_dir)
+        should_poll = (
+            os.path.commonpath([abs_well_known_dir, abs_cert_path])
+            == abs_well_known_dir
+        )
+    except ValueError:
+        should_poll = False
 
-    # If we have neither a config path nor a well-known mount directory, exit immediately.
-    if not cert_config_path and not has_well_known_dir:
-        return None
+    return _get_cert_path_with_optional_polling(cert_config_path, should_poll)
 
-    # If ECP config path is specified but does not exist, and we are on a workstation, fail-fast immediately.
-    if (
-        cert_config_path
-        and not has_well_known_dir
-        and not os.path.exists(cert_config_path)
-    ):
-        return None
 
+def _get_cert_path_with_optional_polling(cert_config_path, should_poll):
+    """Gets the certificate path, optionally polling until it is ready.
+
+    Args:
+        cert_config_path (str): The path to the certificate configuration file.
+        should_poll (bool): If True, the function will poll for the file and
+            certificate to be ready. If False, it will check only once and
+            return early if they are not immediately available.
+
+    Returns:
+        str: The path to the certificate file, or None if unavailable.
+
+    Raises:
+        google.auth.exceptions.RefreshError: If the certificate config file
+            or the certificate file cannot be found after retries.
+    """
     has_logged_config_warning = False
     has_logged_cert_warning = False
 
     for interval in _POLLING_INTERVALS:
         try:
-            # Path A: Config file is explicitly set
-            if cert_config_path:
-                with open(cert_config_path, "r") as f:
-                    cert_config = json.load(f)
-
-                cert_configs = (
-                    cert_config.get("cert_configs")
-                    if isinstance(cert_config, dict)
-                    else None
-                )
-                workload_config = (
-                    cert_configs.get("workload")
-                    if isinstance(cert_configs, dict)
-                    else None
-                )
-
-                if (
-                    not isinstance(workload_config, dict)
-                    or "cert_path" not in workload_config
-                ):
-                    return None
-
-                cert_path = workload_config["cert_path"]
-                if _is_certificate_file_ready(cert_path):
-                    return cert_path
+            cert_path = _parse_cert_path_from_config(cert_config_path)
 
-                # The config was parsed, but the cert file is not ready yet
-                target_path = cert_path
+            if cert_path is None:
+                return None
 
-            # Path B: Config is NOT set, fallback to the well-known path
-            else:
-                if _is_certificate_file_ready(_WELL_KNOWN_CERT_PATH):
-                    return _WELL_KNOWN_CERT_PATH
+            if _is_certificate_file_ready(cert_path):
+                return cert_path
 
-                # The well-known cert file is not ready yet
-                target_path = _WELL_KNOWN_CERT_PATH
+            # The config was parsed, but the cert file is not ready yet
+            if not should_poll:
+                # If polling is disabled, return early.
+                return None
 
-            # Log a warning on the first failed attempt to load the certificate file
             if not has_logged_cert_warning:
                 warnings.warn(
-                    f"Certificate file not ready at {target_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
+                    f"Certificate file not ready at {cert_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
                 )
                 has_logged_cert_warning = True
 
@@ -164,25 +159,24 @@ def get_agent_identity_certificate_path():
             )
             return None
         except (IOError, ValueError, KeyError) as e:
-            if cert_config_path and os.path.exists(cert_config_path):
+            if os.path.exists(cert_config_path):
                 # If the file exists but has invalid JSON or is unreadable,
-                # we assume it is in its final format and fail-fast by returning None.
+                # we assume it is in its final format and return early (returning None).
+                return None
+
+            if not should_poll:
+                # If polling is disabled, return early if the file doesn't exist.
                 return None
 
-            if not has_logged_config_warning and cert_config_path:
+            if not has_logged_config_warning:
                 warnings.warn(
                     f"Certificate config file not found or incomplete: {e} (from "
                     f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). "
                     f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
                 )
                 has_logged_config_warning = True
-            pass
 
-        # A sleep is required in two cases:
-        # 1. The config file is not found (the except block).
-        # 2. The config file/well-known path is found, but the certificate is not yet available.
-        # In both cases, we need to poll, so we sleep on every iteration
-        # that doesn't return a certificate.
+        # Sleep before the next polling attempt.
         time.sleep(interval)
 
     raise exceptions.RefreshError(
@@ -193,6 +187,40 @@ def get_agent_identity_certificate_path():
     )
 
 
+def _parse_cert_path_from_config(cert_config_path):
+    """Reads the cert config file and returns the cert_path.
+
+    Args:
+        cert_config_path (str): The path to the certificate configuration file.
+
+    Returns:
+        Optional[str]: The path to the certificate file, or None if not found
+            in the config.
+
+    Raises:
+        IOError: If the certificate config file cannot be read.
+        ValueError: If the certificate config file contains invalid JSON.
+        KeyError: If the certificate config file does not contain the
+            expected structure.
+    """
+    import json
+
+    with open(cert_config_path, "r", encoding="utf-8") as f:
+        cert_config = json.load(f)
+
+    cert_configs = (
+        cert_config.get("cert_configs") if isinstance(cert_config, dict) else None
+    )
+    workload_config = (
+        cert_configs.get("workload") if isinstance(cert_configs, dict) else None
+    )
+
+    if not isinstance(workload_config, dict) or "cert_path" not in workload_config:
+        return None
+
+    return workload_config["cert_path"]
+
+
 def get_and_parse_agent_identity_certificate():
     """Gets and parses the agent identity certificate if not opted out.
 

packages/google-auth/google/auth/transport/_mtls_helper.py

@@ -26,6 +26,7 @@
 from typing import cast, Generator, List, Optional, Tuple, Union
 
 from google.auth import _agent_identity_utils
+from google.auth import _cloud_sdk
 from google.auth import environment_vars
 from google.auth import exceptions
 
@@ -51,30 +52,10 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-# A flag to track if we have already logged a warning about mTLS auto-enablement failures.
-# This prevents log spam when client libraries create transports or session instances
-# frequently within a single process.
-_has_logged_mtls_warning = False
-
-
 _PASSPHRASE_REGEX = re.compile(
     b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
 )
 
-# Temporary patch to accomodate incorrect cert config in Cloud Run prod environment.
-_WELL_KNOWN_CLOUD_RUN_CERT_PATH = (
-    "/var/run/secrets/workload-spiffe-credentials/certificates.pem"
-)
-_WELL_KNOWN_CLOUD_RUN_KEY_PATH = (
-    "/var/run/secrets/workload-spiffe-credentials/private_key.pem"
-)
-_INCORRECT_CLOUD_RUN_CERT_PATH = (
-    "/var/lib/volumes/certificate/workload-certificates/certificates.pem"
-)
-_INCORRECT_CLOUD_RUN_KEY_PATH = (
-    "/var/lib/volumes/certificate/workload-certificates/private_key.pem"
-)
-
 
 class _MemfdCreationError(OSError):
     """Raised when Linux in-memory virtual file creation (memfd) fails."""
@@ -436,22 +417,37 @@ def _get_cert_config_path(certificate_config_path=None, include_context_aware=Tr
         The absolute path of the certificate config file, and None if the file does not exist.
     """
 
+    source = "function argument"
+    is_explicit = True
     if certificate_config_path is None:
         env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None)
         if env_path is not None and env_path != "":
             certificate_config_path = env_path
+            source = (
+                f"environment variable {environment_vars.GOOGLE_API_CERTIFICATE_CONFIG}"
+            )
         else:
             env_path = environ.get(
                 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
                 None,
             )
             if include_context_aware and env_path is not None and env_path != "":
                 certificate_config_path = env_path
+                source = f"environment variable {environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH}"
             else:
-                certificate_config_path = CERTIFICATE_CONFIGURATION_DEFAULT_PATH
+                certificate_config_path = os.path.join(
+                    _cloud_sdk.get_config_path(), "certificate_config.json"
+                )
+                is_explicit = False
 
     certificate_config_path = path.expanduser(certificate_config_path)
     if not path.exists(certificate_config_path):
+        if is_explicit:
+            _LOGGER.debug(
+                "Certificate configuration file explicitly specified via %s at %s does not exist",
+                source,
+                certificate_config_path,
+            )
         return None
     return certificate_config_path
 
@@ -489,25 +485,6 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
     cert_path = workload["cert_path"]
     key_path = workload["key_path"]
 
-    # == BEGIN Temporary Cloud Run PATCH ==
-    # See https://github.com/googleapis/google-auth-library-python/issues/1881
-    if (cert_path == _INCORRECT_CLOUD_RUN_CERT_PATH) and (
-        key_path == _INCORRECT_CLOUD_RUN_KEY_PATH
-    ):
-        if not path.exists(cert_path) and not path.exists(key_path):
-            _LOGGER.debug(
-                "Applying Cloud Run certificate path patch. "
-                "Configured paths not found: %s, %s. "
-                "Using well-known paths: %s, %s",
-                cert_path,
-                key_path,
-                _WELL_KNOWN_CLOUD_RUN_CERT_PATH,
-                _WELL_KNOWN_CLOUD_RUN_KEY_PATH,
-            )
-            cert_path = _WELL_KNOWN_CLOUD_RUN_CERT_PATH
-            key_path = _WELL_KNOWN_CLOUD_RUN_KEY_PATH
-    # == END Temporary Cloud Run PATCH ==
-
     return cert_path, key_path
 
 
@@ -755,36 +732,33 @@ def check_use_client_cert():
     will default to False.
     If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
     as True (auto-enabled) if a workload config file exists (pointed at by
-    GOOGLE_API_CERTIFICATE_CONFIG) containing a "workload" section.
+    GOOGLE_API_CERTIFICATE_CONFIG or CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
+    or the default path like ~/.config/gcloud/certificate_config.json)
+    containing a "workload" section.
     Otherwise, it returns False.
 
     Returns:
         bool: Whether the client certificate should be used for mTLS connection.
     """
-    global _has_logged_mtls_warning
     env_override = _check_use_client_cert_env()
     if env_override is not None:
         return env_override
 
     # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
 
-    # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set.
-    cert_path = getenv(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) or getenv(
-        environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH
-    )
+    # Check if a workload config file exists.
+    cert_path = _get_cert_config_path(include_context_aware=True)
 
     if cert_path:
         try:
             with open(cert_path, "r") as f:
                 content = json.load(f)
         except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
-            if not _has_logged_mtls_warning:
-                _LOGGER.warning(
-                    "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
-                    cert_path,
-                    e,
-                )
-                _has_logged_mtls_warning = True
+            _LOGGER.debug(
+                "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
+                cert_path,
+                e,
+            )
             return False
 
         # Structural validation
@@ -794,12 +768,10 @@ def check_use_client_cert():
                 return True
 
         # If we got here, the file exists but the expected structure is missing
-        if not _has_logged_mtls_warning:
-            _LOGGER.warning(
-                "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
-                cert_path,
-            )
-            _has_logged_mtls_warning = True
+        _LOGGER.debug(
+            "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
+            cert_path,
+        )
     return False