How to use Cognito for AppSync mutation call (Python)

Question

I'd like to call mutations from AppSync using my Python function but use a Cognito user for the authorization as "API-KEY", "IAM" and other methods are not suitable for my application.

My mutation looks like this (test purposes):

mutation XYZ {
    updateTask(input: {id: "a1b2c3", name: "newTaskName"}) {
        id
        name
    }
}

Answer 1

I am assuming that the user is already created and enabled by some means. If your AppSync API is secured only using Cognito, you are always going to need a username and a password to begin with. For example, you can use below code to login and get the AccessToken from the response:

import boto3
def get_user_auth(event, context):
    client = boto3.client('cognito-idp')
    response = client.initiate_auth(
        UserPoolId='xxxxxxxxx',
        ClientId='xxxxxxxxxxxxxx',
        AuthFlow='USER_PASSWORD_AUTH',
        AuthParameters={
            'USERNAME': 'xxxxxx',
            'PASSWORD': 'xxxxxx'
        }
    )
    return response

Note: Make sure that you have "Enable username password based authentication (ALLOW_USER_PASSWORD_AUTH)" enabled.

Once you have the access token, you can use this in HTTP headers within your request as follows:

{
  "authorization": "<YOUR-VERY-VERY-LONG-ACCESS-TOKEN>"
} 

For example:

import requests
from requests_aws4auth import AWS4Auth
import boto3

session = requests.Session()

APPSYNC_API_ENDPOINT_URL = '<YOUR-API-URL>'

mutation = """mutation XYZ {updateTask(input: {id: "a1b2c3", name: "newTaskName"}) {id, name}}"""

response = session.request(
    url=APPSYNC_API_ENDPOINT_URL,
    method='POST',
    headers={'authorization': '<YOUR-VERY-VERY-LONG-ACCESS-TOKEN>'},
    json={'mutation': mutation}
)

print(response.json()['data'])

Since this access token has some expiration, you might also need to refresh this token by using the RefreshToken from the above response. Like so:

def refresh_token(self, username, refresh_token):
    try:
        return client.initiate_auth(
            ClientId=self.client_id,
            AuthFlow='REFRESH_TOKEN_AUTH',
            AuthParameters={
                'REFRESH_TOKEN': refresh_token,
                # 'SECRET_HASH': self.get_secret_hash(username)
                # If the User Pool has been defined with App Client secret,
                # you will have to generate secret hash as well.
            }
        )
    except botocore.exceptions.ClientError as e:
        return e.response

Example of how you can generate secret hash.