Last active
May 30, 2023 12:21
-
-
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.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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")) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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 isNone
, you don't need to pass thesigned_headers
, just do post as: