Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 2 commits into from Sep 28, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 3 additions & 1 deletion google/oauth2/service_account.py
Expand Up @@ -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
Expand Down
29 changes: 29 additions & 0 deletions tests/oauth2/test_service_account.py
Expand Up @@ -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"
Expand Down