Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save ottokruse/80a082dbd41867ee79660b28076ec7dc to your computer and use it in GitHub Desktop.
Save ottokruse/80a082dbd41867ee79660b28076ec7dc to your computer and use it in GitHub Desktop.
Invoke an AWS API Gateway Endpoint using Cognito Client Credentials
"""
This example does a HTTP POST, change it to whatever you want.
The code here assumes that:
- Your API Gateway has been setup with a Cognito User Pool authorizer
- That User Pool authorizer is set up with http header "authorization" as Token Source
- The API Gateway Method you are invoking is set up to require at least one OAuth scope
"""
from urllib.request import Request, urlopen
from urllib.parse import quote_plus
import argparse
from base64 import urlsafe_b64encode
import json
def acquire_access_token_from_cognito(
*, client_id: str, client_secret: str, cognito_domain_name: str, scope: str
):
credentials = urlsafe_b64encode(f"{client_id}:{client_secret}".encode()).decode()
cognito_token_endpoint = f"https://{cognito_domain_name}/oauth2/token"
with urlopen(
Request(
cognito_token_endpoint,
data=f"grant_type=client_credentials&scope={quote_plus(scope)}".encode(),
headers={
"content-type": "application/x-www-form-urlencoded",
"authorization": f"Basic {credentials}",
},
method="POST",
)
) as response:
return json.loads(response.read())["access_token"]
def invoke_api_with_cognito_authentication(
*, api_url: str, access_token: str, data: bytes = None
):
with urlopen(
Request(
api_url,
data=data,
headers={
"content-type": "application/json",
"authorization": access_token,
},
method="POST",
)
) as response:
return response.status, response.read().decode()
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Call an API that is protected with Cognito using OAuth Client Credentials"
)
parser.add_argument("--client-id", type=str, help="Your client id", required=True)
parser.add_argument(
"--client-secret", type=str, help="Your client secret", required=True
)
parser.add_argument("--api-url", type=str, help="The URL to call", required=True)
parser.add_argument(
"--scope",
type=str,
help="The OAuth scope to request an access token for",
required=True,
)
parser.add_argument(
"--cognito-domain-name",
type=str,
help="The Cognito Auth Domain to authenticate against",
required=True,
)
parser.add_argument("--data", type=str, help="The JSON data to POST to the API")
args = parser.parse_args()
print("Acquiring access token from Cognito using client credentials ...")
access_token = acquire_access_token_from_cognito(
client_id=args.client_id,
client_secret=args.client_secret,
cognito_domain_name=args.cognito_domain_name,
scope=args.scope,
)
print("Successfully acquired access token!")
print("Invoking API ...")
response_status, response_body = invoke_api_with_cognito_authentication(
api_url=args.api_url,
access_token=access_token,
data=args.data.encode() if args.data else None,
)
print(f"Response status: {response_status}")
print("Response body:")
print(response_body)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment