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

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

@@ -193,14 +193,14 @@ def get_agent_identity_certificate_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.
@@ -212,18 +212,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:
@@ -233,9 +233,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

@@ -253,6 +253,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.
 
@@ -317,9 +319,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 "
@@ -489,18 +497,27 @@ 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()
+    method = "GET"
+    body = None
+
+    cert, cert_bytes = _agent_identity_utils.get_agent_identity_certificate_and_bytes()
     if cert:
         if _agent_identity_utils.should_request_bound_token(cert):
-            fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(cert)
-            params["bindCertificateFingerprint"] = fingerprint
+            method = "POST"
+            body = json.dumps({"certificate_chain": cert_bytes.decode("utf-8")}).encode(
+                "utf-8"
+            )
 
     metrics_header = {
         metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds()
     }
+    if method == "POST":
+        metrics_header["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=metrics_header, 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,36 @@ 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()
             }
+
+            method = "GET"
+            body = None
+
+            cert_and_bytes = (
+                _agent_identity_utils.get_agent_identity_certificate_and_bytes()
+            )
+            if cert_and_bytes:
+                cert, cert_bytes = cert_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")
+                    metrics_header["Content-Type"] = "application/json"
+
             id_token = _metadata.get(
-                request, path, params=params, headers=metrics_header
+                request,
+                path,
+                params=params,
+                headers=metrics_header,
+                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

@@ -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,20 +772,21 @@ 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")
 
 
 @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"})
 

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

@@ -467,28 +467,24 @@ 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)
+        import json
+
+        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 = [
             {"email": "service-account@example.com", "scopes": ["one", "two"]},
@@ -498,13 +494,16 @@ 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")
 
     @mock.patch("google.auth.compute_engine._metadata.get")
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
@@ -826,6 +825,113 @@ 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,
+    ):
+        import json
+        import base64
+        from google.auth import metrics
+
+        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"][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,
+    ):
+        import base64
+        from google.auth import metrics
+
+        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._helpers.utcnow",
         return_value=_helpers.utcfromtimestamp(0),