Skip to content

Instantly share code, notes, and snippets.

@Riebart
Last active May 30, 2023 12:21
Show Gist options
  • Star 3 You must be signed in to star a gist
  • Fork 3 You must be signed in to fork a gist
  • Save Riebart/dd68c852ca7fbaccaf3bdac4919f351f to your computer and use it in GitHub Desktop.
Save Riebart/dd68c852ca7fbaccaf3bdac4919f351f to your computer and use it in GitHub Desktop.
Python script to invoke an AWS API Gateway endpoint that requires IAM authentication. Uses boto for signing the request, and requests for issuing it.
#!/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"))
@lengerfulluse
Copy link

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)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment