codekingpro/portable-devtools

1# Copyright 2022 Google LLC2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15"""External Account Authorized User Credentials.16This module provides credentials based on OAuth 2.0 access and refresh tokens.17These credentials usually access resources on behalf of a user (resource18owner).19 20Specifically, these are sourced using external identities via Workforce Identity Federation.21 22Obtaining the initial access and refresh token can be done through the Google Cloud CLI.23 24Example credential:25{26  "type": "external_account_authorized_user",27  "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",28  "refresh_token": "refreshToken",29  "token_url": "https://sts.googleapis.com/v1/oauth/token",30  "token_info_url": "https://sts.googleapis.com/v1/instrospect",31  "client_id": "clientId",32  "client_secret": "clientSecret"33}34"""35 36import datetime37import io38import json39 40from google.auth import _helpers41from google.auth import credentials42from google.auth import exceptions43from google.oauth2 import sts44from google.oauth2 import utils45 46_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user"47 48 49class Credentials(50    credentials.CredentialsWithQuotaProject,51    credentials.ReadOnlyScoped,52    credentials.CredentialsWithTokenUri,53):54    """Credentials for External Account Authorized Users.55 56    This is used to instantiate Credentials for exchanging refresh tokens from57    authorized users for Google access token and authorizing requests to Google58    APIs.59 60    The credentials are considered immutable. If you want to modify the61    quota project, use `with_quota_project` and if you want to modify the token62    uri, use `with_token_uri`.63    """64 65    def __init__(66        self,67        token=None,68        expiry=None,69        refresh_token=None,70        audience=None,71        client_id=None,72        client_secret=None,73        token_url=None,74        token_info_url=None,75        revoke_url=None,76        scopes=None,77        quota_project_id=None,78        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,79    ):80        """Instantiates a external account authorized user credentials object.81 82        Args:83        token (str): The OAuth 2.0 access token. Can be None if refresh information84            is provided.85        expiry (datetime.datetime): The optional expiration datetime of the OAuth 2.0 access86            token.87        refresh_token (str): The optional OAuth 2.0 refresh token. If specified,88            credentials can be refreshed.89        audience (str): The optional STS audience which contains the resource name for the workforce90            pool and the provider identifier in that pool.91        client_id (str): The OAuth 2.0 client ID. Must be specified for refresh, can be left as92            None if the token can not be refreshed.93        client_secret (str): The OAuth 2.0 client secret. Must be specified for refresh, can be94            left as None if the token can not be refreshed.95        token_url (str): The optional STS token exchange endpoint for refresh. Must be specified for96            refresh, can be left as None if the token can not be refreshed.97        token_info_url (str): The optional STS endpoint URL for token introspection.98        revoke_url (str): The optional STS endpoint URL for revoking tokens.99        quota_project_id (str): The optional project ID used for quota and billing.100            This project may be different from the project used to101            create the credentials.102        universe_domain (Optional[str]): The universe domain. The default value103            is googleapis.com.104 105        Returns:106            google.auth.external_account_authorized_user.Credentials: The107                constructed credentials.108        """109        super(Credentials, self).__init__()110 111        self.token = token112        self.expiry = expiry113        self._audience = audience114        self._refresh_token = refresh_token115        self._token_url = token_url116        self._token_info_url = token_info_url117        self._client_id = client_id118        self._client_secret = client_secret119        self._revoke_url = revoke_url120        self._quota_project_id = quota_project_id121        self._scopes = scopes122        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN123 124        if not self.valid and not self.can_refresh:125            raise exceptions.InvalidOperation(126                "Token should be created with fields to make it valid (`token` and "127                "`expiry`), or fields to allow it to refresh (`refresh_token`, "128                "`token_url`, `client_id`, `client_secret`)."129            )130 131        self._client_auth = None132        if self._client_id:133            self._client_auth = utils.ClientAuthentication(134                utils.ClientAuthType.basic, self._client_id, self._client_secret135            )136        self._sts_client = sts.Client(self._token_url, self._client_auth)137 138    @property139    def info(self):140        """Generates the serializable dictionary representation of the current141        credentials.142 143        Returns:144            Mapping: The dictionary representation of the credentials. This is the145                reverse of the "from_info" method defined in this class. It is146                useful for serializing the current credentials so it can deserialized147                later.148        """149        config_info = self.constructor_args()150        config_info.update(type=_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE)151        if config_info["expiry"]:152            config_info["expiry"] = config_info["expiry"].isoformat() + "Z"153 154        return {key: value for key, value in config_info.items() if value is not None}155 156    def constructor_args(self):157        return {158            "audience": self._audience,159            "refresh_token": self._refresh_token,160            "token_url": self._token_url,161            "token_info_url": self._token_info_url,162            "client_id": self._client_id,163            "client_secret": self._client_secret,164            "token": self.token,165            "expiry": self.expiry,166            "revoke_url": self._revoke_url,167            "scopes": self._scopes,168            "quota_project_id": self._quota_project_id,169            "universe_domain": self._universe_domain,170        }171 172    @property173    def scopes(self):174        """Optional[str]: The OAuth 2.0 permission scopes."""175        return self._scopes176 177    @property178    def requires_scopes(self):179        """ False: OAuth 2.0 credentials have their scopes set when180        the initial token is requested and can not be changed."""181        return False182 183    @property184    def client_id(self):185        """Optional[str]: The OAuth 2.0 client ID."""186        return self._client_id187 188    @property189    def client_secret(self):190        """Optional[str]: The OAuth 2.0 client secret."""191        return self._client_secret192 193    @property194    def audience(self):195        """Optional[str]: The STS audience which contains the resource name for the196            workforce pool and the provider identifier in that pool."""197        return self._audience198 199    @property200    def refresh_token(self):201        """Optional[str]: The OAuth 2.0 refresh token."""202        return self._refresh_token203 204    @property205    def token_url(self):206        """Optional[str]: The STS token exchange endpoint for refresh."""207        return self._token_url208 209    @property210    def token_info_url(self):211        """Optional[str]: The STS endpoint for token info."""212        return self._token_info_url213 214    @property215    def revoke_url(self):216        """Optional[str]: The STS endpoint for token revocation."""217        return self._revoke_url218 219    @property220    def is_user(self):221        """ True: This credential always represents a user."""222        return True223 224    @property225    def can_refresh(self):226        return all(227            (self._refresh_token, self._token_url, self._client_id, self._client_secret)228        )229 230    def get_project_id(self, request=None):231        """Retrieves the project ID corresponding to the workload identity or workforce pool.232        For workforce pool credentials, it returns the project ID corresponding to233        the workforce_pool_user_project.234 235        When not determinable, None is returned.236 237        Args:238            request (google.auth.transport.requests.Request): Request object.239                Unused here, but passed from _default.default().240 241        Return:242          str: project ID is not determinable for this credential type so it returns None243        """244 245        return None246 247    def to_json(self, strip=None):248        """Utility function that creates a JSON representation of this249        credential.250        Args:251            strip (Sequence[str]): Optional list of members to exclude from the252                                   generated JSON.253        Returns:254            str: A JSON representation of this instance. When converted into255            a dictionary, it can be passed to from_info()256            to create a new instance.257        """258        strip = strip if strip else []259        return json.dumps({k: v for (k, v) in self.info.items() if k not in strip})260 261    def refresh(self, request):262        """Refreshes the access token.263 264        Args:265            request (google.auth.transport.Request): The object used to make266                HTTP requests.267 268        Raises:269            google.auth.exceptions.RefreshError: If the credentials could270                not be refreshed.271        """272        if not self.can_refresh:273            raise exceptions.RefreshError(274                "The credentials do not contain the necessary fields need to "275                "refresh the access token. You must specify refresh_token, "276                "token_url, client_id, and client_secret."277            )278 279        now = _helpers.utcnow()280        response_data = self._make_sts_request(request)281 282        self.token = response_data.get("access_token")283 284        lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))285        self.expiry = now + lifetime286 287        if "refresh_token" in response_data:288            self._refresh_token = response_data["refresh_token"]289 290    def _make_sts_request(self, request):291        return self._sts_client.refresh_token(request, self._refresh_token)292 293    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)294    def with_quota_project(self, quota_project_id):295        kwargs = self.constructor_args()296        kwargs.update(quota_project_id=quota_project_id)297        return self.__class__(**kwargs)298 299    @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)300    def with_token_uri(self, token_uri):301        kwargs = self.constructor_args()302        kwargs.update(token_url=token_uri)303        return self.__class__(**kwargs)304 305    @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)306    def with_universe_domain(self, universe_domain):307        kwargs = self.constructor_args()308        kwargs.update(universe_domain=universe_domain)309        return self.__class__(**kwargs)310 311    @classmethod312    def from_info(cls, info, **kwargs):313        """Creates a Credentials instance from parsed external account info.314 315        Args:316            info (Mapping[str, str]): The external account info in Google317                format.318            kwargs: Additional arguments to pass to the constructor.319 320        Returns:321            google.auth.external_account_authorized_user.Credentials: The322                constructed credentials.323 324        Raises:325            ValueError: For invalid parameters.326        """327        expiry = info.get("expiry")328        if expiry:329            expiry = datetime.datetime.strptime(330                expiry.rstrip("Z").split(".")[0], "%Y-%m-%dT%H:%M:%S"331            )332        return cls(333            audience=info.get("audience"),334            refresh_token=info.get("refresh_token"),335            token_url=info.get("token_url"),336            token_info_url=info.get("token_info_url"),337            client_id=info.get("client_id"),338            client_secret=info.get("client_secret"),339            token=info.get("token"),340            expiry=expiry,341            revoke_url=info.get("revoke_url"),342            quota_project_id=info.get("quota_project_id"),343            scopes=info.get("scopes"),344            universe_domain=info.get(345                "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN346            ),347            **kwargs348        )349 350    @classmethod351    def from_file(cls, filename, **kwargs):352        """Creates a Credentials instance from an external account json file.353 354        Args:355            filename (str): The path to the external account json file.356            kwargs: Additional arguments to pass to the constructor.357 358        Returns:359            google.auth.external_account_authorized_user.Credentials: The360                constructed credentials.361        """362        with io.open(filename, "r", encoding="utf-8") as json_file:363            data = json.load(json_file)364            return cls.from_info(data, **kwargs)365