fix: disable self signed jwt for domain wide delegation (#873)
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 @@
@_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 @@
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"