fix: disable self signed jwt for domain wide delegation (#873)

2 files changed, 32 insertions, 1 deletions

diff --git a/google/oauth2/service_account.py b/google/oauth2/service_account.py
index 8f18f26..ecaac03 100644
--- a/google/oauth2/service_account.py
+++ b/google/oauth2/service_account.py
@@ -399,7 +399,9 @@ class Credentials(
 
     @_helpers.copy_docstring(credentials.Credentials)
     def refresh(self, request):
-        if self._jwt_credentials is not None:
+        # Since domain wide delegation doesn't work with self signed JWT. If
+        # subject exists, then we should not use self signed JWT.
+        if self._subject is None and self._jwt_credentials is not None:
             self._jwt_credentials.refresh(request)
             self.token = self._jwt_credentials.token
             self.expiry = self._jwt_credentials.expiry
diff --git a/tests/oauth2/test_service_account.py b/tests/oauth2/test_service_account.py
index 370438f..531fc4c 100644
--- a/tests/oauth2/test_service_account.py
+++ b/tests/oauth2/test_service_account.py
@@ -371,6 +371,35 @@ class TestCredentials(object):
         assert credentials.token == token
         assert credentials.expiry == expiry
 
+    @mock.patch("google.oauth2._client.jwt_grant", autospec=True)
+    @mock.patch("google.auth.jwt.Credentials.refresh", autospec=True)
+    def test_refresh_jwt_not_used_for_domain_wide_delegation(
+        self, self_signed_jwt_refresh, jwt_grant
+    ):
+        # Create a domain wide delegation credentials by setting the subject.
+        credentials = service_account.Credentials(
+            SIGNER,
+            self.SERVICE_ACCOUNT_EMAIL,
+            self.TOKEN_URI,
+            always_use_jwt_access=True,
+            subject="subject",
+        )
+        credentials._create_self_signed_jwt("https://pubsub.googleapis.com")
+        jwt_grant.return_value = (
+            "token",
+            _helpers.utcnow() + datetime.timedelta(seconds=500),
+            {},
+        )
+        request = mock.create_autospec(transport.Request, instance=True)
+
+        # Refresh credentials
+        credentials.refresh(request)
+
+        # Make sure we are using jwt_grant and not self signed JWT refresh
+        # method to obtain the token.
+        assert jwt_grant.called
+        assert not self_signed_jwt_refresh.called
+
 
 class TestIDTokenCredentials(object):
     SERVICE_ACCOUNT_EMAIL = "service-account@example.com"