feat(auth): add bound token support for access and JWT id tokens for Cloud Run

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

@@ -16,6 +16,7 @@
 
 import base64
 import hashlib
+import json
 import os
 import re
 import stat
@@ -73,6 +74,21 @@ def _is_certificate_file_ready(path):
         return False
 
 
+def _is_in_well_known_dir(path):
+    """Checks if the given path is inside the well-known Agent Identity directory."""
+    if not path:
+        return False
+    well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH)
+    try:
+        real_path = os.path.realpath(path)
+        real_well_known_dir = os.path.realpath(well_known_dir)
+        return (
+            os.path.commonpath([real_well_known_dir, real_path]) == real_well_known_dir
+        )
+    except ValueError:
+        return False
+
+
 def get_agent_identity_certificate_path():
     """Gets the agent certificate path from the certificate config file.
 
@@ -98,16 +114,7 @@ def get_agent_identity_certificate_path():
     # 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)
-    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
+    should_poll = _is_in_well_known_dir(cert_config_path)
 
     return _get_cert_path_with_optional_polling(cert_config_path, should_poll)
 
@@ -141,9 +148,9 @@ def _get_cert_path_with_optional_polling(cert_config_path, should_poll):
             if _is_certificate_file_ready(cert_path):
                 return cert_path
 
-            # The config was parsed, but the cert file is not ready yet
-            if not should_poll:
-                # If polling is disabled, return early.
+            # The config was parsed, but the cert file is not ready yet.
+            # Only poll if both the config path and cert path are in the well-known directory.
+            if not (should_poll and _is_in_well_known_dir(cert_path)):
                 return None
 
             if not has_logged_cert_warning:
@@ -182,7 +189,7 @@ def _get_cert_path_with_optional_polling(cert_config_path, should_poll):
     raise exceptions.RefreshError(
         "Certificate config or certificate file not found after multiple retries. "
         f"Token binding protection is failing. You can turn off this protection by setting "
-        f"{environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES} to false "
+        f"{environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN} to false "
         "to fall back to unbound tokens."
     )
 
@@ -203,8 +210,6 @@ def _parse_cert_path_from_config(cert_config_path):
         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)
 
@@ -221,64 +226,105 @@ def _parse_cert_path_from_config(cert_config_path):
     return workload_config["cert_path"]
 
 
-def get_and_parse_agent_identity_certificate():
+def _is_bound_token_opted_out():
+    """Returns True only if bound tokens are explicitly disabled via env vars."""
+    val = os.environ.get(
+        environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, ""
+    ).strip()
+    if not val:
+        # Fall back to the deprecated env var for backward compatibility
+        val = os.environ.get(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            "true",
+        ).strip()
+    return val.lower() == "false"
+
+
+def get_agent_identity_certificate_and_bytes():
     """Gets and parses the agent identity certificate if not opted out.
 
     Checks if the user has opted out of certificate-bound tokens. If not,
     it gets the certificate path, reads the file, and parses it.
 
     Returns:
-        The parsed certificate object if found and not opted out, otherwise None.
+        Tuple[Optional[cryptography.x509.Certificate], Optional[bytes]]: A tuple
+            of (parsed certificate object, certificate bytes) if found and not
+            opted out, otherwise (None, None).
     """
     # If the user has opted out of cert bound tokens, there is no need to
     # look up the certificate.
-    is_opted_out = (
-        os.environ.get(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
-            "true",
-        ).lower()
-        == "false"
-    )
-    if is_opted_out:
-        return None
+    if _is_bound_token_opted_out():
+        return None, None
 
     # Respect explicit opt-out of mTLS / client certs
     from google.auth.transport import _mtls_helper
 
     env_override = _mtls_helper._check_use_client_cert_env()
     if env_override is False:
-        return None
+        return None, None
 
     cert_path = get_agent_identity_certificate_path()
     if not cert_path:
-        return None
+        return None, None
 
     try:
         with open(cert_path, "rb") as cert_file:
-            cert_bytes = cert_file.read()
-    except PermissionError as e:
+            raw_bytes = cert_file.read()
+    except OSError as e:
         warnings.warn(
             f"Failed to read agent identity certificate file at {cert_path}: {e}. "
             "Token binding protection cannot be enabled. Falling back to unbound tokens."
         )
-        return None
+        return None, None
 
-    return parse_certificate(cert_bytes)
+    cert_blocks = _mtls_helper._CERT_REGEX.findall(raw_bytes)
+    if not cert_blocks:
+        warnings.warn(
+            f"No PEM certificate blocks found in {cert_path}. "
+            "Token binding protection cannot be enabled. Falling back to unbound tokens."
+        )
+        return None, None
+
+    cert_bytes = b"\n".join(block.strip() for block in cert_blocks) + b"\n"
+    try:
+        return parse_certificate(raw_bytes), cert_bytes
+    except (ValueError, ImportError) as e:
+        warnings.warn(
+            f"Failed to parse agent identity certificate at {cert_path}: {e}. "
+            "Token binding protection cannot be enabled. Falling back to unbound tokens."
+        )
+        return None, None
 
 
 def parse_certificate(cert_bytes):
-    """Parses a PEM-encoded certificate.
+    """Parses a PEM-encoded certificate or certificate chain and returns the leaf certificate.
 
     Args:
         cert_bytes (bytes): The PEM-encoded certificate bytes.
 
     Returns:
-        cryptography.x509.Certificate: The parsed certificate object.
+        cryptography.x509.Certificate: The leaf (first) parsed certificate object.
+
+    Raises:
+        ValueError: If no certificates are found or any certificate in the chain
+            is malformed.
+        ImportError: If the cryptography library is not installed.
     """
+    if not cert_bytes:
+        raise ValueError("Certificate bytes cannot be empty or None.")
+
+    from google.auth.transport import _mtls_helper
+
     try:
         from cryptography import x509
 
-        return x509.load_pem_x509_certificate(cert_bytes)
+        cert_blocks = _mtls_helper._CERT_REGEX.findall(cert_bytes)
+        if not cert_blocks:
+            return x509.load_pem_x509_certificate(cert_bytes)
+        if _mtls_helper._has_unmatched_pem_markers(cert_bytes, cert_blocks):
+            raise ValueError("Malformed or truncated PEM certificate chain.")
+        certs = [x509.load_pem_x509_certificate(block) for block in cert_blocks]
+        return certs[0]
     except ImportError as e:
         raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
 
@@ -350,24 +396,17 @@ def calculate_certificate_fingerprint(cert):
 def should_request_bound_token(cert):
     """Determines if a bound token should be requested.
 
-    This is based on the GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
-    environment variable and whether the certificate is an agent identity cert.
+    This is based on the GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN env var
+    (falls back to the deprecated GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
+    if unset or empty) and whether the certificate is an agent identity cert.
 
     Args:
         cert (cryptography.x509.Certificate): The parsed certificate object.
 
     Returns:
         bool: True if a bound token should be requested, False otherwise.
     """
-    is_agent_cert = _is_agent_identity_certificate(cert)
-    is_opted_in = (
-        os.environ.get(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
-            "true",
-        ).lower()
-        == "true"
-    )
-    if not (is_agent_cert and is_opted_in):
+    if _is_bound_token_opted_out():
         return False
 
     # Respect explicit opt-out of mTLS / client certs
@@ -377,7 +416,7 @@ def should_request_bound_token(cert):
     if env_override is False:
         return False
 
-    return True
+    return _is_agent_identity_certificate(cert)
 
 
 def get_cached_cert_fingerprint(cached_cert):

packages/google-auth/google/auth/aio/transport/sessions.py

@@ -19,7 +19,6 @@
 import inspect
 import logging
 import time
-import urllib.parse
 import warnings
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Mapping, Optional, Union
@@ -43,11 +42,6 @@
         ClientTimeout = None
 
 _LOGGER = logging.getLogger(__name__)
-_MTLS_URL_PREFIXES = [
-    "mtls.googleapis.com",
-    "mtls.sandbox.googleapis.com",
-    "p.googleapis.com",
-]
 
 # Tracks the internal aiohttp installation and usage
 try:
@@ -371,15 +365,10 @@ async def request(
                     )
 
                     async def _recover_auth_state():
-                        is_mtls_endpoint = False
                         if self._is_mtls:
-                            hostname = urllib.parse.urlsplit(url).hostname
-                            if hostname:
-                                is_mtls_endpoint = any(
-                                    hostname == prefix
-                                    or hostname.endswith("." + prefix)
-                                    for prefix in _MTLS_URL_PREFIXES
-                                )
+                            is_mtls_endpoint = (
+                                google.auth.transport._mtls_helper.is_mtls_endpoint(url)
+                            )
                             # Snapshot the stale certificate state BEFORE acquiring the lock.
                             # This represents the cert that caused the 401 rejection.
                             if is_mtls_endpoint: