feat: add support for workforce pool credentials (#868)
Workforce pools (external account credentials for non-Google users) are
organization-level resources which means that issued workforce pool tokens
will not have any client project ID on token exchange as currently designed.
"To use a Google API, the client must identify the application to the server.
If the API requires authentication, the client must also identify the principal
running the application."
The application here is the client project. The token will identify the user
principal but not the application. This will result in APIs rejecting requests
authenticated with these tokens.
Note that passing a `x-goog-user-project` override header on API request is
still not sufficient. The token is still expected to have a client project.
As a result, we have extended the spec to support an additional
`workforce_pool_user_project` for these credentials (workforce pools) which will
be passed when exchanging an external token for a Google Access token. After the
exchange, the issued access token will use the supplied project as the client
project. The underlying principal must still have `serviceusage.services.use`
IAM permission to use the project for billing/quota.
This field is not needed for flows with basic client authentication (e.g. client
ID is supplied). The client ID is sufficient to determine the client project and
any additionally supplied `workforce_pool_user_project` value will be ignored.
Note that this feature is not usable yet publicly.
The additional field has been added to the abstract external account credentials
`google.auth.external_account.Credentials` and the subclass
`google.auth.identity_pool.Credentials`.
diff --git a/google/auth/external_account.py b/google/auth/external_account.py
index 24b93b4..f588981 100644
--- a/google/auth/external_account.py
+++ b/google/auth/external_account.py
@@ -73,6 +73,7 @@
quota_project_id=None,
scopes=None,
default_scopes=None,
+ workforce_pool_user_project=None,
):
"""Instantiates an external account credentials object.
@@ -90,6 +91,11 @@
authorization grant.
default_scopes (Optional[Sequence[str]]): Default scopes passed by a
Google client library. Use 'scopes' for user-defined scopes.
+ workforce_pool_user_project (Optona[str]): The optional workforce pool user
+ project number when the credential corresponds to a workforce pool and not
+ a workload identity pool. The underlying principal must still have
+ serviceusage.services.use IAM permission to use the project for
+ billing/quota.
Raises:
google.auth.exceptions.RefreshError: If the generateAccessToken
endpoint returned an error.
@@ -105,6 +111,7 @@
self._quota_project_id = quota_project_id
self._scopes = scopes
self._default_scopes = default_scopes
+ self._workforce_pool_user_project = workforce_pool_user_project
if self._client_id:
self._client_auth = utils.ClientAuthentication(
@@ -120,6 +127,13 @@
self._impersonated_credentials = None
self._project_id = None
+ if not self.is_workforce_pool and self._workforce_pool_user_project:
+ # Workload identity pools do not support workforce pool user projects.
+ raise ValueError(
+ "workforce_pool_user_project should not be set for non-workforce pool "
+ "credentials"
+ )
+
@property
def info(self):
"""Generates the dictionary representation of the current credentials.
@@ -140,6 +154,7 @@
"quota_project_id": self._quota_project_id,
"client_id": self._client_id,
"client_secret": self._client_secret,
+ "workforce_pool_user_project": self._workforce_pool_user_project,
}
return {key: value for key, value in config_info.items() if value is not None}
@@ -178,12 +193,23 @@
# service account.
if self._service_account_impersonation_url:
return False
+ return self.is_workforce_pool
+
+ @property
+ def is_workforce_pool(self):
+ """Returns whether the credentials represent a workforce pool (True) or
+ workload (False) based on the credentials' audience.
+
+ This will also return True for impersonated workforce pool credentials.
+
+ Returns:
+ bool: True if the credentials represent a workforce pool. False if they
+ represent a workload.
+ """
# Workforce pools representing users have the following audience format:
# //iam.googleapis.com/locations/$location/workforcePools/$poolId/providers/$providerId
p = re.compile(r"//iam\.googleapis\.com/locations/[^/]+/workforcePools/")
- if p.match(self._audience):
- return True
- return False
+ return p.match(self._audience or "") is not None
@property
def requires_scopes(self):
@@ -210,7 +236,7 @@
@_helpers.copy_docstring(credentials.Scoped)
def with_scopes(self, scopes, default_scopes=None):
- return self.__class__(
+ d = dict(
audience=self._audience,
subject_token_type=self._subject_token_type,
token_url=self._token_url,
@@ -221,7 +247,11 @@
quota_project_id=self._quota_project_id,
scopes=scopes,
default_scopes=default_scopes,
+ workforce_pool_user_project=self._workforce_pool_user_project,
)
+ if not self.is_workforce_pool:
+ d.pop("workforce_pool_user_project")
+ return self.__class__(**d)
@abc.abstractmethod
def retrieve_subject_token(self, request):
@@ -238,7 +268,9 @@
raise NotImplementedError("retrieve_subject_token must be implemented")
def get_project_id(self, request):
- """Retrieves the project ID corresponding to the workload identity pool.
+ """Retrieves the project ID corresponding to the workload identity or workforce pool.
+ For workforce pool credentials, it returns the project ID corresponding to
+ the workforce_pool_user_project.
When not determinable, None is returned.
@@ -255,16 +287,17 @@
HTTP requests.
Returns:
Optional[str]: The project ID corresponding to the workload identity pool
- if determinable.
+ or workforce pool if determinable.
"""
if self._project_id:
# If already retrieved, return the cached project ID value.
return self._project_id
scopes = self._scopes if self._scopes is not None else self._default_scopes
# Scopes are required in order to retrieve a valid access token.
- if self.project_number and scopes:
+ project_number = self.project_number or self._workforce_pool_user_project
+ if project_number and scopes:
headers = {}
- url = _CLOUD_RESOURCE_MANAGER + self.project_number
+ url = _CLOUD_RESOURCE_MANAGER + project_number
self.before_request(request, "GET", url, headers)
response = request(url=url, method="GET", headers=headers)
@@ -291,6 +324,11 @@
self.expiry = self._impersonated_credentials.expiry
else:
now = _helpers.utcnow()
+ additional_options = None
+ # Do not pass workforce_pool_user_project when client authentication
+ # is used. The client ID is sufficient for determining the user project.
+ if self._workforce_pool_user_project and not self._client_id:
+ additional_options = {"userProject": self._workforce_pool_user_project}
response_data = self._sts_client.exchange_token(
request=request,
grant_type=_STS_GRANT_TYPE,
@@ -299,6 +337,7 @@
audience=self._audience,
scopes=scopes,
requested_token_type=_STS_REQUESTED_TOKEN_TYPE,
+ additional_options=additional_options,
)
self.token = response_data.get("access_token")
lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
@@ -307,7 +346,7 @@
@_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
def with_quota_project(self, quota_project_id):
# Return copy of instance with the provided quota project ID.
- return self.__class__(
+ d = dict(
audience=self._audience,
subject_token_type=self._subject_token_type,
token_url=self._token_url,
@@ -318,7 +357,11 @@
quota_project_id=quota_project_id,
scopes=self._scopes,
default_scopes=self._default_scopes,
+ workforce_pool_user_project=self._workforce_pool_user_project,
)
+ if not self.is_workforce_pool:
+ d.pop("workforce_pool_user_project")
+ return self.__class__(**d)
def _initialize_impersonated_credentials(self):
"""Generates an impersonated credentials.
@@ -336,7 +379,7 @@
endpoint returned an error.
"""
# Return copy of instance with no service account impersonation.
- source_credentials = self.__class__(
+ d = dict(
audience=self._audience,
subject_token_type=self._subject_token_type,
token_url=self._token_url,
@@ -347,7 +390,11 @@
quota_project_id=self._quota_project_id,
scopes=self._scopes,
default_scopes=self._default_scopes,
+ workforce_pool_user_project=self._workforce_pool_user_project,
)
+ if not self.is_workforce_pool:
+ d.pop("workforce_pool_user_project")
+ source_credentials = self.__class__(**d)
# Determine target_principal.
target_principal = self.service_account_email