From 7c27190302dfddd7da6ba23f706ddd9281e01478 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Sun, 12 Jul 2026 18:39:25 -0700 Subject: [PATCH 1/3] feat(auth): add bound token support for access and JWT id tokens Switch the MDS token acquisition from a GET to a POST request when the agentic cert is detected. * Add `get_agent_identity_certificate_and_bytes()` utility to read the raw certificate bytes alongside the parsed cert. * Update `_metadata.get_service_account_token()` (for access tokens) and `IDTokenCredentials.refresh()` (for ID tokens) to send a POST request with the `certificate_chain` payload instead of a GET request when bound tokens are supported. * Update `_metadata.get()` helper to support `method` and `body` params. * Add and update unit tests to verify the new POST request flows. --- .../google/auth/_agent_identity_utils.py | 27 +++- .../google/auth/compute_engine/_metadata.py | 31 +++- .../google/auth/compute_engine/credentials.py | 26 +++- .../tests/compute_engine/test__metadata.py | 19 ++- .../tests/compute_engine/test_credentials.py | 138 ++++++++++++++++-- 5 files changed, 200 insertions(+), 41 deletions(-) diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py index 4de5d709b4b5..ad71b6004633 100644 --- a/packages/google-auth/google/auth/_agent_identity_utils.py +++ b/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): diff --git a/packages/google-auth/google/auth/compute_engine/_metadata.py b/packages/google-auth/google/auth/compute_engine/_metadata.py index 1ea7792c2cdd..bf4463efd5ee 100644 --- a/packages/google-auth/google/auth/compute_engine/_metadata.py +++ b/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. @@ -319,9 +321,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 +499,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"] ) diff --git a/packages/google-auth/google/auth/compute_engine/credentials.py b/packages/google-auth/google/auth/compute_engine/credentials.py index 3701751bda2b..8bc50195ab09 100644 --- a/packages/google-auth/google/auth/compute_engine/credentials.py +++ b/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) diff --git a/packages/google-auth/tests/compute_engine/test__metadata.py b/packages/google-auth/tests/compute_engine/test__metadata.py index 0fae4bd6ef16..23b3e914b24f 100644 --- a/packages/google-auth/tests/compute_engine/test__metadata.py +++ b/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"}) diff --git a/packages/google-auth/tests/compute_engine/test_credentials.py b/packages/google-auth/tests/compute_engine/test_credentials.py index ab171c5a6041..4d8ce597aa0b 100644 --- a/packages/google-auth/tests/compute_engine/test_credentials.py +++ b/packages/google-auth/tests/compute_engine/test_credentials.py @@ -475,28 +475,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 = [ { @@ -509,13 +505,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") @@ -858,6 +857,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), From 3ab20bc3eea43b949a27018c1488a53e7be3fc0d Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Wed, 29 Jul 2026 10:39:14 -0700 Subject: [PATCH 2/3] Refactor bound token request logic in compute_engine for consistency and update docstrings and comments --- .../google/auth/compute_engine/_metadata.py | 26 ++++++++--------- .../google/auth/compute_engine/credentials.py | 29 +++++++++---------- 2 files changed, 27 insertions(+), 28 deletions(-) diff --git a/packages/google-auth/google/auth/compute_engine/_metadata.py b/packages/google-auth/google/auth/compute_engine/_metadata.py index bf4463efd5ee..b85fc23cc26f 100644 --- a/packages/google-auth/google/auth/compute_engine/_metadata.py +++ b/packages/google-auth/google/auth/compute_engine/_metadata.py @@ -277,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: @@ -499,26 +501,24 @@ def get_service_account_token(request, service_account="default", scopes=None): scopes = ",".join(scopes) params["scopes"] = scopes + headers = {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: - if _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 = { - metrics.API_CLIENT_HEADER: metrics.token_request_access_token_mds() - } - if method == "POST": - metrics_header["Content-Type"] = "application/json" + 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, method=method, body=body + 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 8bc50195ab09..4aa86aa7ea8b 100644 --- a/packages/google-auth/google/auth/compute_engine/credentials.py +++ b/packages/google-auth/google/auth/compute_engine/credentials.py @@ -531,30 +531,29 @@ def _call_metadata_identity_endpoint(self, request): 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_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" + ( + 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, + headers=headers, method=method, body=body, ) From ccc2a5d7f83ef23bb29e246e42218515bc6da501 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Wed, 29 Jul 2026 10:40:58 -0700 Subject: [PATCH 3/3] test(auth): update compute_engine tests for bound token headers and replace outdated mocks --- .../tests/compute_engine/test__metadata.py | 25 ++--- .../tests/compute_engine/test_credentials.py | 98 ++++++++++++++++--- 2 files changed, 97 insertions(+), 26 deletions(-) diff --git a/packages/google-auth/tests/compute_engine/test__metadata.py b/packages/google-auth/tests/compute_engine/test__metadata.py index 23b3e914b24f..1ada83909b65 100644 --- a/packages/google-auth/tests/compute_engine/test__metadata.py +++ b/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", @@ -779,6 +779,7 @@ def test_get_service_account_token_with_bound_token( 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( @@ -794,19 +795,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 ): # 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"}) @@ -815,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(): diff --git a/packages/google-auth/tests/compute_engine/test_credentials.py b/packages/google-auth/tests/compute_engine/test_credentials.py index 4d8ce597aa0b..a1b8a479f615 100644 --- a/packages/google-auth/tests/compute_engine/test_credentials.py +++ b/packages/google-auth/tests/compute_engine/test_credentials.py @@ -515,26 +515,22 @@ def test_refresh_with_agent_identity( 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", @@ -543,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_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 "bindCertificateFingerprint" not in kwargs.get("params", {}) + 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_get_cert_and_bytes.assert_called_once() + kwargs = mock_metadata_get.call_args[1] + assert kwargs.get("method", "GET") == "GET" + assert kwargs.get("body") is None def test_set_blocking_regional_access_boundary_lookup(self): creds = self.credentials @@ -907,6 +932,7 @@ def test_refresh_with_agent_identity( 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() @@ -964,6 +990,50 @@ def test_refresh_with_agent_identity_opt_out_or_not_agent( == 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, + ): + 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_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),