feat: adds new external account authorized user credentials (#1160)

google/auth/_default.py

@@ -35,12 +35,14 @@
 _AUTHORIZED_USER_TYPE = "authorized_user"
 _SERVICE_ACCOUNT_TYPE = "service_account"
 _EXTERNAL_ACCOUNT_TYPE = "external_account"
+_EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE = "external_account_authorized_user"
 _IMPERSONATED_SERVICE_ACCOUNT_TYPE = "impersonated_service_account"
 _GDCH_SERVICE_ACCOUNT_TYPE = "gdch_service_account"
 _VALID_TYPES = (
     _AUTHORIZED_USER_TYPE,
     _SERVICE_ACCOUNT_TYPE,
     _EXTERNAL_ACCOUNT_TYPE,
+    _EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE,
     _IMPERSONATED_SERVICE_ACCOUNT_TYPE,
     _GDCH_SERVICE_ACCOUNT_TYPE,
 )
@@ -158,6 +160,12 @@ def _load_credentials_from_info(
             default_scopes=default_scopes,
             request=request,
         )
+
+    elif credential_type == _EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE:
+        credentials, project_id = _get_external_account_authorized_user_credentials(
+            filename, info, request
+        )
+
     elif credential_type == _IMPERSONATED_SERVICE_ACCOUNT_TYPE:
         credentials, project_id = _get_impersonated_service_account_credentials(
             filename, info, scopes
@@ -363,6 +371,23 @@ def _get_external_account_credentials(
     return credentials, credentials.get_project_id(request=request)
 
 
+def _get_external_account_authorized_user_credentials(
+    filename, info, scopes=None, default_scopes=None, request=None
+):
+    try:
+        from google.auth import external_account_authorized_user
+
+        credentials = external_account_authorized_user.Credentials.from_info(info)
+    except ValueError:
+        raise exceptions.DefaultCredentialsError(
+            "Failed to load external account authorized user credentials from {}".format(
+                filename
+            )
+        )
+
+    return credentials, None
+
+
 def _get_authorized_user_credentials(filename, info, scopes=None):
     from google.oauth2 import credentials
 

google/auth/external_account_authorized_user.py

@@ -0,0 +1,290 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""External Account Authorized User Credentials.
+This module provides credentials based on OAuth 2.0 access and refresh tokens.
+These credentials usually access resources on behalf of a user (resource
+owner).
+
+Specifically, these are sourced using external identities via Workforce Identity Federation.
+
+Obtaining the initial access and refresh token can be done through the Google Cloud CLI.
+
+Example credential:
+{
+  "type": "external_account_authorized_user",
+  "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
+  "refresh_token": "refreshToken",
+  "token_url": "https://sts.googleapis.com/v1/oauth/token",
+  "token_info_url": "https://sts.googleapis.com/v1/instrospect",
+  "client_id": "clientId",
+  "client_secret": "clientSecret"
+}
+"""
+
+import datetime
+import io
+import json
+
+from google.auth import _helpers
+from google.auth import credentials
+from google.auth import exceptions
+from google.oauth2 import sts
+from google.oauth2 import utils
+
+_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user"
+
+
+class Credentials(
+    credentials.CredentialsWithQuotaProject,
+    credentials.ReadOnlyScoped,
+    credentials.CredentialsWithTokenUri,
+):
+    """Credentials for External Account Authorized Users.
+
+    This is used to instantiate Credentials for exchanging refresh tokens from
+    authorized users for Google access token and authorizing requests to Google
+    APIs.
+
+    The credentials are considered immutable. If you want to modify the
+    quota project, use `with_quota_project` and if you want to modify the token
+    uri, use `with_token_uri`.
+    """
+
+    def __init__(
+        self,
+        token=None,
+        expiry=None,
+        refresh_token=None,
+        audience=None,
+        client_id=None,
+        client_secret=None,
+        token_url=None,
+        token_info_url=None,
+        revoke_url=None,
+        quota_project_id=None,
+    ):
+        """Instantiates a external account authorized user credentials object.
+
+        Args:
+        token (str): The OAuth 2.0 access token. Can be None if refresh information
+            is provided.
+        expiry (datetime.datetime): The optional expiration datetime of the OAuth 2.0 access
+            token.
+        refresh_token (str): The optional OAuth 2.0 refresh token. If specified,
+            credentials can be refreshed.
+        audience (str): The optional STS audience which contains the resource name for the workforce
+            pool and the provider identifier in that pool.
+        client_id (str): The OAuth 2.0 client ID. Must be specified for refresh, can be left as
+            None if the token can not be refreshed.
+        client_secret (str): The OAuth 2.0 client secret. Must be specified for refresh, can be
+            left as None if the token can not be refreshed.
+        token_url (str): The optional STS token exchange endpoint. Must be specified fro refresh,
+            can be leftas None if the token can not be refreshed.
+        token_info_url (str): The optional STS endpoint URL for token introspection.
+        revoke_url (str): The optional STS endpoint URL for revoking tokens.
+        quota_project_id (str): The optional project ID used for quota and billing.
+            This project may be different from the project used to
+            create the credentials.
+
+        Returns:
+            google.auth.external_account_authorized_user.Credentials: The
+                constructed credentials.
+        """
+        if not any((refresh_token, token)):
+            raise ValueError("Either `refresh_token` or `token` should be set.")
+
+        super(Credentials, self).__init__()
+
+        self.token = token
+        self.expiry = expiry
+        self._audience = audience
+        self._refresh_token = refresh_token
+        self._token_url = token_url
+        self._token_info_url = token_info_url
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._revoke_url = revoke_url
+        self._quota_project_id = quota_project_id
+
+        self._client_auth = None
+        if self._client_id:
+            self._client_auth = utils.ClientAuthentication(
+                utils.ClientAuthType.basic, self._client_id, self._client_secret
+            )
+        self._sts_client = sts.Client(self._token_url, self._client_auth)
+
+    @property
+    def info(self):
+        """Generates the serializable dictionary representation of the current
+        credentials.
+
+        Returns:
+            Mapping: The dictionary representation of the credentials. This is the
+                reverse of the "from_info" method defined in this class. It is
+                useful for serializing the current credentials so it can deserialized
+                later.
+        """
+        config_info = self.constructor_args()
+        config_info.update(type=_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE)
+        if config_info["expiry"]:
+            config_info["expiry"] = config_info["expiry"].isoformat() + "Z"
+
+        return {key: value for key, value in config_info.items() if value is not None}
+
+    def constructor_args(self):
+        return {
+            "audience": self._audience,
+            "refresh_token": self._refresh_token,
+            "token_url": self._token_url,
+            "token_info_url": self._token_info_url,
+            "client_id": self._client_id,
+            "client_secret": self._client_secret,
+            "token": self.token,
+            "expiry": self.expiry,
+            "revoke_url": self._revoke_url,
+            "quota_project_id": self._quota_project_id,
+        }
+
+    @property
+    def requires_scopes(self):
+        """ False: OAuth 2.0 credentials have their scopes set when
+        the initial token is requested and can not be changed."""
+        return False
+
+    @property
+    def is_user(self):
+        """ True: This credential always represents a user."""
+        return True
+
+    def get_project_id(self):
+        """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.
+        """
+
+        return None
+
+    def to_json(self, strip=None):
+        """Utility function that creates a JSON representation of this
+        credential.
+        Args:
+            strip (Sequence[str]): Optional list of members to exclude from the
+                                   generated JSON.
+        Returns:
+            str: A JSON representation of this instance. When converted into
+            a dictionary, it can be passed to from_info()
+            to create a new instance.
+        """
+        strip = strip if strip else []
+        return json.dumps({k: v for (k, v) in self.info.items() if k not in strip})
+
+    def refresh(self, request):
+        """Refreshes the access token.
+
+        Args:
+            request (google.auth.transport.Request): The object used to make
+                HTTP requests.
+
+        Raises:
+            google.auth.exceptions.RefreshError: If the credentials could
+                not be refreshed.
+        """
+        if not all(
+            (self._refresh_token, self._token_url, self._client_id, self._client_secret)
+        ):
+            raise exceptions.RefreshError(
+                "The credentials do not contain the necessary fields need to "
+                "refresh the access token. You must specify refresh_token, "
+                "token_url, client_id, and client_secret."
+            )
+
+        now = _helpers.utcnow()
+        response_data = self._make_sts_request(request)
+
+        self.token = response_data.get("access_token")
+
+        lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
+        self.expiry = now + lifetime
+
+        if "refresh_token" in response_data:
+            self._refresh_token = response_data["refresh_token"]
+
+    def _make_sts_request(self, request):
+        return self._sts_client.refresh_token(request, self._refresh_token)
+
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        kwargs = self.constructor_args()
+        kwargs.update(quota_project_id=quota_project_id)
+        return self.__class__(**kwargs)
+
+    @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
+    def with_token_uri(self, token_uri):
+        kwargs = self.constructor_args()
+        kwargs.update(token_url=token_uri)
+        return self.__class__(**kwargs)
+
+    @classmethod
+    def from_info(cls, info, **kwargs):
+        """Creates a Credentials instance from parsed external account info.
+
+        Args:
+            info (Mapping[str, str]): The external account info in Google
+                format.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Returns:
+            google.auth.external_account_authorized_user.Credentials: The
+                constructed credentials.
+
+        Raises:
+            ValueError: For invalid parameters.
+        """
+        expiry = info.get("expiry")
+        if expiry:
+            expiry = datetime.datetime.strptime(
+                expiry.rstrip("Z").split(".")[0], "%Y-%m-%dT%H:%M:%S"
+            )
+        return cls(
+            audience=info.get("audience"),
+            refresh_token=info.get("refresh_token"),
+            token_url=info.get("token_url"),
+            token_info_url=info.get("token_info_url"),
+            client_id=info.get("client_id"),
+            client_secret=info.get("client_secret"),
+            token=info.get("token"),
+            expiry=expiry,
+            revoke_url=info.get("revoke_url"),
+            quota_project_id=info.get("quota_project_id"),
+            **kwargs
+        )
+
+    @classmethod
+    def from_file(cls, filename, **kwargs):
+        """Creates a Credentials instance from an external account json file.
+
+        Args:
+            filename (str): The path to the external account json file.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Returns:
+            google.auth.external_account_authorized_user.Credentials: The
+                constructed credentials.
+        """
+        with io.open(filename, "r", encoding="utf-8") as json_file:
+            data = json.load(json_file)
+            return cls.from_info(data, **kwargs)