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"""Pluggable Credentials.16Pluggable Credentials are initialized using external_account arguments which17are typically loaded from third-party executables. Unlike other18credentials that can be initialized with a list of explicit arguments, secrets19or credentials, external account clients use the environment and hints/guidelines20provided by the external_account JSON file to retrieve credentials and exchange21them for Google access tokens.22 23Example credential_source for pluggable credential:24{25    "executable": {26        "command": "/path/to/get/credentials.sh --arg1=value1 --arg2=value2",27        "timeout_millis": 5000,28        "output_file": "/path/to/generated/cached/credentials"29    }30}31"""32 33try:34    from collections.abc import Mapping35# Python 2.7 compatibility36except ImportError:  # pragma: NO COVER37    from collections import Mapping38import json39import os40import subprocess41import sys42import time43 44from google.auth import _helpers45from google.auth import exceptions46from google.auth import external_account47 48# The max supported executable spec version.49EXECUTABLE_SUPPORTED_MAX_VERSION = 150 51EXECUTABLE_TIMEOUT_MILLIS_DEFAULT = 30 * 1000  # 30 seconds52EXECUTABLE_TIMEOUT_MILLIS_LOWER_BOUND = 5 * 1000  # 5 seconds53EXECUTABLE_TIMEOUT_MILLIS_UPPER_BOUND = 120 * 1000  # 2 minutes54 55EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_LOWER_BOUND = 30 * 1000  # 30 seconds56EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_UPPER_BOUND = 30 * 60 * 1000  # 30 minutes57 58 59class Credentials(external_account.Credentials):60    """External account credentials sourced from executables."""61 62    def __init__(63        self,64        audience,65        subject_token_type,66        token_url,67        credential_source,68        *args,69        **kwargs70    ):71        """Instantiates an external account credentials object from a executables.72 73        Args:74            audience (str): The STS audience field.75            subject_token_type (str): The subject token type.76            token_url (str): The STS endpoint URL.77            credential_source (Mapping): The credential source dictionary used to78                provide instructions on how to retrieve external credential to be79                exchanged for Google access tokens.80 81                Example credential_source for pluggable credential:82 83                    {84                        "executable": {85                            "command": "/path/to/get/credentials.sh --arg1=value1 --arg2=value2",86                            "timeout_millis": 5000,87                            "output_file": "/path/to/generated/cached/credentials"88                        }89                    }90            args (List): Optional positional arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method.91            kwargs (Mapping): Optional keyword arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method.92 93        Raises:94            google.auth.exceptions.RefreshError: If an error is encountered during95                access token retrieval logic.96            google.auth.exceptions.InvalidValue: For invalid parameters.97            google.auth.exceptions.MalformedError: For invalid parameters.98 99        .. note:: Typically one of the helper constructors100            :meth:`from_file` or101            :meth:`from_info` are used instead of calling the constructor directly.102        """103 104        self.interactive = kwargs.pop("interactive", False)105        super(Credentials, self).__init__(106            audience=audience,107            subject_token_type=subject_token_type,108            token_url=token_url,109            credential_source=credential_source,110            *args,111            **kwargs112        )113        if not isinstance(credential_source, Mapping):114            self._credential_source_executable = None115            raise exceptions.MalformedError(116                "Missing credential_source. The credential_source is not a dict."117            )118        self._credential_source_executable = credential_source.get("executable")119        if not self._credential_source_executable:120            raise exceptions.MalformedError(121                "Missing credential_source. An 'executable' must be provided."122            )123        self._credential_source_executable_command = self._credential_source_executable.get(124            "command"125        )126        self._credential_source_executable_timeout_millis = self._credential_source_executable.get(127            "timeout_millis"128        )129        self._credential_source_executable_interactive_timeout_millis = self._credential_source_executable.get(130            "interactive_timeout_millis"131        )132        self._credential_source_executable_output_file = self._credential_source_executable.get(133            "output_file"134        )135 136        # Dummy value. This variable is only used via injection, not exposed to ctor137        self._tokeninfo_username = ""138 139        if not self._credential_source_executable_command:140            raise exceptions.MalformedError(141                "Missing command field. Executable command must be provided."142            )143        if not self._credential_source_executable_timeout_millis:144            self._credential_source_executable_timeout_millis = (145                EXECUTABLE_TIMEOUT_MILLIS_DEFAULT146            )147        elif (148            self._credential_source_executable_timeout_millis149            < EXECUTABLE_TIMEOUT_MILLIS_LOWER_BOUND150            or self._credential_source_executable_timeout_millis151            > EXECUTABLE_TIMEOUT_MILLIS_UPPER_BOUND152        ):153            raise exceptions.InvalidValue("Timeout must be between 5 and 120 seconds.")154 155        if self._credential_source_executable_interactive_timeout_millis:156            if (157                self._credential_source_executable_interactive_timeout_millis158                < EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_LOWER_BOUND159                or self._credential_source_executable_interactive_timeout_millis160                > EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_UPPER_BOUND161            ):162                raise exceptions.InvalidValue(163                    "Interactive timeout must be between 30 seconds and 30 minutes."164                )165 166    @_helpers.copy_docstring(external_account.Credentials)167    def retrieve_subject_token(self, request):168        self._validate_running_mode()169 170        # Check output file.171        if self._credential_source_executable_output_file is not None:172            try:173                with open(174                    self._credential_source_executable_output_file, encoding="utf-8"175                ) as output_file:176                    response = json.load(output_file)177            except Exception:178                pass179            else:180                try:181                    # If the cached response is expired, _parse_subject_token will raise an error which will be ignored and we will call the executable again.182                    subject_token = self._parse_subject_token(response)183                    if (184                        "expiration_time" not in response185                    ):  # Always treat missing expiration_time as expired and proceed to executable run.186                        raise exceptions.RefreshError187                except (exceptions.MalformedError, exceptions.InvalidValue):188                    raise189                except exceptions.RefreshError:190                    pass191                else:192                    return subject_token193 194        if not _helpers.is_python_3():195            raise exceptions.RefreshError(196                "Pluggable auth is only supported for python 3.7+"197            )198 199        # Inject env vars.200        env = os.environ.copy()201        self._inject_env_variables(env)202        env["GOOGLE_EXTERNAL_ACCOUNT_REVOKE"] = "0"203 204        # Run executable.205        exe_timeout = (206            self._credential_source_executable_interactive_timeout_millis / 1000207            if self.interactive208            else self._credential_source_executable_timeout_millis / 1000209        )210        exe_stdin = sys.stdin if self.interactive else None211        exe_stdout = sys.stdout if self.interactive else subprocess.PIPE212        exe_stderr = sys.stdout if self.interactive else subprocess.STDOUT213 214        result = subprocess.run(215            self._credential_source_executable_command.split(),216            timeout=exe_timeout,217            stdin=exe_stdin,218            stdout=exe_stdout,219            stderr=exe_stderr,220            env=env,221        )222        if result.returncode != 0:223            raise exceptions.RefreshError(224                "Executable exited with non-zero return code {}. Error: {}".format(225                    result.returncode, result.stdout226                )227            )228 229        # Handle executable output.230        response = json.loads(result.stdout.decode("utf-8")) if result.stdout else None231        if not response and self._credential_source_executable_output_file is not None:232            response = json.load(233                open(self._credential_source_executable_output_file, encoding="utf-8")234            )235 236        subject_token = self._parse_subject_token(response)237        return subject_token238 239    def revoke(self, request):240        """Revokes the subject token using the credential_source object.241 242        Args:243            request (google.auth.transport.Request): A callable used to make244                HTTP requests.245        Raises:246            google.auth.exceptions.RefreshError: If the executable revocation247                not properly executed.248 249        """250        if not self.interactive:251            raise exceptions.InvalidValue(252                "Revoke is only enabled under interactive mode."253            )254        self._validate_running_mode()255 256        if not _helpers.is_python_3():257            raise exceptions.RefreshError(258                "Pluggable auth is only supported for python 3.7+"259            )260 261        # Inject variables262        env = os.environ.copy()263        self._inject_env_variables(env)264        env["GOOGLE_EXTERNAL_ACCOUNT_REVOKE"] = "1"265 266        # Run executable267        result = subprocess.run(268            self._credential_source_executable_command.split(),269            timeout=self._credential_source_executable_interactive_timeout_millis270            / 1000,271            stdout=subprocess.PIPE,272            stderr=subprocess.STDOUT,273            env=env,274        )275 276        if result.returncode != 0:277            raise exceptions.RefreshError(278                "Auth revoke failed on executable. Exit with non-zero return code {}. Error: {}".format(279                    result.returncode, result.stdout280                )281            )282 283        response = json.loads(result.stdout.decode("utf-8"))284        self._validate_revoke_response(response)285 286    @property287    def external_account_id(self):288        """Returns the external account identifier.289 290        When service account impersonation is used the identifier is the service291        account email.292 293        Without service account impersonation, this returns None, unless it is294        being used by the Google Cloud CLI which populates this field.295        """296 297        return self.service_account_email or self._tokeninfo_username298 299    @classmethod300    def from_info(cls, info, **kwargs):301        """Creates a Pluggable Credentials instance from parsed external account info.302 303        Args:304            info (Mapping[str, str]): The Pluggable external account info in Google305                format.306            kwargs: Additional arguments to pass to the constructor.307 308        Returns:309            google.auth.pluggable.Credentials: The constructed310                credentials.311 312        Raises:313            google.auth.exceptions.InvalidValue: For invalid parameters.314            google.auth.exceptions.MalformedError: For invalid parameters.315        """316        return super(Credentials, cls).from_info(info, **kwargs)317 318    @classmethod319    def from_file(cls, filename, **kwargs):320        """Creates an Pluggable Credentials instance from an external account json file.321 322        Args:323            filename (str): The path to the Pluggable external account json file.324            kwargs: Additional arguments to pass to the constructor.325 326        Returns:327            google.auth.pluggable.Credentials: The constructed328                credentials.329        """330        return super(Credentials, cls).from_file(filename, **kwargs)331 332    def _inject_env_variables(self, env):333        env["GOOGLE_EXTERNAL_ACCOUNT_AUDIENCE"] = self._audience334        env["GOOGLE_EXTERNAL_ACCOUNT_TOKEN_TYPE"] = self._subject_token_type335        env["GOOGLE_EXTERNAL_ACCOUNT_ID"] = self.external_account_id336        env["GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE"] = "1" if self.interactive else "0"337 338        if self._service_account_impersonation_url is not None:339            env[340                "GOOGLE_EXTERNAL_ACCOUNT_IMPERSONATED_EMAIL"341            ] = self.service_account_email342        if self._credential_source_executable_output_file is not None:343            env[344                "GOOGLE_EXTERNAL_ACCOUNT_OUTPUT_FILE"345            ] = self._credential_source_executable_output_file346 347    def _parse_subject_token(self, response):348        self._validate_response_schema(response)349        if not response["success"]:350            if "code" not in response or "message" not in response:351                raise exceptions.MalformedError(352                    "Error code and message fields are required in the response."353                )354            raise exceptions.RefreshError(355                "Executable returned unsuccessful response: code: {}, message: {}.".format(356                    response["code"], response["message"]357                )358            )359        if "expiration_time" in response and response["expiration_time"] < time.time():360            raise exceptions.RefreshError(361                "The token returned by the executable is expired."362            )363        if "token_type" not in response:364            raise exceptions.MalformedError(365                "The executable response is missing the token_type field."366            )367        if (368            response["token_type"] == "urn:ietf:params:oauth:token-type:jwt"369            or response["token_type"] == "urn:ietf:params:oauth:token-type:id_token"370        ):  # OIDC371            return response["id_token"]372        elif response["token_type"] == "urn:ietf:params:oauth:token-type:saml2":  # SAML373            return response["saml_response"]374        else:375            raise exceptions.RefreshError("Executable returned unsupported token type.")376 377    def _validate_revoke_response(self, response):378        self._validate_response_schema(response)379        if not response["success"]:380            raise exceptions.RefreshError("Revoke failed with unsuccessful response.")381 382    def _validate_response_schema(self, response):383        if "version" not in response:384            raise exceptions.MalformedError(385                "The executable response is missing the version field."386            )387        if response["version"] > EXECUTABLE_SUPPORTED_MAX_VERSION:388            raise exceptions.RefreshError(389                "Executable returned unsupported version {}.".format(390                    response["version"]391                )392            )393 394        if "success" not in response:395            raise exceptions.MalformedError(396                "The executable response is missing the success field."397            )398 399    def _validate_running_mode(self):400        env_allow_executables = os.environ.get(401            "GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES"402        )403        if env_allow_executables != "1":404            raise exceptions.MalformedError(405                "Executables need to be explicitly allowed (set GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES to '1') to run."406            )407 408        if self.interactive and not self._credential_source_executable_output_file:409            raise exceptions.MalformedError(410                "An output_file must be specified in the credential configuration for interactive mode."411            )412 413        if (414            self.interactive415            and not self._credential_source_executable_interactive_timeout_millis416        ):417            raise exceptions.InvalidOperation(418                "Interactive mode cannot run without an interactive timeout."419            )420 421        if self.interactive and not self.is_workforce_pool:422            raise exceptions.InvalidValue(423                "Interactive mode is only enabled for workforce pool."424            )425 426    def _create_default_metrics_options(self):427        metrics_options = super(Credentials, self)._create_default_metrics_options()428        metrics_options["source"] = "executable"429        return metrics_options430