Making a signed IAM_AUTH request for AWS API Gateway w/ Python

Question

At a high level, I'm using serverless to create an AWS ApiGateway protected by IAM_AUTH in one AWS account, and trying to call the ApiGateway from another AWS account.

My question is, how do I sign an HTTP request so that an IAM_AUTH protected ApiGateway trusts it?

Here's my setup,

Account B has an IAM Role, CrossAccountRole. A Lambda function, in Account B using the CrossAccountRole, assumes the IAM CallApiRole in Account A. The Lambda function uses the assumed CallApiRole role's access_key_id and access_secret to sign an http request and execute the HTTP GET request against the ApiGateway. When I execute the HTTP GET request, I get the following error,

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

My current Account B Lambda function implements 2 different signing methods, but both do not work.

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
    }

Similar to this question, but the top voted answer did not work.

Answer 1

Using boto3 and urllib makes things a bit cleaner:

import urllib.parse
import requests
from aws_requests_auth.aws_auth import AWSRequestsAuth
import boto3

credentials = boto3.Session().get_credentials()

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

auth = AWSRequestsAuth(
    aws_access_key=credentials.access_key,
    aws_secret_access_key=credentials.secret_key,
    aws_token=credentials.token,
    aws_host=urllib.parse.urlparse(endpoint).netloc,
    aws_region="us-east-1",
    aws_service="apigateway",
)

r = requests.get(
    url=endpoint,
    auth=auth,
)
print(r.text)

Answer 2

Copy and pasted from the wrong README example.

This worked for me,

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()
    }