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

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

@@ -221,14 +221,14 @@ def _parse_cert_path_from_config(cert_config_path):
     return workload_config["cert_path"]
 
 
-def get_and_parse_agent_identity_certificate():
+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.
+        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.
@@ -240,18 +240,18 @@ def get_and_parse_agent_identity_certificate():
         == "false"
     )
     if is_opted_out:
-        return None
+        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:
@@ -261,9 +261,22 @@ def get_and_parse_agent_identity_certificate():
             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_bytes
 
-    return parse_certificate(cert_bytes)
+
+def get_and_parse_agent_identity_certificate():
+    """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.
+    """
+    cert, _ = get_agent_identity_certificate_and_bytes()
+    return cert
 
 
 def parse_certificate(cert_bytes):

packages/google-auth/google/auth/compute_engine/_metadata.py

@@ -255,6 +255,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.
 
@@ -275,6 +277,8 @@ def get(
         headers (Optional[Mapping[str, str]]): Headers for the request.
         return_none_for_not_found_error (Optional[bool]): If True, returns None
             for 404 error instead of throwing an exception.
+        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.
         timeout (int): How long to wait, in seconds for the metadata server to respond.
 
     Returns:
@@ -319,9 +323,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 "
@@ -491,18 +501,25 @@ def get_service_account_token(request, service_account="default", scopes=None):
             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
+    headers = {metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds()}
 
-    metrics_header = {
-        metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds()
-    }
+    # Default to standard GET. We conditionally upgrade to POST (bound token)
+    # if certificate is found and conditions for bound token are met.
+    method = "GET"
+    body = None
+
+    cert, cert_bytes = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+    if cert and _agent_identity_utils.should_request_bound_token(cert):
+        method = "POST"
+        body = json.dumps({"certificate_chain": cert_bytes.decode("utf-8")}).encode(
+            "utf-8"
+        )
+        headers["Content-Type"] = "application/json"
 
     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"]
     )

packages/google-auth/google/auth/compute_engine/credentials.py

@@ -20,6 +20,7 @@
 """
 
 import datetime
+import json
 import logging
 from typing import Optional, TYPE_CHECKING
 
@@ -526,13 +527,35 @@ def _call_metadata_identity_endpoint(self, request):
             ValueError: If extracting expiry from the obtained ID token fails.
         """
         try:
+            from google.auth import _agent_identity_utils
+
             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()
-            }
+            headers = {metrics.API_CLIENT_HEADER: metrics.token_request_id_token_mds()}
+
+            # Default to standard GET. We conditionally upgrade to POST (bound token)
+            # if certificate is found and conditions for bound token are met.
+            method = "GET"
+            body = None
+
+            (
+                cert,
+                cert_bytes,
+            ) = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+            if cert and _agent_identity_utils.should_request_bound_token(cert):
+                method = "POST"
+                body = json.dumps(
+                    {"certificate_chain": cert_bytes.decode("utf-8")}
+                ).encode("utf-8")
+                headers["Content-Type"] = "application/json"
+
             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)

packages/google-auth/tests/compute_engine/test__metadata.py

@@ -639,8 +639,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 +672,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 +708,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 +743,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",
@@ -758,14 +757,13 @@ def test_get_service_account_token_with_bound_token(
     mock_metrics_header_value,
     mock_get_and_parse,
     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_and_parse.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"})
@@ -774,40 +772,42 @@ def test_get_service_account_token_with_bound_token(
 
     mock_get_and_parse.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
+    mock_get_and_parse.return_value = (None, None)
     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)
 
     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
 ):
     # 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
+    mock_get_and_parse.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 +816,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():