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"""AWS Credentials and AWS Signature V4 Request Signer.16 17This module provides credentials to access Google Cloud resources from Amazon18Web Services (AWS) workloads. These credentials are recommended over the19use of service account credentials in AWS as they do not involve the management20of long-live service account private keys.21 22AWS Credentials are initialized using external_account arguments which are23typically loaded from the external credentials JSON file.24 25This module also provides a definition for an abstract AWS security credentials supplier.26This supplier can be implemented to return valid AWS security credentials and an AWS region27and used to create AWS credentials. The credentials will then call the28supplier instead of using pre-defined methods such as calling the EC2 metadata endpoints.29 30This module also provides a basic implementation of the31`AWS Signature Version 4`_ request signing algorithm.32 33AWS Credentials use serialized signed requests to the34`AWS STS GetCallerIdentity`_ API that can be exchanged for Google access tokens35via the GCP STS endpoint.36 37.. _AWS Signature Version 4: https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html38.. _AWS STS GetCallerIdentity: https://docs.aws.amazon.com/STS/latest/APIReference/API_GetCallerIdentity.html39"""40 41import abc42from dataclasses import dataclass43import hashlib44import hmac45import http.client as http_client46import json47import os48import posixpath49import re50from typing import Optional51import urllib52from urllib.parse import urljoin53 54from google.auth import _helpers55from google.auth import environment_vars56from google.auth import exceptions57from google.auth import external_account58 59# AWS Signature Version 4 signing algorithm identifier.60_AWS_ALGORITHM = "AWS4-HMAC-SHA256"61# The termination string for the AWS credential scope value as defined in62# https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html63_AWS_REQUEST_TYPE = "aws4_request"64# The AWS authorization header name for the security session token if available.65_AWS_SECURITY_TOKEN_HEADER = "x-amz-security-token"66# The AWS authorization header name for the auto-generated date.67_AWS_DATE_HEADER = "x-amz-date"68# The default AWS regional credential verification URL.69_DEFAULT_AWS_REGIONAL_CREDENTIAL_VERIFICATION_URL = (70    "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15"71)72# IMDSV2 session token lifetime. This is set to a low value because the session token is used immediately.73_IMDSV2_SESSION_TOKEN_TTL_SECONDS = "300"74 75 76class RequestSigner(object):77    """Implements an AWS request signer based on the AWS Signature Version 4 signing78    process.79    https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html80    """81 82    def __init__(self, region_name):83        """Instantiates an AWS request signer used to compute authenticated signed84        requests to AWS APIs based on the AWS Signature Version 4 signing process.85 86        Args:87            region_name (str): The AWS region to use.88        """89 90        self._region_name = region_name91 92    def get_request_options(93        self,94        aws_security_credentials,95        url,96        method,97        request_payload="",98        additional_headers={},99    ):100        """Generates the signed request for the provided HTTP request for calling101        an AWS API. This follows the steps described at:102        https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html103 104        Args:105            aws_security_credentials (AWSSecurityCredentials): The AWS security credentials.106            url (str): The AWS service URL containing the canonical URI and107                query string.108            method (str): The HTTP method used to call this API.109            request_payload (Optional[str]): The optional request payload if110                available.111            additional_headers (Optional[Mapping[str, str]]): The optional112                additional headers needed for the requested AWS API.113 114        Returns:115            Mapping[str, str]: The AWS signed request dictionary object.116        """117 118        additional_headers = additional_headers or {}119 120        uri = urllib.parse.urlparse(url)121        # Normalize the URL path. This is needed for the canonical_uri.122        # os.path.normpath can't be used since it normalizes "/" paths123        # to "\\" in Windows OS.124        normalized_uri = urllib.parse.urlparse(125            urljoin(url, posixpath.normpath(uri.path))126        )127        # Validate provided URL.128        if not uri.hostname or uri.scheme != "https":129            raise exceptions.InvalidResource("Invalid AWS service URL")130 131        header_map = _generate_authentication_header_map(132            host=uri.hostname,133            canonical_uri=normalized_uri.path or "/",134            canonical_querystring=_get_canonical_querystring(uri.query),135            method=method,136            region=self._region_name,137            aws_security_credentials=aws_security_credentials,138            request_payload=request_payload,139            additional_headers=additional_headers,140        )141        headers = {142            "Authorization": header_map.get("authorization_header"),143            "host": uri.hostname,144        }145        # Add x-amz-date if available.146        if "amz_date" in header_map:147            headers[_AWS_DATE_HEADER] = header_map.get("amz_date")148        # Append additional optional headers, eg. X-Amz-Target, Content-Type, etc.149        for key in additional_headers:150            headers[key] = additional_headers[key]151 152        # Add session token if available.153        if aws_security_credentials.session_token is not None:154            headers[_AWS_SECURITY_TOKEN_HEADER] = aws_security_credentials.session_token155 156        signed_request = {"url": url, "method": method, "headers": headers}157        if request_payload:158            signed_request["data"] = request_payload159        return signed_request160 161 162def _get_canonical_querystring(query):163    """Generates the canonical query string given a raw query string.164    Logic is based on165    https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html166 167    Args:168        query (str): The raw query string.169 170    Returns:171        str: The canonical query string.172    """173    # Parse raw query string.174    querystring = urllib.parse.parse_qs(query)175    querystring_encoded_map = {}176    for key in querystring:177        quote_key = urllib.parse.quote(key, safe="-_.~")178        # URI encode key.179        querystring_encoded_map[quote_key] = []180        for item in querystring[key]:181            # For each key, URI encode all values for that key.182            querystring_encoded_map[quote_key].append(183                urllib.parse.quote(item, safe="-_.~")184            )185        # Sort values for each key.186        querystring_encoded_map[quote_key].sort()187    # Sort keys.188    sorted_keys = list(querystring_encoded_map.keys())189    sorted_keys.sort()190    # Reconstruct the query string. Preserve keys with multiple values.191    querystring_encoded_pairs = []192    for key in sorted_keys:193        for item in querystring_encoded_map[key]:194            querystring_encoded_pairs.append("{}={}".format(key, item))195    return "&".join(querystring_encoded_pairs)196 197 198def _sign(key, msg):199    """Creates the HMAC-SHA256 hash of the provided message using the provided200    key.201 202    Args:203        key (str): The HMAC-SHA256 key to use.204        msg (str): The message to hash.205 206    Returns:207        str: The computed hash bytes.208    """209    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()210 211 212def _get_signing_key(key, date_stamp, region_name, service_name):213    """Calculates the signing key used to calculate the signature for214    AWS Signature Version 4 based on:215    https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html216 217    Args:218        key (str): The AWS secret access key.219        date_stamp (str): The '%Y%m%d' date format.220        region_name (str): The AWS region.221        service_name (str): The AWS service name, eg. sts.222 223    Returns:224        str: The signing key bytes.225    """226    k_date = _sign(("AWS4" + key).encode("utf-8"), date_stamp)227    k_region = _sign(k_date, region_name)228    k_service = _sign(k_region, service_name)229    k_signing = _sign(k_service, "aws4_request")230    return k_signing231 232 233def _generate_authentication_header_map(234    host,235    canonical_uri,236    canonical_querystring,237    method,238    region,239    aws_security_credentials,240    request_payload="",241    additional_headers={},242):243    """Generates the authentication header map needed for generating the AWS244    Signature Version 4 signed request.245 246    Args:247        host (str): The AWS service URL hostname.248        canonical_uri (str): The AWS service URL path name.249        canonical_querystring (str): The AWS service URL query string.250        method (str): The HTTP method used to call this API.251        region (str): The AWS region.252        aws_security_credentials (AWSSecurityCredentials): The AWS security credentials.253        request_payload (Optional[str]): The optional request payload if254            available.255        additional_headers (Optional[Mapping[str, str]]): The optional256            additional headers needed for the requested AWS API.257 258    Returns:259        Mapping[str, str]: The AWS authentication header dictionary object.260            This contains the x-amz-date and authorization header information.261    """262    # iam.amazonaws.com host => iam service.263    # sts.us-east-2.amazonaws.com host => sts service.264    service_name = host.split(".")[0]265 266    current_time = _helpers.utcnow()267    amz_date = current_time.strftime("%Y%m%dT%H%M%SZ")268    date_stamp = current_time.strftime("%Y%m%d")269 270    # Change all additional headers to be lower case.271    full_headers = {}272    for key in additional_headers:273        full_headers[key.lower()] = additional_headers[key]274    # Add AWS session token if available.275    if aws_security_credentials.session_token is not None:276        full_headers[277            _AWS_SECURITY_TOKEN_HEADER278        ] = aws_security_credentials.session_token279 280    # Required headers281    full_headers["host"] = host282    # Do not use generated x-amz-date if the date header is provided.283    # Previously the date was not fixed with x-amz- and could be provided284    # manually.285    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-header-value-trim.req286    if "date" not in full_headers:287        full_headers[_AWS_DATE_HEADER] = amz_date288 289    # Header keys need to be sorted alphabetically.290    canonical_headers = ""291    header_keys = list(full_headers.keys())292    header_keys.sort()293    for key in header_keys:294        canonical_headers = "{}{}:{}\n".format(295            canonical_headers, key, full_headers[key]296        )297    signed_headers = ";".join(header_keys)298 299    payload_hash = hashlib.sha256((request_payload or "").encode("utf-8")).hexdigest()300 301    # https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html302    canonical_request = "{}\n{}\n{}\n{}\n{}\n{}".format(303        method,304        canonical_uri,305        canonical_querystring,306        canonical_headers,307        signed_headers,308        payload_hash,309    )310 311    credential_scope = "{}/{}/{}/{}".format(312        date_stamp, region, service_name, _AWS_REQUEST_TYPE313    )314 315    # https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html316    string_to_sign = "{}\n{}\n{}\n{}".format(317        _AWS_ALGORITHM,318        amz_date,319        credential_scope,320        hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),321    )322 323    # https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html324    signing_key = _get_signing_key(325        aws_security_credentials.secret_access_key, date_stamp, region, service_name326    )327    signature = hmac.new(328        signing_key, string_to_sign.encode("utf-8"), hashlib.sha256329    ).hexdigest()330 331    # https://docs.aws.amazon.com/general/latest/gr/sigv4-add-signature-to-request.html332    authorization_header = "{} Credential={}/{}, SignedHeaders={}, Signature={}".format(333        _AWS_ALGORITHM,334        aws_security_credentials.access_key_id,335        credential_scope,336        signed_headers,337        signature,338    )339 340    authentication_header = {"authorization_header": authorization_header}341    # Do not use generated x-amz-date if the date header is provided.342    if "date" not in full_headers:343        authentication_header["amz_date"] = amz_date344    return authentication_header345 346 347@dataclass348class AwsSecurityCredentials:349    """A class that models AWS security credentials with an optional session token.350 351        Attributes:352            access_key_id (str): The AWS security credentials access key id.353            secret_access_key (str): The AWS security credentials secret access key.354            session_token (Optional[str]): The optional AWS security credentials session token. This should be set when using temporary credentials.355    """356 357    access_key_id: str358    secret_access_key: str359    session_token: Optional[str] = None360 361 362class AwsSecurityCredentialsSupplier(metaclass=abc.ABCMeta):363    """Base class for AWS security credential suppliers. This can be implemented with custom logic to retrieve364    AWS security credentials to exchange for a Google Cloud access token. The AWS external account credential does365    not cache the AWS security credentials, so caching logic should be added in the implementation.366    """367 368    @abc.abstractmethod369    def get_aws_security_credentials(self, context, request):370        """Returns the AWS security credentials for the requested context.371 372        .. warning: This is not cached by the calling Google credential, so caching logic should be implemented in the supplier.373 374        Args:375            context (google.auth.externalaccount.SupplierContext): The context object376                containing information about the requested audience and subject token type.377            request (google.auth.transport.Request): The object used to make378                HTTP requests.379 380        Raises:381            google.auth.exceptions.RefreshError: If an error is encountered during382                security credential retrieval logic.383 384        Returns:385            AwsSecurityCredentials: The requested AWS security credentials.386        """387        raise NotImplementedError("")388 389    @abc.abstractmethod390    def get_aws_region(self, context, request):391        """Returns the AWS region for the requested context.392 393        Args:394            context (google.auth.externalaccount.SupplierContext): The context object395                containing information about the requested audience and subject token type.396            request (google.auth.transport.Request): The object used to make397                HTTP requests.398 399        Raises:400            google.auth.exceptions.RefreshError: If an error is encountered during401                region retrieval logic.402 403        Returns:404            str: The AWS region.405        """406        raise NotImplementedError("")407 408 409class _DefaultAwsSecurityCredentialsSupplier(AwsSecurityCredentialsSupplier):410    """Default implementation of AWS security credentials supplier. Supports retrieving411    credentials and region via EC2 metadata endpoints and environment variables.412    """413 414    def __init__(self, credential_source):415        self._region_url = credential_source.get("region_url")416        self._security_credentials_url = credential_source.get("url")417        self._imdsv2_session_token_url = credential_source.get(418            "imdsv2_session_token_url"419        )420 421    @_helpers.copy_docstring(AwsSecurityCredentialsSupplier)422    def get_aws_security_credentials(self, context, request):423 424        # Check environment variables for permanent credentials first.425        # https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html426        env_aws_access_key_id = os.environ.get(environment_vars.AWS_ACCESS_KEY_ID)427        env_aws_secret_access_key = os.environ.get(428            environment_vars.AWS_SECRET_ACCESS_KEY429        )430        # This is normally not available for permanent credentials.431        env_aws_session_token = os.environ.get(environment_vars.AWS_SESSION_TOKEN)432        if env_aws_access_key_id and env_aws_secret_access_key:433            return AwsSecurityCredentials(434                env_aws_access_key_id, env_aws_secret_access_key, env_aws_session_token435            )436 437        imdsv2_session_token = self._get_imdsv2_session_token(request)438        role_name = self._get_metadata_role_name(request, imdsv2_session_token)439 440        # Get security credentials.441        credentials = self._get_metadata_security_credentials(442            request, role_name, imdsv2_session_token443        )444 445        return AwsSecurityCredentials(446            credentials.get("AccessKeyId"),447            credentials.get("SecretAccessKey"),448            credentials.get("Token"),449        )450 451    @_helpers.copy_docstring(AwsSecurityCredentialsSupplier)452    def get_aws_region(self, context, request):453        # The AWS metadata server is not available in some AWS environments454        # such as AWS lambda. Instead, it is available via environment455        # variable.456        env_aws_region = os.environ.get(environment_vars.AWS_REGION)457        if env_aws_region is not None:458            return env_aws_region459 460        env_aws_region = os.environ.get(environment_vars.AWS_DEFAULT_REGION)461        if env_aws_region is not None:462            return env_aws_region463 464        if not self._region_url:465            raise exceptions.RefreshError("Unable to determine AWS region")466 467        headers = None468        imdsv2_session_token = self._get_imdsv2_session_token(request)469        if imdsv2_session_token is not None:470            headers = {"X-aws-ec2-metadata-token": imdsv2_session_token}471 472        response = request(url=self._region_url, method="GET", headers=headers)473 474        # Support both string and bytes type response.data.475        response_body = (476            response.data.decode("utf-8")477            if hasattr(response.data, "decode")478            else response.data479        )480 481        if response.status != http_client.OK:482            raise exceptions.RefreshError(483                "Unable to retrieve AWS region: {}".format(response_body)484            )485 486        # This endpoint will return the region in format: us-east-2b.487        # Only the us-east-2 part should be used.488        return response_body[:-1]489 490    def _get_imdsv2_session_token(self, request):491        if request is not None and self._imdsv2_session_token_url is not None:492            headers = {493                "X-aws-ec2-metadata-token-ttl-seconds": _IMDSV2_SESSION_TOKEN_TTL_SECONDS494            }495 496            imdsv2_session_token_response = request(497                url=self._imdsv2_session_token_url, method="PUT", headers=headers498            )499 500            if imdsv2_session_token_response.status != http_client.OK:501                raise exceptions.RefreshError(502                    "Unable to retrieve AWS Session Token: {}".format(503                        imdsv2_session_token_response.data504                    )505                )506 507            return imdsv2_session_token_response.data508        else:509            return None510 511    def _get_metadata_security_credentials(512        self, request, role_name, imdsv2_session_token513    ):514        """Retrieves the AWS security credentials required for signing AWS515        requests from the AWS metadata server.516 517        Args:518            request (google.auth.transport.Request): A callable used to make519                HTTP requests.520            role_name (str): The AWS role name required by the AWS metadata521                server security_credentials endpoint in order to return the522                credentials.523            imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a524                header in the requests to AWS metadata endpoint.525 526        Returns:527            Mapping[str, str]: The AWS metadata server security credentials528                response.529 530        Raises:531            google.auth.exceptions.RefreshError: If an error occurs while532                retrieving the AWS security credentials.533        """534        headers = {"Content-Type": "application/json"}535        if imdsv2_session_token is not None:536            headers["X-aws-ec2-metadata-token"] = imdsv2_session_token537 538        response = request(539            url="{}/{}".format(self._security_credentials_url, role_name),540            method="GET",541            headers=headers,542        )543 544        # support both string and bytes type response.data545        response_body = (546            response.data.decode("utf-8")547            if hasattr(response.data, "decode")548            else response.data549        )550 551        if response.status != http_client.OK:552            raise exceptions.RefreshError(553                "Unable to retrieve AWS security credentials: {}".format(response_body)554            )555 556        credentials_response = json.loads(response_body)557 558        return credentials_response559 560    def _get_metadata_role_name(self, request, imdsv2_session_token):561        """Retrieves the AWS role currently attached to the current AWS562        workload by querying the AWS metadata server. This is needed for the563        AWS metadata server security credentials endpoint in order to retrieve564        the AWS security credentials needed to sign requests to AWS APIs.565 566        Args:567            request (google.auth.transport.Request): A callable used to make568                HTTP requests.569            imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a570                header in the requests to AWS metadata endpoint.571 572        Returns:573            str: The AWS role name.574 575        Raises:576            google.auth.exceptions.RefreshError: If an error occurs while577                retrieving the AWS role name.578        """579        if self._security_credentials_url is None:580            raise exceptions.RefreshError(581                "Unable to determine the AWS metadata server security credentials endpoint"582            )583 584        headers = None585        if imdsv2_session_token is not None:586            headers = {"X-aws-ec2-metadata-token": imdsv2_session_token}587 588        response = request(589            url=self._security_credentials_url, method="GET", headers=headers590        )591 592        # support both string and bytes type response.data593        response_body = (594            response.data.decode("utf-8")595            if hasattr(response.data, "decode")596            else response.data597        )598 599        if response.status != http_client.OK:600            raise exceptions.RefreshError(601                "Unable to retrieve AWS role name {}".format(response_body)602            )603 604        return response_body605 606 607class Credentials(external_account.Credentials):608    """AWS external account credentials.609    This is used to exchange serialized AWS signature v4 signed requests to610    AWS STS GetCallerIdentity service for Google access tokens.611    """612 613    def __init__(614        self,615        audience,616        subject_token_type,617        token_url=external_account._DEFAULT_TOKEN_URL,618        credential_source=None,619        aws_security_credentials_supplier=None,620        *args,621        **kwargs622    ):623        """Instantiates an AWS workload external account credentials object.624 625        Args:626            audience (str): The STS audience field.627            subject_token_type (str): The subject token type based on the Oauth2.0 token exchange spec.628                Expected values include::629 630                    “urn:ietf:params:aws:token-type:aws4_request”631 632            token_url (Optional [str]): The STS endpoint URL. If not provided, will default to "https://sts.googleapis.com/v1/token".633            credential_source (Optional [Mapping]): The credential source dictionary used634                to provide instructions on how to retrieve external credential to be exchanged for Google access tokens.635                Either a credential source or an AWS security credentials supplier must be provided.636 637                Example credential_source for AWS credential::638 639                    {640                        "environment_id": "aws1",641                        "regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",642                        "region_url": "http://169.254.169.254/latest/meta-data/placement/availability-zone",643                        "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials",644                        imdsv2_session_token_url": "http://169.254.169.254/latest/api/token"645                    }646 647            aws_security_credentials_supplier (Optional [AwsSecurityCredentialsSupplier]): Optional AWS security credentials supplier.648                This will be called to supply valid AWS security credentails which will then649                be exchanged for Google access tokens. Either an AWS security credentials supplier650                or a credential source must be provided.651            args (List): Optional positional arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method.652            kwargs (Mapping): Optional keyword arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method.653 654        Raises:655            google.auth.exceptions.RefreshError: If an error is encountered during656                access token retrieval logic.657            ValueError: For invalid parameters.658 659        .. note:: Typically one of the helper constructors660            :meth:`from_file` or661            :meth:`from_info` are used instead of calling the constructor directly.662        """663        super(Credentials, self).__init__(664            audience=audience,665            subject_token_type=subject_token_type,666            token_url=token_url,667            credential_source=credential_source,668            *args,669            **kwargs670        )671        if credential_source is None and aws_security_credentials_supplier is None:672            raise exceptions.InvalidValue(673                "A valid credential source or AWS security credentials supplier must be provided."674            )675        if (676            credential_source is not None677            and aws_security_credentials_supplier is not None678        ):679            raise exceptions.InvalidValue(680                "AWS credential cannot have both a credential source and an AWS security credentials supplier."681            )682 683        if aws_security_credentials_supplier:684            self._aws_security_credentials_supplier = aws_security_credentials_supplier685            # The regional cred verification URL would normally be provided through the credential source. So set it to the default one here.686            self._cred_verification_url = (687                _DEFAULT_AWS_REGIONAL_CREDENTIAL_VERIFICATION_URL688            )689        else:690            environment_id = credential_source.get("environment_id") or ""691            self._aws_security_credentials_supplier = _DefaultAwsSecurityCredentialsSupplier(692                credential_source693            )694            self._cred_verification_url = credential_source.get(695                "regional_cred_verification_url"696            )697 698            # Get the environment ID, i.e. "aws1". Currently, only one version supported (1).699            matches = re.match(r"^(aws)([\d]+)$", environment_id)700            if matches:701                env_id, env_version = matches.groups()702            else:703                env_id, env_version = (None, None)704 705            if env_id != "aws" or self._cred_verification_url is None:706                raise exceptions.InvalidResource(707                    "No valid AWS 'credential_source' provided"708                )709            elif env_version is None or int(env_version) != 1:710                raise exceptions.InvalidValue(711                    "aws version '{}' is not supported in the current build.".format(712                        env_version713                    )714                )715 716        self._target_resource = audience717        self._request_signer = None718 719    def retrieve_subject_token(self, request):720        """Retrieves the subject token using the credential_source object.721        The subject token is a serialized `AWS GetCallerIdentity signed request`_.722 723        The logic is summarized as:724 725        Retrieve the AWS region from the AWS_REGION or AWS_DEFAULT_REGION726        environment variable or from the AWS metadata server availability-zone727        if not found in the environment variable.728 729        Check AWS credentials in environment variables. If not found, retrieve730        from the AWS metadata server security-credentials endpoint.731 732        When retrieving AWS credentials from the metadata server733        security-credentials endpoint, the AWS role needs to be determined by734        calling the security-credentials endpoint without any argument. Then the735        credentials can be retrieved via: security-credentials/role_name736 737        Generate the signed request to AWS STS GetCallerIdentity action.738 739        Inject x-goog-cloud-target-resource into header and serialize the740        signed request. This will be the subject-token to pass to GCP STS.741 742        .. _AWS GetCallerIdentity signed request:743            https://cloud.google.com/iam/docs/access-resources-aws#exchange-token744 745        Args:746            request (google.auth.transport.Request): A callable used to make747                HTTP requests.748        Returns:749            str: The retrieved subject token.750        """751 752        # Initialize the request signer if not yet initialized after determining753        # the current AWS region.754        if self._request_signer is None:755            self._region = self._aws_security_credentials_supplier.get_aws_region(756                self._supplier_context, request757            )758            self._request_signer = RequestSigner(self._region)759 760        # Retrieve the AWS security credentials needed to generate the signed761        # request.762        aws_security_credentials = self._aws_security_credentials_supplier.get_aws_security_credentials(763            self._supplier_context, request764        )765        # Generate the signed request to AWS STS GetCallerIdentity API.766        # Use the required regional endpoint. Otherwise, the request will fail.767        request_options = self._request_signer.get_request_options(768            aws_security_credentials,769            self._cred_verification_url.replace("{region}", self._region),770            "POST",771        )772        # The GCP STS endpoint expects the headers to be formatted as:773        # [774        #   {key: 'x-amz-date', value: '...'},775        #   {key: 'Authorization', value: '...'},776        #   ...777        # ]778        # And then serialized as:779        # quote(json.dumps({780        #   url: '...',781        #   method: 'POST',782        #   headers: [{key: 'x-amz-date', value: '...'}, ...]783        # }))784        request_headers = request_options.get("headers")785        # The full, canonical resource name of the workload identity pool786        # provider, with or without the HTTPS prefix.787        # Including this header as part of the signature is recommended to788        # ensure data integrity.789        request_headers["x-goog-cloud-target-resource"] = self._target_resource790 791        # Serialize AWS signed request.792        aws_signed_req = {}793        aws_signed_req["url"] = request_options.get("url")794        aws_signed_req["method"] = request_options.get("method")795        aws_signed_req["headers"] = []796        # Reformat header to GCP STS expected format.797        for key in request_headers.keys():798            aws_signed_req["headers"].append(799                {"key": key, "value": request_headers[key]}800            )801 802        return urllib.parse.quote(803            json.dumps(aws_signed_req, separators=(",", ":"), sort_keys=True)804        )805 806    def _create_default_metrics_options(self):807        metrics_options = super(Credentials, self)._create_default_metrics_options()808        metrics_options["source"] = "aws"809        if self._has_custom_supplier():810            metrics_options["source"] = "programmatic"811        return metrics_options812 813    def _has_custom_supplier(self):814        return self._credential_source is None815 816    def _constructor_args(self):817        args = super(Credentials, self)._constructor_args()818        # If a custom supplier was used, append it to the args dict.819        if self._has_custom_supplier():820            args.update(821                {822                    "aws_security_credentials_supplier": self._aws_security_credentials_supplier823                }824            )825        return args826 827    @classmethod828    def from_info(cls, info, **kwargs):829        """Creates an AWS Credentials instance from parsed external account info.830 831        Args:832            info (Mapping[str, str]): The AWS external account info in Google833                format.834            kwargs: Additional arguments to pass to the constructor.835 836        Returns:837            google.auth.aws.Credentials: The constructed credentials.838 839        Raises:840            ValueError: For invalid parameters.841        """842        aws_security_credentials_supplier = info.get(843            "aws_security_credentials_supplier"844        )845        kwargs.update(846            {"aws_security_credentials_supplier": aws_security_credentials_supplier}847        )848        return super(Credentials, cls).from_info(info, **kwargs)849 850    @classmethod851    def from_file(cls, filename, **kwargs):852        """Creates an AWS Credentials instance from an external account json file.853 854        Args:855            filename (str): The path to the AWS external account json file.856            kwargs: Additional arguments to pass to the constructor.857 858        Returns:859            google.auth.aws.Credentials: The constructed credentials.860        """861        return super(Credentials, cls).from_file(filename, **kwargs)862