使用 Python 为 AWS API Gateway 发出签名的 IAM_AUTH 请求

Question

概括地说，我使用无服务器在一个 AWS 账户中创建受 IAM_AUTH 保护的 AWS ApiGateway，并尝试从另一个 AWS 账户调用 ApiGateway。

我的问题是，如何签署 HTTP 请求，以便 IAM_AUTH 受保护的 ApiGateway 信任它吗？

这是我的设置，

账户 B 有一个 IAM 角色 CrossAccountRole。账户 B 中使用 CrossAccountRole 的 Lambda 函数假定账户 A 中的 IAM CallApiRole。Lambda 函数使用假定的 CallApiRole 角色的 access_key_id 和 access_secret 签署 http 请求并针对 ApiGateway 执行 HTTP GET 请求。当我执行 HTTP GET 请求时，我收到以下错误，

{"message":"The security token included in the request is invalid."}

我当前的账户 B Lambda 函数实现了 2 种不同的签名方法，但都不起作用。

import json
import sys, os, base64, datetime, hashlib, hmac, urllib
import boto3
import requests
from aws_requests_auth.aws_auth import AWSRequestsAuth


def lambda_handler(event, context):

    #
    # Option 1. use aws_requests_auth to sign request
    #
    access_key = os.getenv("AWS_ACCESS_KEY_ID")
    secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")

    auth = AWSRequestsAuth(aws_access_key=access_key,
                          aws_secret_access_key=secret_key,
                          aws_host='aaaaaaaaaa.execute-api.us-east-1.amazonaws.com',
                          aws_region='us-east-1',
                          aws_service='execute-api')

    endpoint = 'https://aaaaaaaaaa.execute-api.us-east-1.amazonaws.com/dev/hello'

    print("{}".format(endpoint))
    print(auth)

    r = requests.get(endpoint, auth=auth)



    #
    # Option 2. Manually sign the request
    #
    # 1) Assume arn:aws:iam::AAAAAAAAAAAA:role/CallApiRole
    # 2) Sign request
    # 3) HTTP get the ApiGateway
    #
    # https://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html#sig-v4-examples-get-auth-header
    #
    method = 'GET'
    service = 'execute-api'
    host = 'aaaaaaaaaa.execute-api.us-east-1.amazonaws.com'
    region = 'us-east-1'
    endpoint = 'https://aaaaaaaaaa.execute-api.us-east-1.amazonaws.com/dev/hello'
    request_parameters = ''

    def sign(key, msg):
        return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()

    def getSignatureKey(key, dateStamp, regionName, serviceName):
        kDate = sign(('AWS4' + key).encode('utf-8'), dateStamp)
        kRegion = sign(kDate, regionName)
        kService = sign(kRegion, serviceName)
        kSigning = sign(kService, 'aws4_request')
        return kSigning

    sts_client = boto3.client('sts')

    assumed_role = sts_client.assume_role(
        RoleArn="arn:aws:iam::AAAAAAAAAAAA:role/CallApiRole",
        RoleSessionName="HiMomImComputering")

    credentials = assumed_role["Credentials"]
    access_key = credentials["AccessKeyId"]
    secret_key = credentials["SecretAccessKey"]

    access_key = os.getenv("AWS_ACCESS_KEY_ID")
    secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")


    # Create a date for headers and the credential string
    t = datetime.datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ') # Format date as YYYYMMDD'T'HHMMSS'Z'
    datestamp = t.strftime('%Y%m%d') # Date w/o time, used in credential scope


    # ************* TASK 1: CREATE A CANONICAL REQUEST *************
    # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html

    # Because almost all information is being passed in the query string,
    # the order of these steps is slightly different than examples that
    # use an authorization header.

    # Step 1: Define the verb (GET, POST, etc.)--already done.

    # Step 2: Create canonical URI--the part of the URI from domain to query
    # string (use '/' if no path)
    canonical_uri = '/'

    # Step 3: Create the canonical headers and signed headers. Header names
    # must be trimmed and lowercase, and sorted in code point order from
    # low to high. Note trailing \n in canonical_headers.
    # signed_headers is the list of headers that are being included
    # as part of the signing process. For requests that use query strings,
    # only "host" is included in the signed headers.
    canonical_headers = 'host:' + host + '\n'
    signed_headers = 'host'

    # Match the algorithm to the hashing algorithm you use, either SHA-1 or
    # SHA-256 (recommended)
    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'

    # Step 4: Create the canonical query string. In this example, request
    # parameters are in the query string. Query string values must
    # be URL-encoded (space=%20). The parameters must be sorted by name.
    # use urllib.parse.quote_plus() if using Python 3
    # canonical_querystring = 'Action=CreateUser&UserName=NewUser&Version=2010-05-08'
    canonical_querystring = ''
    canonical_querystring += 'X-Amz-Algorithm=AWS4-HMAC-SHA256'
    canonical_querystring += '&X-Amz-Credential=' + urllib.parse.quote_plus(access_key + '/' + credential_scope)
    canonical_querystring += '&X-Amz-Date=' + amz_date
    canonical_querystring += '&X-Amz-Expires=30'
    canonical_querystring += '&X-Amz-SignedHeaders=' + signed_headers

    # Step 5: Create payload hash. For GET requests, the payload is an
    # empty string ("").
    payload_hash = hashlib.sha256(('').encode('utf-8')).hexdigest()

    # Step 6: Combine elements to create canonical request
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash


    # ************* TASK 2: CREATE THE STRING TO SIGN*************
    string_to_sign = algorithm + '\n' +  amz_date + '\n' +  credential_scope + '\n' +  hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()

    # ************* TASK 3: CALCULATE THE SIGNATURE *************
    # Create the signing key
    signing_key = getSignatureKey(secret_key, datestamp, region, service)

    # Sign the string_to_sign using the signing_key
    signature = hmac.new(signing_key, (string_to_sign).encode("utf-8"), hashlib.sha256).hexdigest()


    # ************* TASK 4: ADD SIGNING INFORMATION TO THE REQUEST *************
    # The auth information can be either in a query string
    # value or in a header named Authorization. This code shows how to put
    # everything into a query string.
    canonical_querystring += '&X-Amz-Signature=' + signature


    # ************* SEND THE REQUEST *************
    # The 'host' header is added automatically by the Python 'request' lib. But it
    # must exist as a header in the request.
    request_url = endpoint + "?" + canonical_querystring

    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    r = requests.get(request_url)

    print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
    print('Response code: %d\n' % r.status_code)
    print(r.text)

    return {
        'statusCode': 200,
        'body': r.text
    }

与question 类似，但top voted answer 不起作用。

Answer 1

从错误的 README 示例中复制并粘贴。

这对我有用，

import os
from aws_requests_auth.aws_auth import AWSRequestsAuth

def lambda_handler(event, context):
    auth = AWSRequestsAuth(aws_access_key=os.environ['AWS_ACCESS_KEY_ID'],
                           aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY'],
                           aws_token=os.environ['AWS_SESSION_TOKEN'],
                           aws_host='abc123.execute-api.us-east-1.amazonaws.com',
                           aws_region='us-east-1',
                           aws_service='apipgateway')

    endpoint = 'https://abc123.execute-api.us-east-1.amazonaws.com/dev/hello'

    r = requests.get(endpoint, auth=auth)

    return {
        'statusCode': 200,
        'body': r.json()
    }