fix: key Eventarc publisher client cache by source principal for impersonated credentials

src/google/adk/integrations/eventarc/_client.py

@@ -76,13 +76,25 @@ def _get_credential_id(credentials: typing.Any) -> str:
   ):
     return "ComputeEngineCredentials"
 
+  module_name = getattr(credentials.__class__, "__module__", "")
+
+  # Impersonated credentials act as the target service account, but which
+  # source principal is authorized to impersonate it is part of the identity.
+  # Keying on the target alone would let a caller that cannot impersonate the
+  # target reuse a client that was authorized for a different source principal.
+  if module_name == "google.auth.impersonated_credentials":
+    target = getattr(credentials, "service_account_email", None)
+    source = getattr(credentials, "_source_credentials", None)
+    if target is not None and source is not None:
+      return f"Impersonated:{target}:{_get_credential_id(source)}"
+    return str(id(credentials))
+
   sa_email = getattr(credentials, "service_account_email", None)
   if sa_email is not None:
-    # This covers both standard ServiceAccountCredentials and ImpersonatedCredentials
+    # This covers standard ServiceAccountCredentials.
     return str(sa_email)
 
   # Handle User Credentials (like local ADC) using refresh token hash
-  module_name = getattr(credentials.__class__, "__module__", "")
   if (
       module_name.startswith("google.oauth2.credentials")
       and getattr(credentials, "refresh_token", None) is not None

tests/unittests/integrations/eventarc/test_client.py

@@ -30,6 +30,38 @@
 
 class TestEventarcClient(unittest.IsolatedAsyncioTestCase):
 
+  def test_impersonated_credentials_are_keyed_by_source_principal(self):
+    def make_sa(email):
+      return google.oauth2.service_account.Credentials(
+          signer=mock.Mock(),
+          service_account_email=email,
+          token_uri="https://oauth2.mtls.googleapis.com/token",
+      )
+
+    def impersonate(source):
+      return google.auth.impersonated_credentials.Credentials(
+          source_credentials=source,
+          target_principal="target@test.com",
+          target_scopes=[],
+      )
+
+    authorized = impersonate(make_sa("authorized@test.com"))
+    same_source = impersonate(make_sa("authorized@test.com"))
+    other_source = impersonate(make_sa("unauthorized@test.com"))
+
+    self.assertEqual(
+        client._get_credential_id(authorized),
+        client._get_credential_id(same_source),
+    )
+    self.assertNotEqual(
+        client._get_credential_id(authorized),
+        client._get_credential_id(other_source),
+    )
+    self.assertNotEqual(
+        client._get_cache_key(authorized, "ua", "project"),
+        client._get_cache_key(other_source, "ua", "project"),
+    )
+
   def test_get_credential_id(self):
     # Service Account
     sa_creds = google.oauth2.service_account.Credentials(
@@ -39,13 +71,16 @@ def test_get_credential_id(self):
     )
     self.assertEqual(client._get_credential_id(sa_creds), "test@test.com")
 
-    # Impersonated (Uses service_account_email under the hood in google-auth)
+    # Impersonated (the identity includes the source principal)
     imp_creds = google.auth.impersonated_credentials.Credentials(
-        source_credentials=mock.Mock(),
+        source_credentials=sa_creds,
         target_principal="imp@test.com",
         target_scopes=[],
     )
-    self.assertEqual(client._get_credential_id(imp_creds), "imp@test.com")
+    self.assertEqual(
+        client._get_credential_id(imp_creds),
+        "Impersonated:imp@test.com:test@test.com",
+    )
 
     # Compute Engine (ADC)
     gce_creds = google.auth.compute_engine.credentials.Credentials()