Invoke an AWS API Gateway Endpoint using Cognito Client Credentials

invoke-aws-api-gateway-with-cognito-client-credentials.py

"""
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)