diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py
index 4de5d709b4b5..edab2112a85d 100644
--- a/packages/google-auth/google/auth/_agent_identity_utils.py
+++ b/packages/google-auth/google/auth/_agent_identity_utils.py
@@ -73,6 +73,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 +113,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,8 +147,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:
+            # 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)):
                 # If polling is disabled, return early.
                 return None
 
@@ -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."
     )
 
@@ -221,64 +228,101 @@ 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)
+    if val is not None:
+        return val.lower() == "false"
+
+    # Fall back to the deprecated env var for backward compatibility
+    return (
+        os.environ.get(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            "true",
+        ).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
+
+    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
 
-    return parse_certificate(cert_bytes)
+    cert_bytes = b"\n".join(block.strip() for block in cert_blocks) + b"\n"
+    try:
+        return parse_certificate(cert_bytes), cert_bytes
+    except ValueError 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.
+    """Validates a PEM-encoded 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.
     """
     try:
         from cryptography import x509
+        from google.auth.transport import _mtls_helper
 
-        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)
+        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,8 +394,9 @@ 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) and whether the certificate is an agent identity cert.
 
     Args:
         cert (cryptography.x509.Certificate): The parsed certificate object.
@@ -360,14 +405,7 @@ def should_request_bound_token(cert):
         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 not is_agent_cert or _is_bound_token_opted_out():
         return False
 
     # Respect explicit opt-out of mTLS / client certs
diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index f2ced1280e50..e5db7f76f45e 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -21,7 +21,6 @@
 import logging
 import time
 from typing import Mapping, Optional, TYPE_CHECKING, Union
-import urllib.parse
 import warnings
 
 from google.auth import _exponential_backoff, exceptions
@@ -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:
diff --git a/packages/google-auth/google/auth/compute_engine/_metadata.py b/packages/google-auth/google/auth/compute_engine/_metadata.py
index 1ea7792c2cdd..55386a6772f0 100644
--- a/packages/google-auth/google/auth/compute_engine/_metadata.py
+++ b/packages/google-auth/google/auth/compute_engine/_metadata.py
@@ -27,6 +27,7 @@
 
 import requests
 
+from google.auth import _agent_identity_utils
 from google.auth import _helpers
 from google.auth import environment_vars
 from google.auth import exceptions
@@ -255,6 +256,8 @@ def get(
     headers=None,
     return_none_for_not_found_error=False,
     timeout=_METADATA_DEFAULT_TIMEOUT,
+    method="GET",
+    body=None,
 ):
     """Fetch a resource from the metadata server.
 
@@ -276,6 +279,8 @@ def get(
         return_none_for_not_found_error (Optional[bool]): If True, returns None
             for 404 error instead of throwing an exception.
         timeout (int): How long to wait, in seconds for the metadata server to respond.
+        method (str): The HTTP method to use for the request. Defaults to "GET".
+        body (Optional[bytes]): The HTTP request body payload to send. Defaults to None.
 
     Returns:
         Union[Mapping, str]: If the metadata server returns JSON, a mapping of
@@ -288,8 +293,12 @@ def get(
         google.auth.exceptions.MutualTLSChannelError: if using mtls and the environment
             configuration is invalid for mTLS (for example, the metadata host
             has been overridden in strict mTLS mode).
+        ValueError: if a request body is specified with the GET method.
 
     """
+    if body is not None and method.upper() == "GET":
+        raise ValueError("Request body cannot be specified with GET method.")
+
     use_mtls = _mtls.should_use_mds_mtls()
     # Prepare the request object for mTLS if needed.
     # This will create a new request object with the mTLS session.
@@ -319,9 +328,15 @@ def get(
     last_exception = None
     for attempt in backoff:
         try:
-            response = request(
-                url=url, method="GET", headers=headers_to_use, timeout=timeout
-            )
+            kwargs = {
+                "url": url,
+                "method": method,
+                "headers": headers_to_use,
+                "timeout": timeout,
+            }
+            if body is not None:
+                kwargs["body"] = body
+            response = request(**kwargs)
             if response.status in transport.DEFAULT_RETRYABLE_STATUS_CODES:
                 _LOGGER.warning(
                     "Compute Engine Metadata server unavailable on "
@@ -465,6 +480,32 @@ def get_service_account_info(request, service_account="default"):
     return get(request, path, params={"recursive": "true"})
 
 
+def _get_token_request_params(metrics_header_value):
+    """Returns (method, body, headers) for a metadata server token request.
+
+    Defaults to a standard GET request with the x-goog-api-client metrics header.
+    Upgrades to a POST request with a JSON certificate_chain body and
+    Content-Type header if an Agent Identity certificate is present and bound
+    tokens are enabled.
+
+    Args:
+        metrics_header_value (str): Value for the x-goog-api-client header.
+
+    Returns:
+        Tuple[str, Optional[bytes], Mapping[str, str]]: A tuple of
+            (HTTP method, request body bytes, request headers).
+    """
+    headers = {metrics.API_CLIENT_HEADER: metrics_header_value}
+    cert, cert_bytes = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+    if cert and _agent_identity_utils.should_request_bound_token(cert):
+        headers["Content-Type"] = "application/json"
+        body = json.dumps({"certificate_chain": cert_bytes.decode("utf-8")}).encode(
+            "utf-8"
+        )
+        return "POST", body, headers
+    return "GET", None, headers
+
+
 def get_service_account_token(request, service_account="default", scopes=None):
     """Get the OAuth 2.0 access token for a service account.
 
@@ -483,26 +524,20 @@ def get_service_account_token(request, service_account="default", scopes=None):
         google.auth.exceptions.TransportError: if an error occurred while
             retrieving metadata.
     """
-    from google.auth import _agent_identity_utils
-
     params = {}
     if scopes:
         if not isinstance(scopes, str):
             scopes = ",".join(scopes)
         params["scopes"] = scopes
 
-    cert = _agent_identity_utils.get_and_parse_agent_identity_certificate()
-    if cert:
-        if _agent_identity_utils.should_request_bound_token(cert):
-            fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(cert)
-            params["bindCertificateFingerprint"] = fingerprint
-
-    metrics_header = {
-        metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds()
-    }
+    method, body, headers = _get_token_request_params(
+        metrics.token_request_access_token_mds()
+    )
 
     path = "instance/service-accounts/{0}/token".format(service_account)
-    token_json = get(request, path, params=params, headers=metrics_header)
+    token_json = get(
+        request, path, params=params, headers=headers, method=method, body=body
+    )
     token_expiry = _helpers.utcnow() + datetime.timedelta(
         seconds=token_json["expires_in"]
     )
diff --git a/packages/google-auth/google/auth/compute_engine/credentials.py b/packages/google-auth/google/auth/compute_engine/credentials.py
index 3701751bda2b..827b472d70d6 100644
--- a/packages/google-auth/google/auth/compute_engine/credentials.py
+++ b/packages/google-auth/google/auth/compute_engine/credentials.py
@@ -528,11 +528,17 @@ def _call_metadata_identity_endpoint(self, request):
         try:
             path = "instance/service-accounts/default/identity"
             params = {"audience": self._target_audience, "format": "full"}
-            metrics_header = {
-                metrics.API_CLIENT_HEADER: metrics.token_request_id_token_mds()
-            }
+            method, body, headers = _metadata._get_token_request_params(
+                metrics.token_request_id_token_mds()
+            )
+
             id_token = _metadata.get(
-                request, path, params=params, headers=metrics_header
+                request,
+                path,
+                params=params,
+                headers=headers,
+                method=method,
+                body=body,
             )
         except exceptions.TransportError as caught_exc:
             new_exc = exceptions.RefreshError(caught_exc)
diff --git a/packages/google-auth/google/auth/environment_vars.py b/packages/google-auth/google/auth/environment_vars.py
index 7d82d288a24c..2ded6a0bc87c 100644
--- a/packages/google-auth/google/auth/environment_vars.py
+++ b/packages/google-auth/google/auth/environment_vars.py
@@ -129,10 +129,17 @@
 """Environment variable defining the location of Google API certificate config
 file. This variable is the fallback of GOOGLE_API_CERTIFICATE_CONFIG."""
 
+GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN = "GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN"
+"""Environment variable controlling whether to enable runtime bound tokens."""
+
 GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES = (
     "GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES"
 )
-"""Environment variable to prevent agent token sharing for GCP services."""
+"""Environment variable to prevent agent token sharing for GCP services.
+
+.. deprecated::
+    Use :data:`GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN` instead.
+"""
 
 GOOGLE_API_USE_MTLS_ENDPOINT = "GOOGLE_API_USE_MTLS_ENDPOINT"
 """Environment variable controlling whether to use mTLS endpoint or not."""
diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py
index 92272c243c3e..285e0bf13656 100644
--- a/packages/google-auth/google/auth/transport/_mtls_helper.py
+++ b/packages/google-auth/google/auth/transport/_mtls_helper.py
@@ -37,7 +37,7 @@
 CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
 _CERT_PROVIDER_COMMAND = "cert_provider_command"
 _CERT_REGEX = re.compile(
-    b"-----BEGIN CERTIFICATE-----.+-----END CERTIFICATE-----\r?\n?", re.DOTALL
+    b"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----\r?\n?", re.DOTALL
 )
 
 # support various format of key files, e.g.
@@ -46,7 +46,7 @@
 # "-----BEGIN RSA PRIVATE KEY-----..."
 # "-----BEGIN ENCRYPTED PRIVATE KEY-----"
 _KEY_REGEX = re.compile(
-    b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+-----END [A-Z ]*PRIVATE KEY-----\r?\n?",
+    b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+?-----END [A-Z ]*PRIVATE KEY-----\r?\n?",
     re.DOTALL,
 )
 
@@ -534,13 +534,16 @@ def _read_cert_file(cert_path):
         cert_data = cert_file.read()
 
     cert_match = re.findall(_CERT_REGEX, cert_data)
-    if len(cert_match) != 1:
+    if not cert_match:
         raise exceptions.ClientCertError(
-            "Certificate file {} is in an invalid format, a single PEM formatted certificate is expected".format(
+            "Certificate file {} is in an invalid format, at least one PEM formatted certificate is expected".format(
                 cert_path
             )
         )
-    return cert_match[0]
+    return b"".join(
+        m if m.endswith(b"\n") or i == len(cert_match) - 1 else m + b"\n"
+        for i, m in enumerate(cert_match)
+    )
 
 
 def _read_key_file(key_path):
@@ -591,8 +594,12 @@ def _run_cert_provider_command(command, expect_encrypted_key=False):
 
     # Extract certificate (chain), key and passphrase.
     cert_match = re.findall(_CERT_REGEX, stdout)
-    if len(cert_match) != 1:
+    if not cert_match:
         raise exceptions.ClientCertError("Client SSL certificate is missing or invalid")
+    cert_chain = b"".join(
+        m if m.endswith(b"\n") or i == len(cert_match) - 1 else m + b"\n"
+        for i, m in enumerate(cert_match)
+    )
     key_match = re.findall(_KEY_REGEX, stdout)
     if len(key_match) != 1:
         raise exceptions.ClientCertError("Client SSL key is missing or invalid")
@@ -603,13 +610,13 @@ def _run_cert_provider_command(command, expect_encrypted_key=False):
             raise exceptions.ClientCertError("Passphrase is missing or invalid")
         if b"ENCRYPTED" not in key_match[0]:
             raise exceptions.ClientCertError("Encrypted private key is expected")
-        return cert_match[0], key_match[0], passphrase_match[0].strip()
+        return cert_chain, key_match[0], passphrase_match[0].strip()
 
     if b"ENCRYPTED" in key_match[0]:
         raise exceptions.ClientCertError("Encrypted private key is not expected")
     if len(passphrase_match) > 0:
         raise exceptions.ClientCertError("Passphrase is not expected")
-    return cert_match[0], key_match[0], None
+    return cert_chain, key_match[0], None
 
 
 def get_client_ssl_credentials(
@@ -847,16 +854,19 @@ def call_client_cert_callback():
     ".mtls.googleapis.com",
     ".mtls.sandbox.googleapis.com",
     ".p.googleapis.com",
+    ".mtls.run.app",
 )
 _MTLS_EXACT_HOSTS = (
     "mtls.googleapis.com",
     "mtls.sandbox.googleapis.com",
     "p.googleapis.com",
+    "mtls.run.app",
 )
 
 
 def is_mtls_endpoint(url: Optional[Union[str, bytes, object]]) -> bool:
-    """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint.
+    """Checks if the given URL corresponds to an mTLS (Google APIs or Cloud Run)
+    or Private Service Connect (PSC) endpoint.
 
     Args:
         url (Optional[Union[str, bytes, object]]): The request URL.
diff --git a/packages/google-auth/tests/compute_engine/test__metadata.py b/packages/google-auth/tests/compute_engine/test__metadata.py
index 0fae4bd6ef16..a3df910c9707 100644
--- a/packages/google-auth/tests/compute_engine/test__metadata.py
+++ b/packages/google-auth/tests/compute_engine/test__metadata.py
@@ -487,6 +487,57 @@ def test_get_failure_bad_json():
     )
 
 
+def test_get_body_with_get_method_raises_value_error():
+    request = make_request("{}")
+
+    with pytest.raises(
+        ValueError, match="Request body cannot be specified with GET method."
+    ):
+        _metadata.get(request, PATH, method="GET", body=b"some_body")
+
+    request.assert_not_called()
+
+
+@mock.patch("time.sleep", return_value=None)
+def test_get_post_retry_preserves_method_body_and_headers(mock_sleep):
+    response_503 = mock.create_autospec(transport.Response, instance=True)
+    response_503.status = http_client.SERVICE_UNAVAILABLE
+    response_503.data = _helpers.to_bytes("Service Unavailable")
+    response_503.headers = {}
+
+    response_ok = mock.create_autospec(transport.Response, instance=True)
+    response_ok.status = http_client.OK
+    response_ok.data = _helpers.to_bytes(
+        json.dumps({"access_token": "bound_token", "expires_in": 3600})
+    )
+    response_ok.headers = {"content-type": "application/json"}
+
+    request = mock.create_autospec(transport.Request)
+    request.side_effect = [
+        response_503,
+        exceptions.TransportError("transient transport error"),
+        response_ok,
+    ]
+
+    expected_body = json.dumps({"certificate_chain": "fake_pem_chain"}).encode("utf-8")
+    result = _metadata.get(
+        request,
+        PATH,
+        method="POST",
+        body=expected_body,
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert result == {"access_token": "bound_token", "expires_in": 3600}
+    assert request.call_count == 3
+    for call_args in request.call_args_list:
+        _, kwargs = call_args
+        assert kwargs["method"] == "POST"
+        assert kwargs["body"] == expected_body
+        assert kwargs["headers"]["Content-Type"] == "application/json"
+        assert kwargs["headers"][_metadata._METADATA_FLAVOR_HEADER] == "Google"
+
+
 def test_get_project_id():
     project = "example-project"
     request = make_request(project, headers={"content-type": "text/plain"})
@@ -639,8 +690,8 @@ def test_get_universe_domain_other_error():
 
 
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate",
-    return_value=None,
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes",
+    return_value=(None, None),
 )
 @mock.patch(
     "google.auth.metrics.token_request_access_token_mds",
@@ -672,8 +723,8 @@ def test_get_service_account_token(
 
 
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate",
-    return_value=None,
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes",
+    return_value=(None, None),
 )
 @mock.patch(
     "google.auth.metrics.token_request_access_token_mds",
@@ -708,8 +759,8 @@ def test_get_service_account_token_with_scopes_list(
 
 
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate",
-    return_value=None,
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes",
+    return_value=(None, None),
 )
 @mock.patch(
     "google.auth.metrics.token_request_access_token_mds",
@@ -743,10 +794,9 @@ def test_get_service_account_token_with_scopes_string(
     assert expiry == utcnow() + datetime.timedelta(seconds=ttl)
 
 
-@mock.patch("google.auth._agent_identity_utils.calculate_certificate_fingerprint")
 @mock.patch("google.auth._agent_identity_utils.should_request_bound_token")
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate"
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
 )
 @mock.patch(
     "google.auth.metrics.token_request_access_token_mds",
@@ -756,38 +806,39 @@ def test_get_service_account_token_with_scopes_string(
 def test_get_service_account_token_with_bound_token(
     utcnow,
     mock_metrics_header_value,
-    mock_get_and_parse,
+    mock_get_cert_and_bytes,
     mock_should_request,
-    mock_calculate_fingerprint,
 ):
     # Test the successful path where a certificate is found and a bound token
     # is requested.
     mock_cert = mock.sentinel.cert
-    mock_get_and_parse.return_value = mock_cert
+    mock_cert_bytes = b"fake_cert_bytes"
+    mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes)
     mock_should_request.return_value = True
-    mock_calculate_fingerprint.return_value = "fake_fingerprint"
 
     token_response = json.dumps({"access_token": "token", "expires_in": 3600})
     request = make_request(token_response, headers={"content-type": "application/json"})
 
     _metadata.get_service_account_token(request)
 
-    mock_get_and_parse.assert_called_once()
+    mock_get_cert_and_bytes.assert_called_once()
     mock_should_request.assert_called_once_with(mock_cert)
-    mock_calculate_fingerprint.assert_called_once_with(mock_cert)
 
     request.assert_called_once()
     _, kwargs = request.call_args
-    url = kwargs["url"]
-    assert "bindCertificateFingerprint=fake_fingerprint" in url
+    assert kwargs["method"] == "POST"
+    assert kwargs["body"] == json.dumps(
+        {"certificate_chain": mock_cert_bytes.decode("utf-8")}
+    ).encode("utf-8")
+    assert kwargs["headers"]["Content-Type"] == "application/json"
 
 
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate"
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
 )
-def test_get_service_account_token_no_cert(mock_get_and_parse):
-    # Test that no fingerprint is added when no certificate is found.
-    mock_get_and_parse.return_value = None
+def test_get_service_account_token_no_cert(mock_get_cert_and_bytes):
+    # Test that a standard GET request with body=None is sent when no certificate is found.
+    mock_get_cert_and_bytes.return_value = (None, None)
     token_response = json.dumps({"access_token": "token", "expires_in": 3600})
     request = make_request(token_response, headers={"content-type": "application/json"})
 
@@ -795,19 +846,19 @@ def test_get_service_account_token_no_cert(mock_get_and_parse):
 
     request.assert_called_once()
     _, kwargs = request.call_args
-    url = kwargs["url"]
-    assert "bindCertificateFingerprint" not in url
+    assert kwargs.get("method", "GET") == "GET"
+    assert kwargs.get("body") is None
 
 
 @mock.patch("google.auth._agent_identity_utils.should_request_bound_token")
 @mock.patch(
-    "google.auth._agent_identity_utils.get_and_parse_agent_identity_certificate"
+    "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
 )
 def test_get_service_account_token_should_not_bind(
-    mock_get_and_parse, mock_should_request
+    mock_get_cert_and_bytes, mock_should_request
 ):
-    # Test that no fingerprint is added when a cert is found but should not be used.
-    mock_get_and_parse.return_value = mock.sentinel.cert
+    # Test that a standard GET request with body=None is sent when a cert is found but should not be used.
+    mock_get_cert_and_bytes.return_value = (mock.sentinel.cert, b"fake_cert_bytes")
     mock_should_request.return_value = False
     token_response = json.dumps({"access_token": "token", "expires_in": 3600})
     request = make_request(token_response, headers={"content-type": "application/json"})
@@ -816,8 +867,8 @@ def test_get_service_account_token_should_not_bind(
 
     request.assert_called_once()
     _, kwargs = request.call_args
-    url = kwargs["url"]
-    assert "bindCertificateFingerprint" not in url
+    assert kwargs.get("method", "GET") == "GET"
+    assert kwargs.get("body") is None
 
 
 def test_get_service_account_info():
diff --git a/packages/google-auth/tests/compute_engine/test_credentials.py b/packages/google-auth/tests/compute_engine/test_credentials.py
index ab171c5a6041..316d4d1b1906 100644
--- a/packages/google-auth/tests/compute_engine/test_credentials.py
+++ b/packages/google-auth/tests/compute_engine/test_credentials.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 import base64
 import datetime
+import json
 import re
 from unittest import mock
 
@@ -22,6 +23,7 @@
 from google.auth import _helpers
 from google.auth import exceptions
 from google.auth import jwt
+from google.auth import metrics
 from google.auth import transport
 from google.auth.compute_engine import credentials
 from google.auth.transport import requests
@@ -475,28 +477,22 @@ def test_regional_access_boundary_disabled_state_transitions(
         assert creds._is_regional_access_boundary_lookup_required() is False
 
     @mock.patch("google.auth.compute_engine._metadata.get")
-    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    @mock.patch("google.auth._agent_identity_utils.parse_certificate")
     @mock.patch(
-        "google.auth._agent_identity_utils.should_request_bound_token",
-        return_value=True,
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
     )
     @mock.patch(
-        "google.auth._agent_identity_utils.calculate_certificate_fingerprint",
-        return_value="fingerprint",
+        "google.auth._agent_identity_utils.should_request_bound_token",
+        return_value=True,
     )
     def test_refresh_with_agent_identity(
         self,
-        mock_calculate_fingerprint,
         mock_should_request,
-        mock_parse_certificate,
-        mock_get_path,
+        mock_get_cert_and_bytes,
         mock_metadata_get,
-        tmpdir,
     ):
-        cert_path = tmpdir.join("cert.pem")
-        cert_path.write(b"cert_content")
-        mock_get_path.return_value = str(cert_path)
+        mock_cert = mock.sentinel.cert
+        mock_cert_bytes = b"cert_content"
+        mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes)
 
         mock_metadata_get.side_effect = [
             {
@@ -509,33 +505,32 @@ def test_refresh_with_agent_identity(
         self.credentials.refresh(None)
 
         assert self.credentials.token == "token"
-        mock_parse_certificate.assert_called_once_with(b"cert_content")
-        mock_should_request.assert_called_once_with(mock_parse_certificate.return_value)
+        mock_get_cert_and_bytes.assert_called_once()
+        mock_should_request.assert_called_once_with(mock_cert)
         kwargs = mock_metadata_get.call_args[1]
         assert kwargs["params"] == {
             "scopes": "one,two",
-            "bindCertificateFingerprint": "fingerprint",
         }
+        assert kwargs["method"] == "POST"
+        assert kwargs["body"] == json.dumps(
+            {"certificate_chain": mock_cert_bytes.decode("utf-8")}
+        ).encode("utf-8")
+        assert kwargs["headers"]["Content-Type"] == "application/json"
 
-    @mock.patch("google.auth.compute_engine._metadata.get")
-    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    @mock.patch("google.auth._agent_identity_utils.parse_certificate")
+    @mock.patch(
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
+    )
     @mock.patch(
         "google.auth._agent_identity_utils.should_request_bound_token",
         return_value=False,
     )
+    @mock.patch("google.auth.compute_engine._metadata.get")
     def test_refresh_with_agent_identity_opt_out_or_not_agent(
         self,
-        mock_should_request,
-        mock_parse_certificate,
-        mock_get_path,
         mock_metadata_get,
-        tmpdir,
+        mock_should_request,
+        mock_get_cert_and_bytes,
     ):
-        cert_path = tmpdir.join("cert.pem")
-        cert_path.write(b"cert_content")
-        mock_get_path.return_value = str(cert_path)
-
         mock_metadata_get.side_effect = [
             {
                 "email": "service-account@project.iam.gserviceaccount.com",
@@ -544,13 +539,42 @@ def test_refresh_with_agent_identity_opt_out_or_not_agent(
             {"access_token": "token", "expires_in": 500},
         ]
 
+        mock_cert = mock.sentinel.cert
+        mock_cert_bytes = b"cert_content"
+        mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes)
+
+        self.credentials.refresh(None)
+
+        assert self.credentials.token == "token"
+        mock_get_cert_and_bytes.assert_called_once()
+        mock_should_request.assert_called_once_with(mock_cert)
+        kwargs = mock_metadata_get.call_args[1]
+        assert kwargs.get("method", "GET") == "GET"
+        assert kwargs.get("body") is None
+
+    @mock.patch(
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
+    )
+    @mock.patch("google.auth.compute_engine._metadata.get")
+    def test_refresh_without_agent_identity_certificate(
+        self,
+        mock_metadata_get,
+        mock_get_cert_and_bytes,
+    ):
+        mock_metadata_get.side_effect = [
+            {"email": "service-account@example.com", "scopes": ["one", "two"]},
+            {"access_token": "token", "expires_in": 500},
+        ]
+
+        mock_get_cert_and_bytes.return_value = (None, None)
+
         self.credentials.refresh(None)
 
         assert self.credentials.token == "token"
-        mock_parse_certificate.assert_called_once_with(b"cert_content")
-        mock_should_request.assert_called_once_with(mock_parse_certificate.return_value)
+        mock_get_cert_and_bytes.assert_called_once()
         kwargs = mock_metadata_get.call_args[1]
-        assert "bindCertificateFingerprint" not in kwargs.get("params", {})
+        assert kwargs.get("method", "GET") == "GET"
+        assert kwargs.get("body") is None
 
     def test_set_blocking_regional_access_boundary_lookup(self):
         creds = self.credentials
@@ -858,6 +882,148 @@ def test_with_target_audience_integration(self):
 
         assert self.credentials.token is not None
 
+    @mock.patch(
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
+    )
+    @mock.patch(
+        "google.auth._agent_identity_utils.should_request_bound_token",
+        return_value=True,
+    )
+    @mock.patch("google.auth.compute_engine._metadata.get")
+    def test_refresh_with_agent_identity(
+        self,
+        mock_metadata_get,
+        mock_should_request,
+        mock_get_cert_and_bytes,
+    ):
+        id_token = "{}.{}.{}".format(
+            base64.b64encode(b'{"some":"some"}').decode("utf-8"),
+            base64.b64encode(b'{"exp": 3210}').decode("utf-8"),
+            base64.b64encode(b"token").decode("utf-8"),
+        )
+        mock_metadata_get.side_effect = [
+            {"email": "service-account@example.com", "scopes": ["one", "two"]},
+            id_token,
+        ]
+
+        mock_cert = mock.sentinel.cert
+        mock_cert_bytes = b"cert_content"
+        mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes)
+
+        request = mock.create_autospec(transport.Request, instance=True)
+        self.credentials = credentials.IDTokenCredentials(
+            request=request,
+            target_audience="https://audience.com",
+            use_metadata_identity_endpoint=True,
+        )
+
+        self.credentials.refresh(None)
+
+        assert self.credentials.token == id_token
+        mock_get_cert_and_bytes.assert_called_once()
+        mock_should_request.assert_called_once_with(mock_cert)
+
+        kwargs = mock_metadata_get.call_args[1]
+        assert kwargs["method"] == "POST"
+        assert kwargs["body"] == json.dumps(
+            {"certificate_chain": mock_cert_bytes.decode("utf-8")}
+        ).encode("utf-8")
+        assert kwargs["headers"]["Content-Type"] == "application/json"
+        assert (
+            kwargs["headers"][metrics.API_CLIENT_HEADER]
+            == metrics.token_request_id_token_mds()
+        )
+
+    @mock.patch(
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
+    )
+    @mock.patch(
+        "google.auth._agent_identity_utils.should_request_bound_token",
+        return_value=False,
+    )
+    @mock.patch("google.auth.compute_engine._metadata.get")
+    def test_refresh_with_agent_identity_opt_out_or_not_agent(
+        self,
+        mock_metadata_get,
+        mock_should_request,
+        mock_get_cert_and_bytes,
+    ):
+        id_token = "{}.{}.{}".format(
+            base64.b64encode(b'{"some":"some"}').decode("utf-8"),
+            base64.b64encode(b'{"exp": 3210}').decode("utf-8"),
+            base64.b64encode(b"token").decode("utf-8"),
+        )
+        mock_metadata_get.side_effect = [
+            {"email": "service-account@example.com", "scopes": ["one", "two"]},
+            id_token,
+        ]
+
+        mock_cert = mock.sentinel.cert
+        mock_cert_bytes = b"cert_content"
+        mock_get_cert_and_bytes.return_value = (mock_cert, mock_cert_bytes)
+
+        request = mock.create_autospec(transport.Request, instance=True)
+        self.credentials = credentials.IDTokenCredentials(
+            request=request,
+            target_audience="https://audience.com",
+            use_metadata_identity_endpoint=True,
+        )
+
+        self.credentials.refresh(None)
+
+        assert self.credentials.token == id_token
+        mock_get_cert_and_bytes.assert_called_once()
+        mock_should_request.assert_called_once_with(mock_cert)
+
+        kwargs = mock_metadata_get.call_args[1]
+        assert kwargs["method"] == "GET"
+        assert kwargs["body"] is None
+        assert (
+            kwargs["headers"][metrics.API_CLIENT_HEADER]
+            == metrics.token_request_id_token_mds()
+        )
+
+    @mock.patch(
+        "google.auth._agent_identity_utils.get_agent_identity_certificate_and_bytes"
+    )
+    @mock.patch("google.auth.compute_engine._metadata.get")
+    def test_refresh_without_agent_identity_certificate(
+        self,
+        mock_metadata_get,
+        mock_get_cert_and_bytes,
+    ):
+        id_token = "{}.{}.{}".format(
+            base64.b64encode(b'{"some":"some"}').decode("utf-8"),
+            base64.b64encode(b'{"exp": 3210}').decode("utf-8"),
+            base64.b64encode(b"token").decode("utf-8"),
+        )
+        mock_metadata_get.side_effect = [
+            {"email": "service-account@example.com", "scopes": ["one", "two"]},
+            id_token,
+        ]
+
+        mock_get_cert_and_bytes.return_value = (None, None)
+
+        request = mock.create_autospec(transport.Request, instance=True)
+        self.credentials = credentials.IDTokenCredentials(
+            request=request,
+            target_audience="https://audience.com",
+            use_metadata_identity_endpoint=True,
+        )
+
+        self.credentials.refresh(None)
+
+        assert self.credentials.token == id_token
+        mock_get_cert_and_bytes.assert_called_once()
+
+        kwargs = mock_metadata_get.call_args[1]
+        assert kwargs["method"] == "GET"
+        assert kwargs["body"] is None
+        assert (
+            kwargs["headers"][metrics.API_CLIENT_HEADER]
+            == metrics.token_request_id_token_mds()
+        )
+
     @mock.patch(
         "google.auth._helpers.utcnow",
         return_value=_helpers.utcfromtimestamp(0),
diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py
index 6b830e048412..0feb940d5258 100644
--- a/packages/google-auth/tests/test_agent_identity_utils.py
+++ b/packages/google-auth/tests/test_agent_identity_utils.py
@@ -48,9 +48,37 @@
 )
 
 
+# A mock PEM-encoded certificate with a valid Agent Identity SPIFFE ID.
+AGENT_IDENTITY_CERT_BYTES = (
+    b"-----BEGIN CERTIFICATE-----\n"
+    b"MIIDEjCCAfqgAwIBAgIUKZAXnXnxf8hsn+ojS1N8bN3hXrUwDQYJKoZIhvcNAQEL\n"
+    b"BQAwHjEcMBoGA1UEAwwTYWdlbnQtaWRlbnRpdHktdGVzdDAeFw0yNDAxMDEwMDAw\n"
+    b"MDBaFw0zNDAxMDEwMDAwMDBaMB4xHDAaBgNVBAMME2FnZW50LWlkZW50aXR5LXRl\n"
+    b"c3QwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQC+3eSHkp1oBj7rFehL\n"
+    b"5VJmBF9KLZ5PXQuZNYrGSpxGJ0Dx1T5ancrl8e66AfAepw9O4zdA+8Afub39PQLh\n"
+    b"wMTEY3O9Uqetch+2apkwXQ+yYpnorgMwqykY77ptApA8WPHzEOj58FPtyC4UqXJ7\n"
+    b"YKVpN92lVi1l73XBn6axo/q72KjeEdssR6UMtAd3dbGqY3af/AZNppJWRmCMWs8Z\n"
+    b"oAuxTH5LqYuxwvCDfYLpmQSbv4IJ/UBjkvjRIlzPo2zHg9PMdf/j6Bg9n3kkFaVH\n"
+    b"Oep+Zm+DtdT8JvwnG3sQ8Qn/ZCqU0z3DkT//XaElAikLwr/1MFrVHhfrYEWnDvuy\n"
+    b"9PHxAgMBAAGjSDBGMEQGA1UdEQQ9MDuGOXNwaWZmZTovL2FnZW50cy5nbG9iYWwu\n"
+    b"cHJvai0xMjM0NS5zeXN0ZW0uaWQuZ29vZy93b3JrbG9hZDANBgkqhkiG9w0BAQsF\n"
+    b"AAOCAQEAWEyBk7TbetWeQTYEdJwH/pNmiqoCzDYcqCSuNqJhrItHuLmSAlKBGCz6\n"
+    b"I6ptzY6vT7ARXoW07ivf9Ffl3TMUDLjd5Tkfn1q8JjyM1Ugbfuq7rdF2g9+5h6wg\n"
+    b"tjeV10LqAimr+fFaNvRiGsMfokuwPyUKYe/9d6x5NhcTTNgMQDG5SWnRe1JqPy94\n"
+    b"GKilWCyzDl4qzHAU5gc7lZ/6WKbYPwjJDDT4/d3AvNx1O/cQCG7Mz4veDuG2Jqh+\n"
+    b"FPUqQ4G9RL4zdPuXlbKfSknkmZWld1+adyitai6BzDCG9zkEEVJmLE2/e3XvNC93\n"
+    b"fa2asspu5y/ViCmPS0J2rzWEk7zI5w==\n"
+    b"-----END CERTIFICATE-----\n"
+)
+
+
 class TestAgentIdentityUtils:
     @pytest.fixture(autouse=True)
     def clean_env(self, monkeypatch):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG,
+            raising=False,
+        )
         monkeypatch.delenv(
             environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
             raising=False,
@@ -59,17 +87,79 @@ def clean_env(self, monkeypatch):
             environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE,
             raising=False,
         )
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            raising=False,
+        )
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            raising=False,
+        )
 
     @mock.patch("cryptography.x509.load_pem_x509_certificate")
     def test_parse_certificate(self, mock_load_cert):
+        mock_load_cert.return_value = mock.sentinel.cert
         result = _agent_identity_utils.parse_certificate(b"cert_bytes")
         mock_load_cert.assert_called_once_with(b"cert_bytes")
-        assert result == mock_load_cert.return_value
+        assert result == mock.sentinel.cert
+
+    def test_parse_certificate_empty_bytes_raises_value_error(self):
+        with pytest.raises(ValueError):
+            _agent_identity_utils.parse_certificate(b"")
+
+    @pytest.mark.parametrize(
+        "second_cert_block",
+        [
+            # Valid Base64, invalid ASN.1 DER
+            b"-----BEGIN CERTIFICATE-----\n"
+            + base64.b64encode(b"not valid asn1 der payload")
+            + b"\n-----END CERTIFICATE-----\n",
+            # Corrupted Base64
+            b"-----BEGIN CERTIFICATE-----\n!!!not_base64!!!\n-----END CERTIFICATE-----\n",
+            # Non-UTF-8 bytes
+            b"-----BEGIN CERTIFICATE-----\n\xff\xfe\xfd\n-----END CERTIFICATE-----\n",
+        ],
+    )
+    def test_parse_certificate_full_chain_rejects_malformed_intermediate(
+        self, second_cert_block, monkeypatch
+    ):
+        monkeypatch.delattr(x509, "load_pem_x509_certificates", raising=False)
+        chain_bytes = NON_AGENT_IDENTITY_CERT_BYTES + second_cert_block
+        with pytest.raises(ValueError):
+            _agent_identity_utils.parse_certificate(chain_bytes)
 
     def test_is_certificate_file_ready_empty_path(self):
         result = _agent_identity_utils._is_certificate_file_ready("")
         assert result is False
 
+    def test_is_in_well_known_dir_empty_path(self):
+        assert _agent_identity_utils._is_in_well_known_dir("") is False
+
+    def test_is_in_well_known_dir_resolves_symlinks(self, tmpdir, monkeypatch):
+        real_run_dir = tmpdir.mkdir("run")
+        var_dir = tmpdir.mkdir("var")
+        symlink_var_run = var_dir.join("run")
+        os.symlink(str(real_run_dir), str(symlink_var_run))
+
+        well_known_via_symlink = os.path.join(
+            str(symlink_var_run),
+            "secrets",
+            "workload-spiffe-credentials",
+            "certificates.pem",
+        )
+        monkeypatch.setattr(
+            "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH",
+            well_known_via_symlink,
+        )
+
+        resolved_cert_path = os.path.join(
+            str(real_run_dir),
+            "secrets",
+            "workload-spiffe-credentials",
+            "certificates.pem",
+        )
+        assert _agent_identity_utils._is_in_well_known_dir(resolved_cert_path) is True
+
     def test_get_agent_identity_certificate_path_empty_env(self, monkeypatch):
         monkeypatch.delenv(
             environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
@@ -207,34 +297,94 @@ def test_calculate_certificate_fingerprint(self):
 
         assert fingerprint == expected_fingerprint
 
-    @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate")
-    def test_should_request_bound_token(self, mock_is_agent, monkeypatch):
-        # Agent cert, default env var (opt-in)
-        mock_is_agent.return_value = True
+    def test_is_bound_token_opted_out(self, monkeypatch):
+        # Default (both unset) -> not opted out
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            raising=False,
+        )
         monkeypatch.delenv(
             environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
             raising=False,
         )
-        assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+        assert not _agent_identity_utils._is_bound_token_opted_out()
 
-        # Agent cert, explicit opt-in
+        # Explicit opt-in / non-false values on primary -> not opted out
+        for val in ("true", "TRUE", "1", "invalid", ""):
+            monkeypatch.setenv(
+                environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+                val,
+            )
+            assert not _agent_identity_utils._is_bound_token_opted_out()
+
+        # Explicit opt-out via primary -> opted out
+        for val in ("false", "FALSE"):
+            monkeypatch.setenv(
+                environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+                val,
+            )
+            assert _agent_identity_utils._is_bound_token_opted_out()
+
+        # Primary overrides fallback (primary false, fallback true -> opted out)
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "false",
+        )
         monkeypatch.setenv(
             environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
             "true",
         )
-        assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+        assert _agent_identity_utils._is_bound_token_opted_out()
 
-        # Agent cert, explicit opt-out
+        # Primary overrides fallback (primary true, fallback false -> not opted out)
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "true",
+        )
         monkeypatch.setenv(
             environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
             "false",
         )
+        assert not _agent_identity_utils._is_bound_token_opted_out()
+
+        # Fallback when primary is unset
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            raising=False,
+        )
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            "false",
+        )
+        assert _agent_identity_utils._is_bound_token_opted_out()
+
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            "true",
+        )
+        assert not _agent_identity_utils._is_bound_token_opted_out()
+
+    @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate")
+    def test_should_request_bound_token(self, mock_is_agent, monkeypatch):
+        # Agent cert, opted in
+        mock_is_agent.return_value = True
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "true",
+        )
+        assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+
+        # Agent cert, opted out
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "false",
+        )
         assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
 
-        # Non-agent cert, opt-in
+        # Non-agent cert, opted in
         mock_is_agent.return_value = False
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
             "true",
         )
         assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
@@ -381,9 +531,8 @@ def test_get_agent_identity_certificate_path_failure(
             _agent_identity_utils.get_agent_identity_certificate_path()
 
         assert "not found after multiple retries" in str(excinfo.value)
-        assert (
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
-            in str(excinfo.value)
+        assert environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN in str(
+            excinfo.value
         )
         assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
@@ -412,9 +561,8 @@ def test_get_agent_identity_certificate_path_fail_fast_config_missing(
         mock_sleep.assert_not_called()
 
     @mock.patch("time.sleep")
-    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_fail_fast_cert_missing(
-        self, mock_exists, mock_sleep, tmpdir, monkeypatch
+        self, mock_sleep, tmpdir, monkeypatch
     ):
         # Simulate config path outside well-known dir where config is valid but cert is missing.
         well_known_path = tmpdir.mkdir("well_known_cert").join("certificates.pem")
@@ -434,20 +582,14 @@ def test_get_agent_identity_certificate_path_fail_fast_cert_missing(
             environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
         )
 
-        def exists_side_effect(path):
-            return path == str(config_path)
-
-        mock_exists.side_effect = exists_side_effect
-
         result = _agent_identity_utils.get_agent_identity_certificate_path()
 
         assert result is None
         mock_sleep.assert_not_called()
 
     @mock.patch("time.sleep")
-    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_cert_not_found(
-        self, mock_exists, mock_sleep, tmpdir, monkeypatch
+        self, mock_sleep, tmpdir, monkeypatch
     ):
         monkeypatch.setattr(
             "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH",
@@ -462,11 +604,6 @@ def test_get_agent_identity_certificate_path_cert_not_found(
             environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
         )
 
-        def exists_side_effect(path):
-            return path == str(config_path)
-
-        mock_exists.side_effect = exists_side_effect
-
         with pytest.raises(exceptions.RefreshError):
             _agent_identity_utils.get_agent_identity_certificate_path()
 
@@ -589,89 +726,250 @@ def test_get_agent_identity_certificate_path_permission_error_cert_file(
         mock_sleep.assert_not_called()
 
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_opted_out(
-        self, mock_get_path, monkeypatch
+    def test_get_agent_identity_certificate_and_bytes_success(
+        self, mock_get_path, tmpdir, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
-            "false",
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "true",
         )
-        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
-        assert result is None
-        mock_get_path.assert_not_called()
+        cert_file = tmpdir.join("cert.pem")
+        cert_file.write_binary(NON_AGENT_IDENTITY_CERT_BYTES)
+        mock_get_path.return_value = str(cert_file)
+
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+
+        assert isinstance(cert, x509.Certificate)
+        assert cert_bytes == NON_AGENT_IDENTITY_CERT_BYTES
 
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_no_path(
-        self, mock_get_path, monkeypatch
+    def test_get_agent_identity_certificate_and_bytes_combined_bundle(
+        self, mock_get_path, tmpdir, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
             "true",
         )
-        mock_get_path.return_value = None
-        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
-        assert result is None
-        mock_get_path.assert_called_once()
+        non_utf8_bag_attrs = b"Bag Attributes\n    friendlyName: \xff\xfe\n"
+        private_key_pem = (
+            b"-----BEGIN PRIVATE KEY-----\n"
+            b"MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC3\n"
+            b"-----END PRIVATE KEY-----\n"
+        )
+        combined_bundle = (
+            non_utf8_bag_attrs
+            + NON_AGENT_IDENTITY_CERT_BYTES.rstrip(b"\n")
+            + b"   \n"
+            + private_key_pem
+            + NON_AGENT_IDENTITY_CERT_BYTES
+        )
+        cert_file = tmpdir.join("credentialbundle.pem")
+        cert_file.write_binary(combined_bundle)
+        mock_get_path.return_value = str(cert_file)
+
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+
+        expected_certs = NON_AGENT_IDENTITY_CERT_BYTES + NON_AGENT_IDENTITY_CERT_BYTES
+        assert isinstance(cert, x509.Certificate)
+        assert cert_bytes == expected_certs
+        assert b"PRIVATE KEY" not in cert_bytes
+        assert cert_bytes.decode("utf-8") == expected_certs.decode("utf-8")
 
-    @mock.patch("google.auth._agent_identity_utils.parse_certificate")
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_success(
-        self, mock_get_path, mock_parse_certificate, monkeypatch
+    def test_get_agent_identity_certificate_and_bytes_no_cert_blocks(
+        self, mock_get_path, tmpdir, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
             "true",
         )
-        mock_get_path.return_value = "/fake/cert.pem"
-        mock_open = mock.mock_open(read_data=b"cert_bytes")
+        cert_file = tmpdir.join("empty_or_key_only.pem")
+        cert_file.write_binary(
+            b"-----BEGIN PRIVATE KEY-----\nMIIB\n-----END PRIVATE KEY-----\n"
+        )
+        mock_get_path.return_value = str(cert_file)
 
-        with mock.patch("builtins.open", mock_open):
-            result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+        with pytest.warns(UserWarning, match="No PEM certificate blocks found"):
+            (
+                cert,
+                cert_bytes,
+            ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
 
-        mock_open.assert_called_once_with("/fake/cert.pem", "rb")
-        mock_parse_certificate.assert_called_once_with(b"cert_bytes")
-        assert result == mock_parse_certificate.return_value
+        assert cert is None
+        assert cert_bytes is None
 
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_use_client_cert_false(
-        self, mock_get_path, monkeypatch
+    def test_get_agent_identity_certificate_and_bytes_os_error(
+        self, mock_get_path, tmpdir, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
-            "false",
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "true",
         )
-        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+        missing_cert_file = tmpdir.join("deleted_during_rotation.pem")
+        mock_get_path.return_value = str(missing_cert_file)
+
+        with pytest.warns(
+            UserWarning, match="Failed to read agent identity certificate file"
+        ):
+            (
+                cert,
+                cert_bytes,
+            ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+
+        assert cert is None
+        assert cert_bytes is None
+
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_agent_identity_certificate_and_bytes_corrupt_cert_value_error(
+        self, mock_get_path, tmpdir, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "true",
+        )
+        cert_file = tmpdir.join("corrupt_cert.pem")
+        cert_file.write_binary(
+            b"-----BEGIN CERTIFICATE-----\nnot_valid_base64_or_der\n-----END CERTIFICATE-----\n"
+        )
+        mock_get_path.return_value = str(cert_file)
+
+        with pytest.warns(
+            UserWarning, match="Failed to parse agent identity certificate"
+        ):
+            (
+                cert,
+                cert_bytes,
+            ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+
+        assert cert is None
+        assert cert_bytes is None
+
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_agent_identity_certificate_and_bytes_corrupt_intermediate_warns(
+        self, mock_get_path, tmpdir
+    ):
+        corrupt_intermediate = (
+            b"-----BEGIN CERTIFICATE-----\n"
+            + base64.b64encode(b"invalid_asn1_der_intermediate")
+            + b"\n-----END CERTIFICATE-----\n"
+        )
+        cert_file = tmpdir.join("chain_with_corrupt_intermediate.pem")
+        cert_file.write_binary(AGENT_IDENTITY_CERT_BYTES + corrupt_intermediate)
+        mock_get_path.return_value = str(cert_file)
+
+        with pytest.warns(
+            UserWarning, match="Failed to parse agent identity certificate"
+        ):
+            (
+                cert,
+                cert_bytes,
+            ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+
+        assert cert is None
+        assert cert_bytes is None
+
+    @mock.patch("time.sleep")
+    def test_get_agent_identity_certificate_path_well_known_config_external_missing_cert_no_poll(
+        self, mock_sleep, tmpdir, monkeypatch
+    ):
+        well_known_dir = tmpdir.mkdir("workload-spiffe-credentials")
+        external_dir = tmpdir.mkdir("external_certs")
+        monkeypatch.setattr(
+            "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH",
+            str(well_known_dir.join("certificates.pem")),
+        )
+        config_path = well_known_dir.join("config.json")
+        config_path.write(
+            json.dumps(
+                {
+                    "cert_configs": {
+                        "workload": {
+                            "cert_path": str(external_dir.join("missing_cert.pem"))
+                        }
+                    }
+                }
+            )
+        )
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
         assert result is None
-        mock_get_path.assert_not_called()
+        mock_sleep.assert_not_called()
 
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_use_client_cert_invalid(
+    def test_get_agent_identity_certificate_and_bytes_opted_out(
         self, mock_get_path, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
-            "foo",
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
+            "false",
         )
-        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
-        assert result is None
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+        assert cert is None
+        assert cert_bytes is None
         mock_get_path.assert_not_called()
 
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
-    def test_get_and_parse_agent_identity_certificate_file_read_error(
+    def test_get_agent_identity_certificate_and_bytes_no_path(
         self, mock_get_path, monkeypatch
     ):
         monkeypatch.setenv(
-            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN,
             "true",
         )
-        mock_get_path.return_value = "/fake/cert.pem"
-        mock_open = mock.mock_open()
-        mock_open.side_effect = PermissionError("Permission denied")
+        mock_get_path.return_value = None
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+        assert cert is None
+        assert cert_bytes is None
+        mock_get_path.assert_called_once()
 
-        with mock.patch("builtins.open", mock_open):
-            result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_agent_identity_certificate_and_bytes_use_client_cert_false(
+        self, mock_get_path, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "false",
+        )
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+        assert cert is None
+        assert cert_bytes is None
+        mock_get_path.assert_not_called()
 
-        assert result is None
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_agent_identity_certificate_and_bytes_use_client_cert_invalid(
+        self, mock_get_path, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "foo",
+        )
+        (
+            cert,
+            cert_bytes,
+        ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+        assert cert is None
+        assert cert_bytes is None
+        mock_get_path.assert_not_called()
 
     def test_get_cached_cert_fingerprint_no_cert(self):
         with pytest.raises(ValueError, match="mTLS connection is not configured."):
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index f6f1185a660e..a0b75a208186 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -533,7 +533,14 @@ async def test_psc_endpoint_triggers_cert_rotation(self):
         await session.close()
 
     @pytest.mark.asyncio
-    async def test_non_mtls_url_bypasses_rotation(self):
+    @pytest.mark.parametrize(
+        "non_mtls_url",
+        [
+            "https://pubsub.googleapis.com/test",
+            "https://my-service-xyz-uc.a.run.app/test",
+        ],
+    )
+    async def test_non_mtls_url_bypasses_rotation(self, non_mtls_url):
         """Verifies that standard non-mTLS URLs bypass certificate rotation."""
         mock_creds = mock.AsyncMock(spec=credentials.Credentials)
         mock_creds.before_request = mock.AsyncMock(return_value=None)
@@ -564,7 +571,7 @@ async def test_non_mtls_url_bypasses_rotation(self):
                 session, "configure_mtls_channel", new_callable=mock.AsyncMock
             ) as mock_conf,
         ):
-            resp = await session.request("GET", "https://pubsub.googleapis.com/test")
+            resp = await session.request("GET", non_mtls_url)
 
             assert resp == mock_resp_200
             mock_check.assert_not_called()
@@ -856,7 +863,14 @@ async def slow_refresh(*args, **kwargs):
         await session.close()
 
     @pytest.mark.asyncio
-    async def test_cert_rotation_with_completed_mtls_init_task(self):
+    @pytest.mark.parametrize(
+        "mtls_url",
+        [
+            "https://pubsub.mtls.googleapis.com/test",
+            "https://my-service-123456.us-central1.mtls.run.app/test",
+        ],
+    )
+    async def test_cert_rotation_with_completed_mtls_init_task(self, mtls_url):
         """
         Verifies that when _mtls_init_task is already completed, receiving a 401
         with rotated certificates properly resets _mtls_init_task and reconfigures mTLS.
@@ -898,10 +912,8 @@ async def dummy_completed():
             ) as mock_conf,
         ):
             mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp")
-            # Must use a hostname matching _MTLS_URL_PREFIXES (e.g. *.mtls.googleapis.com)
-            resp = await session.request(
-                "GET", "https://pubsub.mtls.googleapis.com/test"
-            )
+            # Must use a hostname matching _mtls_helper.is_mtls_endpoint
+            resp = await session.request("GET", mtls_url)
             assert resp == mock_resp_200
             mock_conf.assert_called_once()
             # Verify the previous completed task was cleared during rotation
diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py
index f1d096ff64cd..362d826e8371 100644
--- a/packages/google-auth/tests/transport/test__mtls_helper.py
+++ b/packages/google-auth/tests/transport/test__mtls_helper.py
@@ -56,7 +56,7 @@ def check_cert_and_key(content, expected_cert, expected_key):
     success = True
 
     cert_match = re.findall(_mtls_helper._CERT_REGEX, content)
-    success = success and len(cert_match) == 1 and cert_match[0] == expected_cert
+    success = success and len(cert_match) >= 1 and b"".join(cert_match) == expected_cert
 
     key_match = re.findall(_mtls_helper._KEY_REGEX, content)
     success = success and len(key_match) == 1 and key_match[0] == expected_key
@@ -67,32 +67,40 @@ def check_cert_and_key(content, expected_cert, expected_key):
 class TestCertAndKeyRegex(object):
     def test_cert_and_key(self):
         # Test single cert and single key
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.public_cert_bytes + pytest.private_key_bytes,
             pytest.public_cert_bytes,
             pytest.private_key_bytes,
         )
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.private_key_bytes + pytest.public_cert_bytes,
             pytest.public_cert_bytes,
             pytest.private_key_bytes,
         )
 
         # Test cert chain and single key
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.public_cert_bytes
             + pytest.public_cert_bytes
             + pytest.private_key_bytes,
             pytest.public_cert_bytes + pytest.public_cert_bytes,
             pytest.private_key_bytes,
         )
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.private_key_bytes
             + pytest.public_cert_bytes
             + pytest.public_cert_bytes,
             pytest.public_cert_bytes + pytest.public_cert_bytes,
             pytest.private_key_bytes,
         )
+        # Test interleaved key between certificates in a combined bundle
+        assert check_cert_and_key(
+            pytest.public_cert_bytes
+            + pytest.private_key_bytes
+            + pytest.public_cert_bytes,
+            pytest.public_cert_bytes + pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
 
     def test_key(self):
         # Create some fake keys for regex check.
@@ -109,13 +117,13 @@ def test_key(self):
         /fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQAB
         -----END EC PRIVATE KEY-----"""
 
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.public_cert_bytes + KEY, pytest.public_cert_bytes, KEY
         )
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.public_cert_bytes + RSA_KEY, pytest.public_cert_bytes, RSA_KEY
         )
-        check_cert_and_key(
+        assert check_cert_and_key(
             pytest.public_cert_bytes + EC_KEY, pytest.public_cert_bytes, EC_KEY
         )
 
@@ -795,6 +803,26 @@ def test_invalid_key_file(self):
         with pytest.raises(exceptions.ClientCertError):
             _mtls_helper._read_cert_and_key_files(cert_path, key_path)
 
+    def test_combined_bundle_with_interleaved_key(self, tmp_path):
+        bundle_file = tmp_path / "credentialbundle.pem"
+        bundle_file.write_bytes(
+            pytest.public_cert_bytes
+            + pytest.private_key_bytes
+            + pytest.public_cert_bytes
+        )
+        actual_cert, actual_key = _mtls_helper._read_cert_and_key_files(
+            str(bundle_file), str(bundle_file)
+        )
+        assert actual_cert == pytest.public_cert_bytes + pytest.public_cert_bytes
+        assert actual_key == pytest.private_key_bytes
+
+    def test_multiple_keys_raises_error(self, tmp_path):
+        cert_path = os.path.join(pytest.data_dir, "public_cert.pem")
+        key_file = tmp_path / "two_keys.pem"
+        key_file.write_bytes(pytest.private_key_bytes + pytest.private_key_bytes)
+        with pytest.raises(exceptions.ClientCertError):
+            _mtls_helper._read_cert_and_key_files(cert_path, str(key_file))
+
 
 class TestGetCertConfigPath(object):
     def test_success_with_override(self):
@@ -1927,6 +1955,12 @@ class TestIsMtlsEndpoint(object):
             "https://p.googleapis.com/",
             "https://p.googleapis.com:443/v1",
             "https://p.googleapis.com.",
+            "https://mtls.run.app",
+            "https://mtls.run.app/",
+            "https://my-service-123456.us-central1.mtls.run.app",
+            "https://my-service-123456.us-central1.mtls.run.app/v1/invocations",
+            "https://tag---my-service-123456.us-central1.mtls.run.app.",
+            b"https://my-service-123456.us-central1.mtls.run.app",
         ],
     )
     def test_is_mtls_endpoint_true(self, url):
@@ -1940,6 +1974,11 @@ def test_is_mtls_endpoint_true(self, url):
             "https://storage.googleapis.com:443/b/my-bucket",
             "https://storage.googleapis.com:443/bucket/mtls.googleapis.com?pageSize=10#frag",
             "https://storage.googleapis.com/bucket/mtls.googleapis.com",
+            "https://my-service-xyz-uc.a.run.app",
+            "https://my-service-123456.us-central1.run.app",
+            "https://my-service-xyz-uc.a.run.app/mtls.run.app",
+            "https://fake-mtls.run.app/v1",
+            "https://fake-mtls.run.app.attacker.com/v1",
             "https://[2001:db8::1]:443/mtls.googleapis.com",
             "https://[::1]:8443/mtls.googleapis.com",
             "https://logging.googleapis.com/v2/entries?filter=mtls.googleapis.com",
@@ -1958,6 +1997,7 @@ def test_is_mtls_endpoint_true(self, url):
                 "https://storage.googleapis.com/bucket/mtls.googleapis.com"
             ),
             "https://.",
+            "https://[::1",
             "",
             None,
             123,
diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py
index c106a87f08fb..af74c0661a4a 100644
--- a/packages/google-auth/tests/transport/test_requests.py
+++ b/packages/google-auth/tests/transport/test_requests.py
@@ -940,7 +940,14 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self):
             # Assert mTLS check logic was SKIPPED (Inner Check was False)
             assert not mock_helper.check_parameters_for_unauthorized_response.called
 
-    def test_cert_rotation_skipped_on_non_mtls_url(self):
+    @pytest.mark.parametrize(
+        "non_mtls_url",
+        [
+            "https://storage.googleapis.com/bucket/mtls.googleapis.com",
+            "https://my-service-xyz-uc.a.run.app/mtls.run.app",
+        ],
+    )
+    def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url):
         """
         Tests that mTLS cert rotation is skipped on non-mTLS URLs even if
         mTLS is enabled and an UNAUTHORIZED (401) response is received.
@@ -953,7 +960,6 @@ def test_cert_rotation_skipped_on_non_mtls_url(self):
                 make_response(status=http_client.OK),
             ]
         )
-        non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com"
         authed_session = google.auth.transport.requests.AuthorizedSession(
             credentials, refresh_timeout=60
         )
@@ -973,10 +979,18 @@ def test_cert_rotation_skipped_on_non_mtls_url(self):
             # Assert mTLS check logic was SKIPPED
             assert not mock_check_params.called
 
-    def test_cert_rotation_triggered_on_psc_url(self):
+    @pytest.mark.parametrize(
+        "mtls_url",
+        [
+            "https://storage.p.googleapis.com/b/my-bucket",
+            "https://my-service-123456.us-central1.mtls.run.app/v1",
+        ],
+    )
+    def test_cert_rotation_triggered_on_psc_url(self, mtls_url):
         """
-        Tests that mTLS cert rotation IS triggered on a Private Service Connect
-        (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received.
+        Tests that mTLS cert rotation IS triggered on Private Service Connect
+        (PSC) and Cloud Run mTLS endpoints when an UNAUTHORIZED (401) response
+        is received.
         """
         credentials = mock.Mock(wraps=CredentialsStub())
         adapter = AdapterStub(
@@ -985,11 +999,10 @@ def test_cert_rotation_triggered_on_psc_url(self):
                 make_response(status=http_client.OK),
             ]
         )
-        psc_url = "https://storage.p.googleapis.com/b/my-bucket"
         authed_session = google.auth.transport.requests.AuthorizedSession(
             credentials, refresh_timeout=60
         )
-        authed_session.mount(psc_url, adapter)
+        authed_session.mount(mtls_url, adapter)
         authed_session._is_mtls = True
         authed_session._cached_cert = b"cached_cert"
 
@@ -998,9 +1011,9 @@ def test_cert_rotation_triggered_on_psc_url(self):
             "check_parameters_for_unauthorized_response",
             return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"),
         ) as mock_check_params:
-            authed_session.request("GET", psc_url)
+            authed_session.request("GET", mtls_url)
 
-            # Assert mTLS check logic was called on PSC endpoint
+            # Assert mTLS check logic was called on PSC / Cloud Run mTLS endpoint
             mock_check_params.assert_called_once()
             assert credentials.refresh.called
 
diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py
index 0fbee087e11f..1d9dd45a3c88 100644
--- a/packages/google-auth/tests/transport/test_urllib3.py
+++ b/packages/google-auth/tests/transport/test_urllib3.py
@@ -658,7 +658,14 @@ def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self):
             # Assert mTLS check logic was SKIPPED (Inner Check was False)
             assert not mock_helper.check_parameters_for_unauthorized_response.called
 
-    def test_cert_rotation_skipped_on_non_mtls_url(self):
+    @pytest.mark.parametrize(
+        "non_mtls_url",
+        [
+            "https://storage.googleapis.com/bucket/mtls.googleapis.com",
+            "https://my-service-xyz-uc.a.run.app/mtls.run.app",
+        ],
+    )
+    def test_cert_rotation_skipped_on_non_mtls_url(self, non_mtls_url):
         """
         Tests that mTLS cert rotation is skipped on non-mTLS URLs even if
         mTLS is enabled and an UNAUTHORIZED (401) response is received.
@@ -670,7 +677,6 @@ def test_cert_rotation_skipped_on_non_mtls_url(self):
                 ResponseStub(status=http_client.OK),
             ]
         )
-        non_mtls_url = "https://storage.googleapis.com/bucket/mtls.googleapis.com"
         authed_http = google.auth.transport.urllib3.AuthorizedHttp(
             credentials, http=http
         )
@@ -689,10 +695,18 @@ def test_cert_rotation_skipped_on_non_mtls_url(self):
             # Assert mTLS check logic was SKIPPED
             assert not mock_check_params.called
 
-    def test_cert_rotation_triggered_on_psc_url(self):
+    @pytest.mark.parametrize(
+        "mtls_url",
+        [
+            "https://storage.p.googleapis.com/b/my-bucket",
+            "https://my-service-123456.us-central1.mtls.run.app/v1",
+        ],
+    )
+    def test_cert_rotation_triggered_on_psc_url(self, mtls_url):
         """
-        Tests that mTLS cert rotation IS triggered on a Private Service Connect
-        (PSC) mTLS endpoint when an UNAUTHORIZED (401) response is received.
+        Tests that mTLS cert rotation IS triggered on Private Service Connect
+        (PSC) and Cloud Run mTLS endpoints when an UNAUTHORIZED (401) response
+        is received.
         """
         credentials = mock.Mock(wraps=CredentialsStub())
         http = HttpStub(
@@ -701,7 +715,6 @@ def test_cert_rotation_triggered_on_psc_url(self):
                 ResponseStub(status=http_client.OK),
             ]
         )
-        psc_url = "https://storage.p.googleapis.com/b/my-bucket"
         authed_http = google.auth.transport.urllib3.AuthorizedHttp(
             credentials, http=http
         )
@@ -713,9 +726,9 @@ def test_cert_rotation_triggered_on_psc_url(self):
             "check_parameters_for_unauthorized_response",
             return_value=(b"new_cert", b"new_key", "old_fp", "old_fp"),
         ) as mock_check_params:
-            authed_http.urlopen("GET", psc_url)
+            authed_http.urlopen("GET", mtls_url)
 
-            # Assert mTLS check logic was called on PSC endpoint
+            # Assert mTLS check logic was called on PSC / Cloud Run mTLS endpoint
             mock_check_params.assert_called_once()
             assert credentials.refresh.called