codekingpro/portable-devtools

1# Copyright 2020 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 Credentials.16 17This module provides credentials that exchange workload identity pool external18credentials for Google access tokens. This facilitates accessing Google Cloud19Platform resources from on-prem and non-Google Cloud platforms (e.g. AWS,20Microsoft Azure, OIDC identity providers), using native credentials retrieved21from the current environment without the need to copy, save and manage22long-lived service account credentials.23 24Specifically, this is intended to use access tokens acquired using the GCP STS25token exchange endpoint following the `OAuth 2.0 Token Exchange`_ spec.26 27.. _OAuth 2.0 Token Exchange: https://tools.ietf.org/html/rfc869328"""29 30import abc31import copy32from dataclasses import dataclass33import datetime34import io35import json36import re37 38from google.auth import _helpers39from google.auth import credentials40from google.auth import exceptions41from google.auth import impersonated_credentials42from google.auth import metrics43from google.oauth2 import sts44from google.oauth2 import utils45 46# External account JSON type identifier.47_EXTERNAL_ACCOUNT_JSON_TYPE = "external_account"48# The token exchange grant_type used for exchanging credentials.49_STS_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"50# The token exchange requested_token_type. This is always an access_token.51_STS_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"52# Cloud resource manager URL used to retrieve project information.53_CLOUD_RESOURCE_MANAGER = "https://cloudresourcemanager.googleapis.com/v1/projects/"54# Default Google sts token url.55_DEFAULT_TOKEN_URL = "https://sts.googleapis.com/v1/token"56 57 58@dataclass59class SupplierContext:60    """A context class that contains information about the requested third party credential that is passed61        to AWS security credential and subject token suppliers.62 63        Attributes:64            subject_token_type (str): The requested subject token type based on the Oauth2.0 token exchange spec.65                Expected values include::66 67                    “urn:ietf:params:oauth:token-type:jwt”68                    “urn:ietf:params:oauth:token-type:id-token”69                    “urn:ietf:params:oauth:token-type:saml2”70                    “urn:ietf:params:aws:token-type:aws4_request”71 72            audience (str): The requested audience for the subject token.73    """74 75    subject_token_type: str76    audience: str77 78 79class Credentials(80    credentials.Scoped,81    credentials.CredentialsWithQuotaProject,82    credentials.CredentialsWithTokenUri,83    metaclass=abc.ABCMeta,84):85    """Base class for all external account credentials.86 87    This is used to instantiate Credentials for exchanging external account88    credentials for Google access token and authorizing requests to Google APIs.89    The base class implements the common logic for exchanging external account90    credentials for Google access tokens.91    """92 93    def __init__(94        self,95        audience,96        subject_token_type,97        token_url,98        credential_source,99        service_account_impersonation_url=None,100        service_account_impersonation_options=None,101        client_id=None,102        client_secret=None,103        token_info_url=None,104        quota_project_id=None,105        scopes=None,106        default_scopes=None,107        workforce_pool_user_project=None,108        universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,109        trust_boundary=None,110    ):111        """Instantiates an external account credentials object.112 113        Args:114            audience (str): The STS audience field.115            subject_token_type (str): The subject token type based on the Oauth2.0 token exchange spec.116                Expected values include::117 118                    “urn:ietf:params:oauth:token-type:jwt”119                    “urn:ietf:params:oauth:token-type:id-token”120                    “urn:ietf:params:oauth:token-type:saml2”121                    “urn:ietf:params:aws:token-type:aws4_request”122 123            token_url (str): The STS endpoint URL.124            credential_source (Mapping): The credential source dictionary.125            service_account_impersonation_url (Optional[str]): The optional service account126                impersonation generateAccessToken URL.127            client_id (Optional[str]): The optional client ID.128            client_secret (Optional[str]): The optional client secret.129            token_info_url (str): The optional STS endpoint URL for token introspection.130            quota_project_id (Optional[str]): The optional quota project ID.131            scopes (Optional[Sequence[str]]): Optional scopes to request during the132                authorization grant.133            default_scopes (Optional[Sequence[str]]): Default scopes passed by a134                Google client library. Use 'scopes' for user-defined scopes.135            workforce_pool_user_project (Optona[str]): The optional workforce pool user136                project number when the credential corresponds to a workforce pool and not137                a workload identity pool. The underlying principal must still have138                serviceusage.services.use IAM permission to use the project for139                billing/quota.140            universe_domain (str): The universe domain. The default universe141                domain is googleapis.com.142            trust_boundary (str): String representation of trust boundary meta.143        Raises:144            google.auth.exceptions.RefreshError: If the generateAccessToken145                endpoint returned an error.146        """147        super(Credentials, self).__init__()148        self._audience = audience149        self._subject_token_type = subject_token_type150        self._token_url = token_url151        self._token_info_url = token_info_url152        self._credential_source = credential_source153        self._service_account_impersonation_url = service_account_impersonation_url154        self._service_account_impersonation_options = (155            service_account_impersonation_options or {}156        )157        self._client_id = client_id158        self._client_secret = client_secret159        self._quota_project_id = quota_project_id160        self._scopes = scopes161        self._default_scopes = default_scopes162        self._workforce_pool_user_project = workforce_pool_user_project163        self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN164        self._trust_boundary = {165            "locations": [],166            "encoded_locations": "0x0",167        }  # expose a placeholder trust boundary value.168 169        if self._client_id:170            self._client_auth = utils.ClientAuthentication(171                utils.ClientAuthType.basic, self._client_id, self._client_secret172            )173        else:174            self._client_auth = None175        self._sts_client = sts.Client(self._token_url, self._client_auth)176 177        self._metrics_options = self._create_default_metrics_options()178 179        self._impersonated_credentials = None180        self._project_id = None181        self._supplier_context = SupplierContext(182            self._subject_token_type, self._audience183        )184 185        if not self.is_workforce_pool and self._workforce_pool_user_project:186            # Workload identity pools do not support workforce pool user projects.187            raise exceptions.InvalidValue(188                "workforce_pool_user_project should not be set for non-workforce pool "189                "credentials"190            )191 192    @property193    def info(self):194        """Generates the dictionary representation of the current credentials.195 196        Returns:197            Mapping: The dictionary representation of the credentials. This is the198                reverse of "from_info" defined on the subclasses of this class. It is199                useful for serializing the current credentials so it can deserialized200                later.201        """202        config_info = self._constructor_args()203        config_info.update(204            type=_EXTERNAL_ACCOUNT_JSON_TYPE,205            service_account_impersonation=config_info.pop(206                "service_account_impersonation_options", None207            ),208        )209        config_info.pop("scopes", None)210        config_info.pop("default_scopes", None)211        return {key: value for key, value in config_info.items() if value is not None}212 213    def _constructor_args(self):214        args = {215            "audience": self._audience,216            "subject_token_type": self._subject_token_type,217            "token_url": self._token_url,218            "token_info_url": self._token_info_url,219            "service_account_impersonation_url": self._service_account_impersonation_url,220            "service_account_impersonation_options": copy.deepcopy(221                self._service_account_impersonation_options222            )223            or None,224            "credential_source": copy.deepcopy(self._credential_source),225            "quota_project_id": self._quota_project_id,226            "client_id": self._client_id,227            "client_secret": self._client_secret,228            "workforce_pool_user_project": self._workforce_pool_user_project,229            "scopes": self._scopes,230            "default_scopes": self._default_scopes,231            "universe_domain": self._universe_domain,232        }233        if not self.is_workforce_pool:234            args.pop("workforce_pool_user_project")235        return args236 237    @property238    def service_account_email(self):239        """Returns the service account email if service account impersonation is used.240 241        Returns:242            Optional[str]: The service account email if impersonation is used. Otherwise243                None is returned.244        """245        if self._service_account_impersonation_url:246            # Parse email from URL. The formal looks as follows:247            # https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/name@project-id.iam.gserviceaccount.com:generateAccessToken248            url = self._service_account_impersonation_url249            start_index = url.rfind("/")250            end_index = url.find(":generateAccessToken")251            if start_index != -1 and end_index != -1 and start_index < end_index:252                start_index = start_index + 1253                return url[start_index:end_index]254        return None255 256    @property257    def is_user(self):258        """Returns whether the credentials represent a user (True) or workload (False).259        Workloads behave similarly to service accounts. Currently workloads will use260        service account impersonation but will eventually not require impersonation.261        As a result, this property is more reliable than the service account email262        property in determining if the credentials represent a user or workload.263 264        Returns:265            bool: True if the credentials represent a user. False if they represent a266                workload.267        """268        # If service account impersonation is used, the credentials will always represent a269        # service account.270        if self._service_account_impersonation_url:271            return False272        return self.is_workforce_pool273 274    @property275    def is_workforce_pool(self):276        """Returns whether the credentials represent a workforce pool (True) or277        workload (False) based on the credentials' audience.278 279        This will also return True for impersonated workforce pool credentials.280 281        Returns:282            bool: True if the credentials represent a workforce pool. False if they283                represent a workload.284        """285        # Workforce pools representing users have the following audience format:286        # //iam.googleapis.com/locations/$location/workforcePools/$poolId/providers/$providerId287        p = re.compile(r"//iam\.googleapis\.com/locations/[^/]+/workforcePools/")288        return p.match(self._audience or "") is not None289 290    @property291    def requires_scopes(self):292        """Checks if the credentials requires scopes.293 294        Returns:295            bool: True if there are no scopes set otherwise False.296        """297        return not self._scopes and not self._default_scopes298 299    @property300    def project_number(self):301        """Optional[str]: The project number corresponding to the workload identity pool."""302 303        # STS audience pattern:304        # //iam.googleapis.com/projects/$PROJECT_NUMBER/locations/...305        components = self._audience.split("/")306        try:307            project_index = components.index("projects")308            if project_index + 1 < len(components):309                return components[project_index + 1] or None310        except ValueError:311            return None312 313    @property314    def token_info_url(self):315        """Optional[str]: The STS token introspection endpoint."""316 317        return self._token_info_url318 319    @_helpers.copy_docstring(credentials.Scoped)320    def with_scopes(self, scopes, default_scopes=None):321        kwargs = self._constructor_args()322        kwargs.update(scopes=scopes, default_scopes=default_scopes)323        scoped = self.__class__(**kwargs)324        scoped._metrics_options = self._metrics_options325        return scoped326 327    @abc.abstractmethod328    def retrieve_subject_token(self, request):329        """Retrieves the subject token using the credential_source object.330 331        Args:332            request (google.auth.transport.Request): A callable used to make333                HTTP requests.334        Returns:335            str: The retrieved subject token.336        """337        # pylint: disable=missing-raises-doc338        # (pylint doesn't recognize that this is abstract)339        raise NotImplementedError("retrieve_subject_token must be implemented")340 341    def get_project_id(self, request):342        """Retrieves the project ID corresponding to the workload identity or workforce pool.343        For workforce pool credentials, it returns the project ID corresponding to344        the workforce_pool_user_project.345 346        When not determinable, None is returned.347 348        This is introduced to support the current pattern of using the Auth library:349 350            credentials, project_id = google.auth.default()351 352        The resource may not have permission (resourcemanager.projects.get) to353        call this API or the required scopes may not be selected:354        https://cloud.google.com/resource-manager/reference/rest/v1/projects/get#authorization-scopes355 356        Args:357            request (google.auth.transport.Request): A callable used to make358                HTTP requests.359        Returns:360            Optional[str]: The project ID corresponding to the workload identity pool361                or workforce pool if determinable.362        """363        if self._project_id:364            # If already retrieved, return the cached project ID value.365            return self._project_id366        scopes = self._scopes if self._scopes is not None else self._default_scopes367        # Scopes are required in order to retrieve a valid access token.368        project_number = self.project_number or self._workforce_pool_user_project369        if project_number and scopes:370            headers = {}371            url = _CLOUD_RESOURCE_MANAGER + project_number372            self.before_request(request, "GET", url, headers)373            response = request(url=url, method="GET", headers=headers)374 375            response_body = (376                response.data.decode("utf-8")377                if hasattr(response.data, "decode")378                else response.data379            )380            response_data = json.loads(response_body)381 382            if response.status == 200:383                # Cache result as this field is immutable.384                self._project_id = response_data.get("projectId")385                return self._project_id386 387        return None388 389    @_helpers.copy_docstring(credentials.Credentials)390    def refresh(self, request):391        scopes = self._scopes if self._scopes is not None else self._default_scopes392 393        if self._should_initialize_impersonated_credentials():394            self._impersonated_credentials = self._initialize_impersonated_credentials()395 396        if self._impersonated_credentials:397            self._impersonated_credentials.refresh(request)398            self.token = self._impersonated_credentials.token399            self.expiry = self._impersonated_credentials.expiry400        else:401            now = _helpers.utcnow()402            additional_options = None403            # Do not pass workforce_pool_user_project when client authentication404            # is used. The client ID is sufficient for determining the user project.405            if self._workforce_pool_user_project and not self._client_id:406                additional_options = {"userProject": self._workforce_pool_user_project}407            additional_headers = {408                metrics.API_CLIENT_HEADER: metrics.byoid_metrics_header(409                    self._metrics_options410                )411            }412            response_data = self._sts_client.exchange_token(413                request=request,414                grant_type=_STS_GRANT_TYPE,415                subject_token=self.retrieve_subject_token(request),416                subject_token_type=self._subject_token_type,417                audience=self._audience,418                scopes=scopes,419                requested_token_type=_STS_REQUESTED_TOKEN_TYPE,420                additional_options=additional_options,421                additional_headers=additional_headers,422            )423            self.token = response_data.get("access_token")424            expires_in = response_data.get("expires_in")425            # Some services do not respect the OAUTH2.0 RFC and send expires_in as a426            # JSON String.427            if isinstance(expires_in, str):428                expires_in = int(expires_in)429 430            lifetime = datetime.timedelta(seconds=expires_in)431 432            self.expiry = now + lifetime433 434    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)435    def with_quota_project(self, quota_project_id):436        # Return copy of instance with the provided quota project ID.437        kwargs = self._constructor_args()438        kwargs.update(quota_project_id=quota_project_id)439        new_cred = self.__class__(**kwargs)440        new_cred._metrics_options = self._metrics_options441        return new_cred442 443    @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)444    def with_token_uri(self, token_uri):445        kwargs = self._constructor_args()446        kwargs.update(token_url=token_uri)447        new_cred = self.__class__(**kwargs)448        new_cred._metrics_options = self._metrics_options449        return new_cred450 451    @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)452    def with_universe_domain(self, universe_domain):453        kwargs = self._constructor_args()454        kwargs.update(universe_domain=universe_domain)455        new_cred = self.__class__(**kwargs)456        new_cred._metrics_options = self._metrics_options457        return new_cred458 459    def _should_initialize_impersonated_credentials(self):460        return (461            self._service_account_impersonation_url is not None462            and self._impersonated_credentials is None463        )464 465    def _initialize_impersonated_credentials(self):466        """Generates an impersonated credentials.467 468        For more details, see `projects.serviceAccounts.generateAccessToken`_.469 470        .. _projects.serviceAccounts.generateAccessToken: https://cloud.google.com/iam/docs/reference/credentials/rest/v1/projects.serviceAccounts/generateAccessToken471 472        Returns:473            impersonated_credentials.Credential: The impersonated credentials474                object.475 476        Raises:477            google.auth.exceptions.RefreshError: If the generateAccessToken478                endpoint returned an error.479        """480        # Return copy of instance with no service account impersonation.481        kwargs = self._constructor_args()482        kwargs.update(483            service_account_impersonation_url=None,484            service_account_impersonation_options={},485        )486        source_credentials = self.__class__(**kwargs)487        source_credentials._metrics_options = self._metrics_options488 489        # Determine target_principal.490        target_principal = self.service_account_email491        if not target_principal:492            raise exceptions.RefreshError(493                "Unable to determine target principal from service account impersonation URL."494            )495 496        scopes = self._scopes if self._scopes is not None else self._default_scopes497        # Initialize and return impersonated credentials.498        return impersonated_credentials.Credentials(499            source_credentials=source_credentials,500            target_principal=target_principal,501            target_scopes=scopes,502            quota_project_id=self._quota_project_id,503            iam_endpoint_override=self._service_account_impersonation_url,504            lifetime=self._service_account_impersonation_options.get(505                "token_lifetime_seconds"506            ),507        )508 509    def _create_default_metrics_options(self):510        metrics_options = {}511        if self._service_account_impersonation_url:512            metrics_options["sa-impersonation"] = "true"513        else:514            metrics_options["sa-impersonation"] = "false"515        if self._service_account_impersonation_options.get("token_lifetime_seconds"):516            metrics_options["config-lifetime"] = "true"517        else:518            metrics_options["config-lifetime"] = "false"519 520        return metrics_options521 522    @classmethod523    def from_info(cls, info, **kwargs):524        """Creates a Credentials instance from parsed external account info.525 526        Args:527            info (Mapping[str, str]): The external account info in Google528                format.529            kwargs: Additional arguments to pass to the constructor.530 531        Returns:532            google.auth.identity_pool.Credentials: The constructed533                credentials.534 535        Raises:536            InvalidValue: For invalid parameters.537        """538        return cls(539            audience=info.get("audience"),540            subject_token_type=info.get("subject_token_type"),541            token_url=info.get("token_url"),542            token_info_url=info.get("token_info_url"),543            service_account_impersonation_url=info.get(544                "service_account_impersonation_url"545            ),546            service_account_impersonation_options=info.get(547                "service_account_impersonation"548            )549            or {},550            client_id=info.get("client_id"),551            client_secret=info.get("client_secret"),552            credential_source=info.get("credential_source"),553            quota_project_id=info.get("quota_project_id"),554            workforce_pool_user_project=info.get("workforce_pool_user_project"),555            universe_domain=info.get(556                "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN557            ),558            **kwargs559        )560 561    @classmethod562    def from_file(cls, filename, **kwargs):563        """Creates a Credentials instance from an external account json file.564 565        Args:566            filename (str): The path to the external account json file.567            kwargs: Additional arguments to pass to the constructor.568 569        Returns:570            google.auth.identity_pool.Credentials: The constructed571                credentials.572        """573        with io.open(filename, "r", encoding="utf-8") as json_file:574            data = json.load(json_file)575            return cls.from_info(data, **kwargs)576