feat: workload identity federation support (#698)
Using workload identity federation, applications can access Google Cloud resources from Amazon Web Services (AWS), Microsoft Azure or any identity provider that supports OpenID Connect (OIDC). Workload identity federation is recommended for non-Google Cloud environments as it avoids the need to download, manage and store service account private keys locally.
This includes a rollforward of the [previous reverted PR](https://github.com/googleapis/google-auth-library-python/pull/686) and the [fix](https://github.com/googleapis/google-auth-library-python/pull/686) to not pass scopes to user credentials from `google.auth.default()`.
diff --git a/tests/test_impersonated_credentials.py b/tests/test_impersonated_credentials.py
index 305f939..430c770 100644
--- a/tests/test_impersonated_credentials.py
+++ b/tests/test_impersonated_credentials.py
@@ -104,12 +104,17 @@
SIGNER, SERVICE_ACCOUNT_EMAIL, TOKEN_URI
)
USER_SOURCE_CREDENTIALS = credentials.Credentials(token="ABCDE")
+ IAM_ENDPOINT_OVERRIDE = (
+ "https://us-east1-iamcredentials.googleapis.com/v1/projects/-"
+ + "/serviceAccounts/{}:generateAccessToken".format(SERVICE_ACCOUNT_EMAIL)
+ )
def make_credentials(
self,
source_credentials=SOURCE_CREDENTIALS,
lifetime=LIFETIME,
target_principal=TARGET_PRINCIPAL,
+ iam_endpoint_override=None,
):
return Credentials(
@@ -118,6 +123,7 @@
target_scopes=self.TARGET_SCOPES,
delegates=self.DELEGATES,
lifetime=lifetime,
+ iam_endpoint_override=iam_endpoint_override,
)
def test_make_from_user_credentials(self):
@@ -172,6 +178,34 @@
assert credentials.valid
assert not credentials.expired
+ @pytest.mark.parametrize("use_data_bytes", [True, False])
+ def test_refresh_success_iam_endpoint_override(
+ self, use_data_bytes, mock_donor_credentials
+ ):
+ credentials = self.make_credentials(
+ lifetime=None, iam_endpoint_override=self.IAM_ENDPOINT_OVERRIDE
+ )
+ token = "token"
+
+ expire_time = (
+ _helpers.utcnow().replace(microsecond=0) + datetime.timedelta(seconds=500)
+ ).isoformat("T") + "Z"
+ response_body = {"accessToken": token, "expireTime": expire_time}
+
+ request = self.make_request(
+ data=json.dumps(response_body),
+ status=http_client.OK,
+ use_data_bytes=use_data_bytes,
+ )
+
+ credentials.refresh(request)
+
+ assert credentials.valid
+ assert not credentials.expired
+ # Confirm override endpoint used.
+ request_kwargs = request.call_args.kwargs
+ assert request_kwargs["url"] == self.IAM_ENDPOINT_OVERRIDE
+
@pytest.mark.parametrize("time_skew", [100, -100])
def test_refresh_source_credentials(self, time_skew):
credentials = self.make_credentials(lifetime=None)
@@ -317,6 +351,36 @@
quota_project_creds = credentials.with_quota_project("project-foo")
assert quota_project_creds._quota_project_id == "project-foo"
+ @pytest.mark.parametrize("use_data_bytes", [True, False])
+ def test_with_quota_project_iam_endpoint_override(
+ self, use_data_bytes, mock_donor_credentials
+ ):
+ credentials = self.make_credentials(
+ lifetime=None, iam_endpoint_override=self.IAM_ENDPOINT_OVERRIDE
+ )
+ token = "token"
+ # iam_endpoint_override should be copied to created credentials.
+ quota_project_creds = credentials.with_quota_project("project-foo")
+
+ expire_time = (
+ _helpers.utcnow().replace(microsecond=0) + datetime.timedelta(seconds=500)
+ ).isoformat("T") + "Z"
+ response_body = {"accessToken": token, "expireTime": expire_time}
+
+ request = self.make_request(
+ data=json.dumps(response_body),
+ status=http_client.OK,
+ use_data_bytes=use_data_bytes,
+ )
+
+ quota_project_creds.refresh(request)
+
+ assert quota_project_creds.valid
+ assert not quota_project_creds.expired
+ # Confirm override endpoint used.
+ request_kwargs = request.call_args.kwargs
+ assert request_kwargs["url"] == self.IAM_ENDPOINT_OVERRIDE
+
def test_id_token_success(
self, mock_donor_credentials, mock_authorizedsession_idtoken
):