codekingpro/portable-devtools

1# Copyright 2016 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"""Google Compute Engine credentials.16 17This module provides authentication for an application running on Google18Compute Engine using the Compute Engine metadata server.19 20"""21 22import datetime23 24from google.auth import _helpers25from google.auth import credentials26from google.auth import exceptions27from google.auth import iam28from google.auth import jwt29from google.auth import metrics30from google.auth.compute_engine import _metadata31from google.oauth2 import _client32 33 34class Credentials(35    credentials.Scoped,36    credentials.CredentialsWithQuotaProject,37    credentials.CredentialsWithUniverseDomain,38):39    """Compute Engine Credentials.40 41    These credentials use the Google Compute Engine metadata server to obtain42    OAuth 2.0 access tokens associated with the instance's service account,43    and are also used for Cloud Run, Flex and App Engine (except for the Python44    2.7 runtime, which is supported only on older versions of this library).45 46    For more information about Compute Engine authentication, including how47    to configure scopes, see the `Compute Engine authentication48    documentation`_.49 50    .. note:: On Compute Engine the metadata server ignores requested scopes.51        On Cloud Run, Flex and App Engine the server honours requested scopes.52 53    .. _Compute Engine authentication documentation:54        https://cloud.google.com/compute/docs/authentication#using55    """56 57    def __init__(58        self,59        service_account_email="default",60        quota_project_id=None,61        scopes=None,62        default_scopes=None,63        universe_domain=None,64    ):65        """66        Args:67            service_account_email (str): The service account email to use, or68                'default'. A Compute Engine instance may have multiple service69                accounts.70            quota_project_id (Optional[str]): The project ID used for quota and71                billing.72            scopes (Optional[Sequence[str]]): The list of scopes for the credentials.73            default_scopes (Optional[Sequence[str]]): Default scopes passed by a74                Google client library. Use 'scopes' for user-defined scopes.75            universe_domain (Optional[str]): The universe domain. If not76                provided or None, credential will attempt to fetch the value77                from metadata server. If metadata server doesn't have universe78                domain endpoint, then the default googleapis.com will be used.79        """80        super(Credentials, self).__init__()81        self._service_account_email = service_account_email82        self._quota_project_id = quota_project_id83        self._scopes = scopes84        self._default_scopes = default_scopes85        self._universe_domain_cached = False86        if universe_domain:87            self._universe_domain = universe_domain88            self._universe_domain_cached = True89 90    def _retrieve_info(self, request):91        """Retrieve information about the service account.92 93        Updates the scopes and retrieves the full service account email.94 95        Args:96            request (google.auth.transport.Request): The object used to make97                HTTP requests.98        """99        info = _metadata.get_service_account_info(100            request, service_account=self._service_account_email101        )102 103        self._service_account_email = info["email"]104 105        # Don't override scopes requested by the user.106        if self._scopes is None:107            self._scopes = info["scopes"]108 109    def _metric_header_for_usage(self):110        return metrics.CRED_TYPE_SA_MDS111 112    def refresh(self, request):113        """Refresh the access token and scopes.114 115        Args:116            request (google.auth.transport.Request): The object used to make117                HTTP requests.118 119        Raises:120            google.auth.exceptions.RefreshError: If the Compute Engine metadata121                service can't be reached if if the instance has not122                credentials.123        """124        scopes = self._scopes if self._scopes is not None else self._default_scopes125        try:126            self._retrieve_info(request)127            self.token, self.expiry = _metadata.get_service_account_token(128                request, service_account=self._service_account_email, scopes=scopes129            )130        except exceptions.TransportError as caught_exc:131            new_exc = exceptions.RefreshError(caught_exc)132            raise new_exc from caught_exc133 134    @property135    def service_account_email(self):136        """The service account email.137 138        .. note:: This is not guaranteed to be set until :meth:`refresh` has been139            called.140        """141        return self._service_account_email142 143    @property144    def requires_scopes(self):145        return not self._scopes146 147    @property148    def universe_domain(self):149        if self._universe_domain_cached:150            return self._universe_domain151 152        from google.auth.transport import requests as google_auth_requests153 154        self._universe_domain = _metadata.get_universe_domain(155            google_auth_requests.Request()156        )157        self._universe_domain_cached = True158        return self._universe_domain159 160    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)161    def with_quota_project(self, quota_project_id):162        creds = self.__class__(163            service_account_email=self._service_account_email,164            quota_project_id=quota_project_id,165            scopes=self._scopes,166            default_scopes=self._default_scopes,167        )168        creds._universe_domain = self._universe_domain169        creds._universe_domain_cached = self._universe_domain_cached170        return creds171 172    @_helpers.copy_docstring(credentials.Scoped)173    def with_scopes(self, scopes, default_scopes=None):174        # Compute Engine credentials can not be scoped (the metadata service175        # ignores the scopes parameter). App Engine, Cloud Run and Flex support176        # requesting scopes.177        creds = self.__class__(178            scopes=scopes,179            default_scopes=default_scopes,180            service_account_email=self._service_account_email,181            quota_project_id=self._quota_project_id,182        )183        creds._universe_domain = self._universe_domain184        creds._universe_domain_cached = self._universe_domain_cached185        return creds186 187    @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)188    def with_universe_domain(self, universe_domain):189        return self.__class__(190            scopes=self._scopes,191            default_scopes=self._default_scopes,192            service_account_email=self._service_account_email,193            quota_project_id=self._quota_project_id,194            universe_domain=universe_domain,195        )196 197 198_DEFAULT_TOKEN_LIFETIME_SECS = 3600  # 1 hour in seconds199_DEFAULT_TOKEN_URI = "https://www.googleapis.com/oauth2/v4/token"200 201 202class IDTokenCredentials(203    credentials.CredentialsWithQuotaProject,204    credentials.Signing,205    credentials.CredentialsWithTokenUri,206):207    """Open ID Connect ID Token-based service account credentials.208 209    These credentials relies on the default service account of a GCE instance.210 211    ID token can be requested from `GCE metadata server identity endpoint`_, IAM212    token endpoint or other token endpoints you specify. If metadata server213    identity endpoint is not used, the GCE instance must have been started with214    a service account that has access to the IAM Cloud API.215 216    .. _GCE metadata server identity endpoint:217        https://cloud.google.com/compute/docs/instances/verifying-instance-identity218    """219 220    def __init__(221        self,222        request,223        target_audience,224        token_uri=None,225        additional_claims=None,226        service_account_email=None,227        signer=None,228        use_metadata_identity_endpoint=False,229        quota_project_id=None,230    ):231        """232        Args:233            request (google.auth.transport.Request): The object used to make234                HTTP requests.235            target_audience (str): The intended audience for these credentials,236                used when requesting the ID Token. The ID Token's ``aud`` claim237                will be set to this string.238            token_uri (str): The OAuth 2.0 Token URI.239            additional_claims (Mapping[str, str]): Any additional claims for240                the JWT assertion used in the authorization grant.241            service_account_email (str): Optional explicit service account to242                use to sign JWT tokens.243                By default, this is the default GCE service account.244            signer (google.auth.crypt.Signer): The signer used to sign JWTs.245                In case the signer is specified, the request argument will be246                ignored.247            use_metadata_identity_endpoint (bool): Whether to use GCE metadata248                identity endpoint. For backward compatibility the default value249                is False. If set to True, ``token_uri``, ``additional_claims``,250                ``service_account_email``, ``signer`` argument should not be set;251                otherwise ValueError will be raised.252            quota_project_id (Optional[str]): The project ID used for quota and253                billing.254 255        Raises:256            ValueError:257                If ``use_metadata_identity_endpoint`` is set to True, and one of258                ``token_uri``, ``additional_claims``, ``service_account_email``,259                 ``signer`` arguments is set.260        """261        super(IDTokenCredentials, self).__init__()262 263        self._quota_project_id = quota_project_id264        self._use_metadata_identity_endpoint = use_metadata_identity_endpoint265        self._target_audience = target_audience266 267        if use_metadata_identity_endpoint:268            if token_uri or additional_claims or service_account_email or signer:269                raise exceptions.MalformedError(270                    "If use_metadata_identity_endpoint is set, token_uri, "271                    "additional_claims, service_account_email, signer arguments"272                    " must not be set"273                )274            self._token_uri = None275            self._additional_claims = None276            self._signer = None277 278        if service_account_email is None:279            sa_info = _metadata.get_service_account_info(request)280            self._service_account_email = sa_info["email"]281        else:282            self._service_account_email = service_account_email283 284        if not use_metadata_identity_endpoint:285            if signer is None:286                signer = iam.Signer(287                    request=request,288                    credentials=Credentials(),289                    service_account_email=self._service_account_email,290                )291            self._signer = signer292            self._token_uri = token_uri or _DEFAULT_TOKEN_URI293 294            if additional_claims is not None:295                self._additional_claims = additional_claims296            else:297                self._additional_claims = {}298 299    def with_target_audience(self, target_audience):300        """Create a copy of these credentials with the specified target301        audience.302        Args:303            target_audience (str): The intended audience for these credentials,304            used when requesting the ID Token.305        Returns:306            google.auth.service_account.IDTokenCredentials: A new credentials307                instance.308        """309        # since the signer is already instantiated,310        # the request is not needed311        if self._use_metadata_identity_endpoint:312            return self.__class__(313                None,314                target_audience=target_audience,315                use_metadata_identity_endpoint=True,316                quota_project_id=self._quota_project_id,317            )318        else:319            return self.__class__(320                None,321                service_account_email=self._service_account_email,322                token_uri=self._token_uri,323                target_audience=target_audience,324                additional_claims=self._additional_claims.copy(),325                signer=self.signer,326                use_metadata_identity_endpoint=False,327                quota_project_id=self._quota_project_id,328            )329 330    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)331    def with_quota_project(self, quota_project_id):332 333        # since the signer is already instantiated,334        # the request is not needed335        if self._use_metadata_identity_endpoint:336            return self.__class__(337                None,338                target_audience=self._target_audience,339                use_metadata_identity_endpoint=True,340                quota_project_id=quota_project_id,341            )342        else:343            return self.__class__(344                None,345                service_account_email=self._service_account_email,346                token_uri=self._token_uri,347                target_audience=self._target_audience,348                additional_claims=self._additional_claims.copy(),349                signer=self.signer,350                use_metadata_identity_endpoint=False,351                quota_project_id=quota_project_id,352            )353 354    @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)355    def with_token_uri(self, token_uri):356 357        # since the signer is already instantiated,358        # the request is not needed359        if self._use_metadata_identity_endpoint:360            raise exceptions.MalformedError(361                "If use_metadata_identity_endpoint is set, token_uri" " must not be set"362            )363        else:364            return self.__class__(365                None,366                service_account_email=self._service_account_email,367                token_uri=token_uri,368                target_audience=self._target_audience,369                additional_claims=self._additional_claims.copy(),370                signer=self.signer,371                use_metadata_identity_endpoint=False,372                quota_project_id=self.quota_project_id,373            )374 375    def _make_authorization_grant_assertion(self):376        """Create the OAuth 2.0 assertion.377        This assertion is used during the OAuth 2.0 grant to acquire an378        ID token.379        Returns:380            bytes: The authorization grant assertion.381        """382        now = _helpers.utcnow()383        lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)384        expiry = now + lifetime385 386        payload = {387            "iat": _helpers.datetime_to_secs(now),388            "exp": _helpers.datetime_to_secs(expiry),389            # The issuer must be the service account email.390            "iss": self.service_account_email,391            # The audience must be the auth token endpoint's URI392            "aud": self._token_uri,393            # The target audience specifies which service the ID token is394            # intended for.395            "target_audience": self._target_audience,396        }397 398        payload.update(self._additional_claims)399 400        token = jwt.encode(self._signer, payload)401 402        return token403 404    def _call_metadata_identity_endpoint(self, request):405        """Request ID token from metadata identity endpoint.406 407        Args:408            request (google.auth.transport.Request): The object used to make409                HTTP requests.410 411        Returns:412            Tuple[str, datetime.datetime]: The ID token and the expiry of the ID token.413 414        Raises:415            google.auth.exceptions.RefreshError: If the Compute Engine metadata416                service can't be reached or if the instance has no credentials.417            ValueError: If extracting expiry from the obtained ID token fails.418        """419        try:420            path = "instance/service-accounts/default/identity"421            params = {"audience": self._target_audience, "format": "full"}422            metrics_header = {423                metrics.API_CLIENT_HEADER: metrics.token_request_id_token_mds()424            }425            id_token = _metadata.get(426                request, path, params=params, headers=metrics_header427            )428        except exceptions.TransportError as caught_exc:429            new_exc = exceptions.RefreshError(caught_exc)430            raise new_exc from caught_exc431 432        _, payload, _, _ = jwt._unverified_decode(id_token)433        return id_token, datetime.datetime.utcfromtimestamp(payload["exp"])434 435    def refresh(self, request):436        """Refreshes the ID token.437 438        Args:439            request (google.auth.transport.Request): The object used to make440                HTTP requests.441 442        Raises:443            google.auth.exceptions.RefreshError: If the credentials could444                not be refreshed.445            ValueError: If extracting expiry from the obtained ID token fails.446        """447        if self._use_metadata_identity_endpoint:448            self.token, self.expiry = self._call_metadata_identity_endpoint(request)449        else:450            assertion = self._make_authorization_grant_assertion()451            access_token, expiry, _ = _client.id_token_jwt_grant(452                request, self._token_uri, assertion453            )454            self.token = access_token455            self.expiry = expiry456 457    @property  # type: ignore458    @_helpers.copy_docstring(credentials.Signing)459    def signer(self):460        return self._signer461 462    def sign_bytes(self, message):463        """Signs the given message.464 465        Args:466            message (bytes): The message to sign.467 468        Returns:469            bytes: The message's cryptographic signature.470 471        Raises:472            ValueError:473                Signer is not available if metadata identity endpoint is used.474        """475        if self._use_metadata_identity_endpoint:476            raise exceptions.InvalidOperation(477                "Signer is not available if metadata identity endpoint is used"478            )479        return self._signer.sign(message)480 481    @property482    def service_account_email(self):483        """The service account email."""484        return self._service_account_email485 486    @property487    def signer_email(self):488        return self._service_account_email489