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"""Experimental GDCH credentials support.16"""17 18import datetime19 20from google.auth import _helpers21from google.auth import _service_account_info22from google.auth import credentials23from google.auth import exceptions24from google.auth import jwt25from google.oauth2 import _client26 27 28TOKEN_EXCHANGE_TYPE = "urn:ietf:params:oauth:token-type:token-exchange"29ACCESS_TOKEN_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"30SERVICE_ACCOUNT_TOKEN_TYPE = "urn:k8s:params:oauth:token-type:serviceaccount"31JWT_LIFETIME = datetime.timedelta(seconds=3600)  # 1 hour32 33 34class ServiceAccountCredentials(credentials.Credentials):35    """Credentials for GDCH (`Google Distributed Cloud Hosted`_) for service36    account users.37 38    .. _Google Distributed Cloud Hosted:39        https://cloud.google.com/blog/topics/hybrid-cloud/\40            announcing-google-distributed-cloud-edge-and-hosted41 42    To create a GDCH service account credential, first create a JSON file of43    the following format::44 45        {46            "type": "gdch_service_account",47            "format_version": "1",48            "project": "<project name>",49            "private_key_id": "<key id>",50            "private_key": "-----BEGIN EC PRIVATE KEY-----\n<key bytes>\n-----END EC PRIVATE KEY-----\n",51            "name": "<service identity name>",52            "ca_cert_path": "<CA cert path>",53            "token_uri": "https://service-identity.<Domain>/authenticate"54        }55 56    The "format_version" field stands for the format of the JSON file. For now57    it is always "1". The `private_key_id` and `private_key` is used for signing.58    The `ca_cert_path` is used for token server TLS certificate verification.59 60    After the JSON file is created, set `GOOGLE_APPLICATION_CREDENTIALS` environment61    variable to the JSON file path, then use the following code to create the62    credential::63 64        import google.auth65 66        credential, _ = google.auth.default()67        credential = credential.with_gdch_audience("<the audience>")68 69    We can also create the credential directly::70 71        from google.oauth import gdch_credentials72 73        credential = gdch_credentials.ServiceAccountCredentials.from_service_account_file("<the json file path>")74        credential = credential.with_gdch_audience("<the audience>")75 76    The token is obtained in the following way. This class first creates a77    self signed JWT. It uses the `name` value as the `iss` and `sub` claim, and78    the `token_uri` as the `aud` claim, and signs the JWT with the `private_key`.79    It then sends the JWT to the `token_uri` to exchange a final token for80    `audience`.81    """82 83    def __init__(84        self, signer, service_identity_name, project, audience, token_uri, ca_cert_path85    ):86        """87        Args:88            signer (google.auth.crypt.Signer): The signer used to sign JWTs.89            service_identity_name (str): The service identity name. It will be90                used as the `iss` and `sub` claim in the self signed JWT.91            project (str): The project.92            audience (str): The audience for the final token.93            token_uri (str): The token server uri.94            ca_cert_path (str): The CA cert path for token server side TLS95                certificate verification. If the token server uses well known96                CA, then this parameter can be `None`.97        """98        super(ServiceAccountCredentials, self).__init__()99        self._signer = signer100        self._service_identity_name = service_identity_name101        self._project = project102        self._audience = audience103        self._token_uri = token_uri104        self._ca_cert_path = ca_cert_path105 106    def _create_jwt(self):107        now = _helpers.utcnow()108        expiry = now + JWT_LIFETIME109        iss_sub_value = "system:serviceaccount:{}:{}".format(110            self._project, self._service_identity_name111        )112 113        payload = {114            "iss": iss_sub_value,115            "sub": iss_sub_value,116            "aud": self._token_uri,117            "iat": _helpers.datetime_to_secs(now),118            "exp": _helpers.datetime_to_secs(expiry),119        }120 121        return _helpers.from_bytes(jwt.encode(self._signer, payload))122 123    @_helpers.copy_docstring(credentials.Credentials)124    def refresh(self, request):125        import google.auth.transport.requests126 127        if not isinstance(request, google.auth.transport.requests.Request):128            raise exceptions.RefreshError(129                "For GDCH service account credentials, request must be a google.auth.transport.requests.Request object"130            )131 132        # Create a self signed JWT, and do token exchange.133        jwt_token = self._create_jwt()134        request_body = {135            "grant_type": TOKEN_EXCHANGE_TYPE,136            "audience": self._audience,137            "requested_token_type": ACCESS_TOKEN_TOKEN_TYPE,138            "subject_token": jwt_token,139            "subject_token_type": SERVICE_ACCOUNT_TOKEN_TYPE,140        }141        response_data = _client._token_endpoint_request(142            request,143            self._token_uri,144            request_body,145            access_token=None,146            use_json=True,147            verify=self._ca_cert_path,148        )149 150        self.token, _, self.expiry, _ = _client._handle_refresh_grant_response(151            response_data, None152        )153 154    def with_gdch_audience(self, audience):155        """Create a copy of GDCH credentials with the specified audience.156 157        Args:158            audience (str): The intended audience for GDCH credentials.159        """160        return self.__class__(161            self._signer,162            self._service_identity_name,163            self._project,164            audience,165            self._token_uri,166            self._ca_cert_path,167        )168 169    @classmethod170    def _from_signer_and_info(cls, signer, info):171        """Creates a Credentials instance from a signer and service account172        info.173 174        Args:175            signer (google.auth.crypt.Signer): The signer used to sign JWTs.176            info (Mapping[str, str]): The service account info.177 178        Returns:179            google.oauth2.gdch_credentials.ServiceAccountCredentials: The constructed180                credentials.181 182        Raises:183            ValueError: If the info is not in the expected format.184        """185        if info["format_version"] != "1":186            raise ValueError("Only format version 1 is supported")187 188        return cls(189            signer,190            info["name"],  # service_identity_name191            info["project"],192            None,  # audience193            info["token_uri"],194            info.get("ca_cert_path", None),195        )196 197    @classmethod198    def from_service_account_info(cls, info):199        """Creates a Credentials instance from parsed service account info.200 201        Args:202            info (Mapping[str, str]): The service account info in Google203                format.204            kwargs: Additional arguments to pass to the constructor.205 206        Returns:207            google.oauth2.gdch_credentials.ServiceAccountCredentials: The constructed208                credentials.209 210        Raises:211            ValueError: If the info is not in the expected format.212        """213        signer = _service_account_info.from_dict(214            info,215            require=[216                "format_version",217                "private_key_id",218                "private_key",219                "name",220                "project",221                "token_uri",222            ],223            use_rsa_signer=False,224        )225        return cls._from_signer_and_info(signer, info)226 227    @classmethod228    def from_service_account_file(cls, filename):229        """Creates a Credentials instance from a service account json file.230 231        Args:232            filename (str): The path to the service account json file.233            kwargs: Additional arguments to pass to the constructor.234 235        Returns:236            google.oauth2.gdch_credentials.ServiceAccountCredentials: The constructed237                credentials.238        """239        info, signer = _service_account_info.from_filename(240            filename,241            require=[242                "format_version",243                "private_key_id",244                "private_key",245                "name",246                "project",247                "token_uri",248            ],249            use_rsa_signer=False,250        )251        return cls._from_signer_and_info(signer, info)252