Python script to invoke an AWS API Gateway endpoint that requires IAM authentication. Uses boto for signing the request, and requests for issuing it.

apigateway-invoke.py

#!/usr/bin/env python3
"""
Provides a simple Python wrapper for invoking an API Gateway endpoint using IAM signed requests.

Example:
  python3 apigateway-invoke.py GET \
    https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/default/MethodName | jq .
"""

try:
    from urllib.parse import urlparse, urlencode, parse_qs
except ImportError:
    from urlparse import urlparse, parse_qs
    from urllib import urlencode

import re
import sys

import requests
from boto3 import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest


def signing_headers(method, url_string, body):
    # Adapted from:
    #   https://github.com/jmenga/requests-aws-sign/blob/master/requests_aws_sign/requests_aws_sign.py
    region = re.search("execute-api.(.*).amazonaws.com", url_string).group(1)
    url = urlparse(url_string)
    path = url.path or '/'
    querystring = ''
    if url.query:
        querystring = '?' + urlencode(
            parse_qs(url.query, keep_blank_values=True), doseq=True)

    safe_url = url.scheme + '://' + url.netloc.split(
        ':')[0] + path + querystring
    request = AWSRequest(method=method.upper(), url=safe_url, data=body)
    SigV4Auth(Session().get_credentials(), "execute-api",
              region).add_auth(request)
    return dict(request.headers.items())


if __name__ == "__main__":
    method = sys.argv[1]
    url = sys.argv[2]

    if not sys.stdin.isatty():
        body = sys.stdin.read()
    else:
        body = None

    r = requests.get(url, headers=signing_headers(method, url, body))
    print(r.content.decode("utf-8"))

This worked great for GET requests, but I am not able to get it to run with POST request. I am trying to run an sql query in the the body... any tips?

Depends on the error you're getting. I'd start by making sure that the body you're sending exactly matches what was signed.

Specifically, sys.stdin.read() returns a string, which is (eventually) utf-8 encoded, but if you're passing in a dict other things happen.

Since requests.post() expects that data be, among other things, a bytes-like-object, you'll need to ensure that you are transforming your string into bytes exactly like the botocore AWS signing functions do.

@Riebart - thanks for the response! I needed to transform the body into a JSON object and it worked.

I tried to use json.loads('json:str') with requests.post(url, signed_headers, json=jsonData), it does not work, finally it turned out that if your APIGateWay's auth method is None, you don't need to pass the signed_headers, just do post as:

r = requests.post(url, json)