feat(keycardai-oauth): impersonate() takes a per-call issuer for multi-zone clients (ECO-107) (#337)

Both impersonate() methods accept a keyword-only issuer and pass it to build_http_context the way exchange_token and client_credentials_grant do, so a multi-zone client can pick the zone for an impersonation.

Co-authored-by: devin-ai-keycard <devin-ai@keycard.ai>
Co-authored-by: Larry Osakwe <larry@keycard.ai>

Author: 3 people

packages/oauth/src/keycardai/oauth/client.py

@@ -1019,6 +1019,7 @@ async def impersonate(
         resource: str,
         scope: str | None = None,
         timeout: float | None = None,
+        issuer: str | None = None,
     ) -> TokenResponse:
         """Impersonate a user to obtain a resource token.
 
@@ -1034,6 +1035,10 @@ async def impersonate(
             resource: Target resource URI (e.g. "https://graph.microsoft.com").
             scope: Optional scope string for the requested token.
             timeout: Optional request timeout override.
+            issuer: Issuer URL selecting which credentials a zone-aware auth
+                strategy (e.g. MultiZoneBasicAuth) applies for this call.
+                Defaults to the client's issuer. Single-credential strategies
+                ignore it.
 
         Returns:
             TokenResponse with the resource access token.
@@ -1066,7 +1071,7 @@ async def impersonate(
             endpoint=endpoints.token,
             transport=self.transport,
             auth=self.auth_strategy,
-            issuer=self.issuer,
+            issuer=issuer if issuer is not None else self.issuer,
             user_agent=self.config.user_agent,
             custom_headers=self.config.custom_headers,
             timeout=timeout or self.config.timeout,
@@ -1804,6 +1809,7 @@ def impersonate(
         resource: str,
         scope: str | None = None,
         timeout: float | None = None,
+        issuer: str | None = None,
     ) -> TokenResponse:
         """Impersonate a user to obtain a resource token.
 
@@ -1819,6 +1825,10 @@ def impersonate(
             resource: Target resource URI (e.g. "https://graph.microsoft.com").
             scope: Optional scope string for the requested token.
             timeout: Optional request timeout override.
+            issuer: Issuer URL selecting which credentials a zone-aware auth
+                strategy (e.g. MultiZoneBasicAuth) applies for this call.
+                Defaults to the client's issuer. Single-credential strategies
+                ignore it.
 
         Returns:
             TokenResponse with the resource access token.
@@ -1851,7 +1861,7 @@ def impersonate(
             endpoint=endpoints.token,
             transport=self.transport,
             auth=self.auth_strategy,
-            issuer=self.issuer,
+            issuer=issuer if issuer is not None else self.issuer,
             user_agent=self.config.user_agent,
             custom_headers=self.config.custom_headers,
             timeout=timeout or self.config.timeout,

packages/oauth/tests/keycardai/oauth/test_multi_zone_client.py

@@ -178,6 +178,51 @@ def test_client_credentials_grant_unknown_issuer_fails_closed(self):
             )
         assert transport.requests == []
 
+    def test_impersonate_sends_basic_header_for_selected_issuer(self):
+        transport = FakeTransport()
+        client = make_client(transport)
+
+        response = client.impersonate(
+            user_identifier="user@example.com",
+            resource="https://api.example.com",
+            issuer=ZONE2,
+        )
+
+        assert response.access_token == "issued_token"
+        assert len(transport.requests) == 1
+        sent = transport.requests[0]
+        assert sent.headers["Authorization"] == _basic_header(
+            "client_id_2", "client_secret_2"
+        )
+        # The issuer selector is not a wire field.
+        assert b"issuer" not in sent.body
+
+    def test_impersonate_defaults_to_client_issuer(self):
+        transport = FakeTransport()
+        client = make_client(transport)
+
+        client.impersonate(
+            user_identifier="user@example.com",
+            resource="https://api.example.com",
+        )
+
+        sent = transport.requests[0]
+        assert sent.headers["Authorization"] == _basic_header(
+            "client_id_1", "client_secret_1"
+        )
+
+    def test_impersonate_unknown_issuer_fails_closed(self):
+        transport = FakeTransport()
+        client = make_client(transport)
+
+        with pytest.raises(KeyError, match="not configured"):
+            client.impersonate(
+                user_identifier="user@example.com",
+                resource="https://api.example.com",
+                issuer="https://unknown.keycard.cloud",
+            )
+        assert transport.requests == []
+
 
 class TestAsyncClientMultiZone:
     @pytest.mark.asyncio
@@ -222,3 +267,49 @@ async def test_exchange_token_unknown_issuer_fails_closed(self):
                 issuer="https://unknown.keycard.cloud",
             )
         assert transport.requests == []
+
+    @pytest.mark.asyncio
+    async def test_impersonate_sends_basic_header_for_selected_issuer(self):
+        transport = FakeAsyncTransport()
+        client = make_async_client(transport)
+
+        response = await client.impersonate(
+            user_identifier="user@example.com",
+            resource="https://api.example.com",
+            issuer=ZONE2,
+        )
+
+        assert response.access_token == "issued_token"
+        sent = transport.requests[0]
+        assert sent.headers["Authorization"] == _basic_header(
+            "client_id_2", "client_secret_2"
+        )
+        assert b"issuer" not in sent.body
+
+    @pytest.mark.asyncio
+    async def test_impersonate_defaults_to_client_issuer(self):
+        transport = FakeAsyncTransport()
+        client = make_async_client(transport)
+
+        await client.impersonate(
+            user_identifier="user@example.com",
+            resource="https://api.example.com",
+        )
+
+        sent = transport.requests[0]
+        assert sent.headers["Authorization"] == _basic_header(
+            "client_id_1", "client_secret_1"
+        )
+
+    @pytest.mark.asyncio
+    async def test_impersonate_unknown_issuer_fails_closed(self):
+        transport = FakeAsyncTransport()
+        client = make_async_client(transport)
+
+        with pytest.raises(KeyError, match="not configured"):
+            await client.impersonate(
+                user_identifier="user@example.com",
+                resource="https://api.example.com",
+                issuer="https://unknown.keycard.cloud",
+            )
+        assert transport.requests == []