(self, request)
return urllib.parse.quote(
            json.dumps(aws_signed_req, separators=(",", ":"), sort_keys=True)
        )
Retrieves the subject token using the credential_source object.
        The subject token is a serialized `AWS GetCallerIdentity signed request`_.

        The logic is summarized as:

        Retrieve the AWS region from the AWS_REGION or AWS_DEFAULT_REGION
        environment variable or from the AWS metadata server availability-zone
        if not found in the environment variable.

        Check AWS credentials in environment variables. If not found, retrieve
        from the AWS metadata server security-credentials endpoint.

        When retrieving AWS credentials from the metadata server
        security-credentials endpoint, the AWS role needs to be determined by
        calling the security-credentials endpoint without any argument. Then the
        credentials can be retrieved via: security-credentials/role_name

        Generate the signed request to AWS STS GetCallerIdentity action.

        Inject x-goog-cloud-target-resource into header and serialize the
        signed request. This will be the subject-token to pass to GCP STS.

        .. _AWS GetCallerIdentity signed request:
            https://cloud.google.com/iam/docs/access-resources-aws#exchange-token

        Args:
            request (google.auth.transport.Request): A callable used to make
                HTTP requests.
        Returns:
            str: The retrieved subject token.
Retrieves the subject token using the credential_source object.
        The subject token is a serialized `AWS GetCallerIdentity signed request`_.
Retrieves the subject token using the credential_source object . The subject token is a serialized AWS GetCallerIdentity signed request _ .
def retrieve_subject_token(self, request):
        """Retrieves the subject token using the credential_source object.
        The subject token is a serialized `AWS GetCallerIdentity signed request`_.

        The logic is summarized as:

        Retrieve the AWS region from the AWS_REGION or AWS_DEFAULT_REGION
        environment variable or from the AWS metadata server availability-zone
        if not found in the environment variable.

        Check AWS credentials in environment variables. If not found, retrieve
        from the AWS metadata server security-credentials endpoint.

        When retrieving AWS credentials from the metadata server
        security-credentials endpoint, the AWS role needs to be determined by
        calling the security-credentials endpoint without any argument. Then the
        credentials can be retrieved via: security-credentials/role_name

        Generate the signed request to AWS STS GetCallerIdentity action.

        Inject x-goog-cloud-target-resource into header and serialize the
        signed request. This will be the subject-token to pass to GCP STS.

        .. _AWS GetCallerIdentity signed request:
            https://cloud.google.com/iam/docs/access-resources-aws#exchange-token

        Args:
            request (google.auth.transport.Request): A callable used to make
                HTTP requests.
        Returns:
            str: The retrieved subject token.
        """
        # Initialize the request signer if not yet initialized after determining
        # the current AWS region.
        if self._request_signer is None:
            self._region = self._get_region(request, self._region_url)
            self._request_signer = RequestSigner(self._region)

        # Retrieve the AWS security credentials needed to generate the signed
        # request.
        aws_security_credentials = self._get_security_credentials(request)
        # Generate the signed request to AWS STS GetCallerIdentity API.
        # Use the required regional endpoint. Otherwise, the request will fail.
        request_options = self._request_signer.get_request_options(
            aws_security_credentials,
            self._cred_verification_url.replace("{region}", self._region),
            "POST",
        )
        # The GCP STS endpoint expects the headers to be formatted as:
        # [
        #   {key: 'x-amz-date', value: '...'},
        #   {key: 'Authorization', value: '...'},
        #   ...
        # ]
        # And then serialized as:
        # quote(json.dumps({
        #   url: '...',
        #   method: 'POST',
        #   headers: [{key: 'x-amz-date', value: '...'}, ...]
        # }))
        request_headers = request_options.get("headers")
        # The full, canonical resource name of the workload identity pool
        # provider, with or without the HTTPS prefix.
        # Including this header as part of the signature is recommended to
        # ensure data integrity.
        request_headers["x-goog-cloud-target-resource"] = self._target_resource

        # Serialize AWS signed request.
        # Keeping inner keys in sorted order makes testing easier for Python
        # versions <=3.5 as the stringified JSON string would have a predictable
        # key order.
        aws_signed_req = {}
        aws_signed_req["url"] = request_options.get("url")
        aws_signed_req["method"] = request_options.get("method")
        aws_signed_req["headers"] = []
        # Reformat header to GCP STS expected format.
        for key in sorted(request_headers.keys()):
            aws_signed_req["headers"].append(
                {"key": key, "value": request_headers[key]}
            )

        return urllib.parse.quote(
            json.dumps(aws_signed_req, separators=(",", ":"), sort_keys=True)
        )
def retrieve_subject_token ( self , request ) : # Initialize the request signer if not yet initialized after determining # the current AWS region. if self . _request_signer is None : self . _region = self . _get_region ( request , self . _region_url ) self . _request_signer = RequestSigner ( self . _region ) # Retrieve the AWS security credentials needed to generate the signed # request. aws_security_credentials = self . _get_security_credentials ( request ) # Generate the signed request to AWS STS GetCallerIdentity API. # Use the required regional endpoint. Otherwise, the request will fail. request_options = self . _request_signer . get_request_options ( aws_security_credentials , self . _cred_verification_url . replace ( "{region}" , self . _region ) , "POST" , ) # The GCP STS endpoint expects the headers to be formatted as: # [ #   {key: 'x-amz-date', value: '...'}, #   {key: 'Authorization', value: '...'}, #   ... # ] # And then serialized as: # quote(json.dumps({ #   url: '...', #   method: 'POST', #   headers: [{key: 'x-amz-date', value: '...'}, ...] # })) request_headers = request_options . get ( "headers" ) # The full, canonical resource name of the workload identity pool # provider, with or without the HTTPS prefix. # Including this header as part of the signature is recommended to # ensure data integrity. request_headers [ "x-goog-cloud-target-resource" ] = self . _target_resource # Serialize AWS signed request. # Keeping inner keys in sorted order makes testing easier for Python # versions <=3.5 as the stringified JSON string would have a predictable # key order. aws_signed_req = { } aws_signed_req [ "url" ] = request_options . get ( "url" ) aws_signed_req [ "method" ] = request_options . get ( "method" ) aws_signed_req [ "headers" ] = [ ] # Reformat header to GCP STS expected format. for key in sorted ( request_headers . keys ( ) ) : aws_signed_req [ "headers" ] . append ( { "key" : key , "value" : request_headers [ key ] } ) return urllib . parse . quote ( json . dumps ( aws_signed_req , separators = ( "," , ":" ) , sort_keys = True ) )