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

google/oauth2/service_account.py

@@ -399,7 +399,9 @@ def _make_authorization_grant_assertion(self):
 
     @_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

tests/oauth2/test_service_account.py

@@ -371,6 +371,35 @@ def test_refresh_with_jwt_credentials(self, make_jwt):
         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"