From fec59912f5b33b92c76fcaeae9050731d904b20c Mon Sep 17 00:00:00 2001 From: devin-ai-keycard Date: Tue, 20 Oct 2026 17:58:08 +0000 Subject: [PATCH] fix(keycardai-oauth): impersonate() accepts a client assertion (ECO-455) Both impersonate() methods take keyword-only client_assertion and client_assertion_type and pass them into the TokenExchangeRequest the way exchange_token() does, so an application whose credential is a workload identity can authenticate the substitute-user exchange in the request body. Omitting them leaves the wire request unchanged. Co-Authored-By: Larry Osakwe --- packages/oauth/src/keycardai/oauth/client.py | 40 ++++++ .../test_impersonate_client_assertion.py | 131 ++++++++++++++++++ 2 files changed, 171 insertions(+) create mode 100644 packages/oauth/tests/keycardai/oauth/test_impersonate_client_assertion.py diff --git a/packages/oauth/src/keycardai/oauth/client.py b/packages/oauth/src/keycardai/oauth/client.py index 7c43144a..08e7603a 100644 --- a/packages/oauth/src/keycardai/oauth/client.py +++ b/packages/oauth/src/keycardai/oauth/client.py @@ -1077,6 +1077,8 @@ async def impersonate( scope: str | None = None, timeout: float | None = None, issuer: str | None = None, + client_assertion: str | None = None, + client_assertion_type: str | None = None, ) -> TokenResponse: """Impersonate a user to obtain a resource token. @@ -1096,6 +1098,13 @@ async def impersonate( strategy (e.g. MultiZoneBasicAuth) applies for this call. Defaults to the client's issuer. Single-credential strategies ignore it. + client_assertion: JWT client assertion authenticating the calling + application in the request body, for credentials that carry no + client secret (for example a workload identity token). Sent as + the ``client_assertion`` form parameter, exactly as + ``exchange_token()`` sends it. + client_assertion_type: The assertion type URN, normally + ``urn:ietf:params:oauth:client-assertion-type:jwt-bearer``. Returns: TokenResponse with the resource access token. @@ -1114,12 +1123,23 @@ async def impersonate( resource="https://graph.microsoft.com", ) print(response.access_token) + + With a workload identity instead of a client secret, authenticate + in the request body: + response = await client.impersonate( + user_identifier="user@example.com", + resource="https://graph.microsoft.com", + client_assertion=workload_identity_token, + client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer", + ) """ request = TokenExchangeRequest( subject_token=build_substitute_user_token(user_identifier), subject_token_type=TokenType.SUBSTITUTE_USER, resource=resource, scope=scope, + client_assertion=client_assertion, + client_assertion_type=client_assertion_type, ) endpoints = await self._get_current_endpoints() @@ -1922,6 +1942,8 @@ def impersonate( scope: str | None = None, timeout: float | None = None, issuer: str | None = None, + client_assertion: str | None = None, + client_assertion_type: str | None = None, ) -> TokenResponse: """Impersonate a user to obtain a resource token. @@ -1941,6 +1963,13 @@ def impersonate( strategy (e.g. MultiZoneBasicAuth) applies for this call. Defaults to the client's issuer. Single-credential strategies ignore it. + client_assertion: JWT client assertion authenticating the calling + application in the request body, for credentials that carry no + client secret (for example a workload identity token). Sent as + the ``client_assertion`` form parameter, exactly as + ``exchange_token()`` sends it. + client_assertion_type: The assertion type URN, normally + ``urn:ietf:params:oauth:client-assertion-type:jwt-bearer``. Returns: TokenResponse with the resource access token. @@ -1959,12 +1988,23 @@ def impersonate( resource="https://graph.microsoft.com", ) print(response.access_token) + + With a workload identity instead of a client secret, authenticate + in the request body: + response = client.impersonate( + user_identifier="user@example.com", + resource="https://graph.microsoft.com", + client_assertion=workload_identity_token, + client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer", + ) """ request = TokenExchangeRequest( subject_token=build_substitute_user_token(user_identifier), subject_token_type=TokenType.SUBSTITUTE_USER, resource=resource, scope=scope, + client_assertion=client_assertion, + client_assertion_type=client_assertion_type, ) endpoints = self._get_current_endpoints() diff --git a/packages/oauth/tests/keycardai/oauth/test_impersonate_client_assertion.py b/packages/oauth/tests/keycardai/oauth/test_impersonate_client_assertion.py new file mode 100644 index 00000000..666253b4 --- /dev/null +++ b/packages/oauth/tests/keycardai/oauth/test_impersonate_client_assertion.py @@ -0,0 +1,131 @@ +"""Wire-level tests for client assertions on Client/AsyncClient.impersonate(). + +An application whose credential is a workload identity has no client secret +and authenticates the substitute-user exchange in the request body, the same +way exchange_token() and client_credentials_grant() accept it. +""" + +import json +from urllib.parse import parse_qs + +import pytest + +from keycardai.oauth import AsyncClient, Client, ClientConfig +from keycardai.oauth.http._wire import HttpRequest, HttpResponse +from keycardai.oauth.types.models import TokenExchangeRequest, TokenType + +ZONE = "https://zone.keycard.cloud" +RESOURCE = "https://api.example.com" +ASSERTION = "header.payload.signature" +ASSERTION_TYPE = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + +TOKEN_RESPONSE = json.dumps( + {"access_token": "issued_token", "token_type": "Bearer", "expires_in": 3600} +).encode() + + +class FakeTransport: + def __init__(self): + self.requests: list[HttpRequest] = [] + + def request_raw(self, request: HttpRequest, timeout=None) -> HttpResponse: + self.requests.append(request) + return HttpResponse( + status=200, headers={"Content-Type": "application/json"}, body=TOKEN_RESPONSE + ) + + +class FakeAsyncTransport(FakeTransport): + async def request_raw(self, request: HttpRequest, timeout=None) -> HttpResponse: + return super().request_raw(request, timeout) + + +def _form(request: HttpRequest) -> dict[str, list[str]]: + return parse_qs(request.body.decode(), keep_blank_values=True) + + +def _make_client(transport) -> Client: + return Client(ZONE, transport=transport, config=ClientConfig(enable_metadata_discovery=False)) + + +def _make_async_client(transport) -> AsyncClient: + return AsyncClient( + ZONE, transport=transport, config=ClientConfig(enable_metadata_discovery=False) + ) + + +class TestSyncImpersonateClientAssertion: + def test_assertion_fields_reach_the_form_body(self): + transport = FakeTransport() + client = _make_client(transport) + + response = client.impersonate( + user_identifier="user@example.com", + resource=RESOURCE, + client_assertion=ASSERTION, + client_assertion_type=ASSERTION_TYPE, + ) + + assert response.access_token == "issued_token" + form = _form(transport.requests[0]) + assert form["client_assertion"] == [ASSERTION] + assert form["client_assertion_type"] == [ASSERTION_TYPE] + assert form["subject_token_type"] == [TokenType.SUBSTITUTE_USER.value] + assert form["resource"] == [RESOURCE] + assert "Authorization" not in transport.requests[0].headers + + def test_omitting_the_assertion_is_unchanged(self): + transport = FakeTransport() + client = _make_client(transport) + + client.impersonate(user_identifier="user@example.com", resource=RESOURCE) + client.exchange_token( + TokenExchangeRequest( + subject_token=transport.requests[0].body.decode().split("subject_token=")[1].split("&")[0], + subject_token_type=TokenType.SUBSTITUTE_USER, + resource=RESOURCE, + ) + ) + + impersonated, baseline = transport.requests + assert impersonated.body == baseline.body + assert b"client_assertion" not in impersonated.body + + +class TestAsyncImpersonateClientAssertion: + @pytest.mark.asyncio + async def test_assertion_fields_reach_the_form_body(self): + transport = FakeAsyncTransport() + client = _make_async_client(transport) + + response = await client.impersonate( + user_identifier="user@example.com", + resource=RESOURCE, + client_assertion=ASSERTION, + client_assertion_type=ASSERTION_TYPE, + ) + + assert response.access_token == "issued_token" + form = _form(transport.requests[0]) + assert form["client_assertion"] == [ASSERTION] + assert form["client_assertion_type"] == [ASSERTION_TYPE] + assert form["subject_token_type"] == [TokenType.SUBSTITUTE_USER.value] + assert "Authorization" not in transport.requests[0].headers + + @pytest.mark.asyncio + async def test_omitting_the_assertion_is_unchanged(self): + transport = FakeAsyncTransport() + client = _make_async_client(transport) + + await client.impersonate(user_identifier="user@example.com", resource=RESOURCE) + await client.exchange_token( + TokenExchangeRequest( + subject_token=transport.requests[0].body.decode().split("subject_token=")[1].split("&")[0], + subject_token_type=TokenType.SUBSTITUTE_USER, + resource=RESOURCE, + ) + ) + + impersonated, baseline = transport.requests + assert impersonated.body == baseline.body + assert b"client_assertion" not in impersonated.body