django-boto3-cognito: AWS' Cognito Developer Authenticated Identities Authflow using Django/Python/Boto3 (For building stand-alone clients)

cognito-developer-authenticated-client-example.py

__author__ = 'dkarchmer'
'''
This script emulates a stand-alone Python based client. It relies on Boto3 to access AWS, but
requires your Django server to have an API for your user to access Cognito based credentials

Because of Cognito, the client (this script) will only get temporary AWS credentials associated
to your user and only your user, and based on whatever you configure your AIM Policy to be.

Most Cognito examples demonstrate how to use Cognito for Mobile Apps, so this scripts demonstrate
how to create a stand-alone Python script but operating similarly to these apps.
'''
import json
import requests
import logging
import boto3
from boto3.session import Session

AWS_REGION = 'us-east-1'

AV_DOMAIN_NAME = 'https://example.com'
AV_API_PREFIX = 'api/v1'
logger = logging.getLogger(__name__)

class Connection(object):

    token = None
    domain = AV_DOMAIN_NAME

    def __init__(self, domain=None):
        if domain:
            self.domain = domain

    def _get_url(self, api):
        url = '{0}/{1}/{2}'.format(self.domain, AV_API_PREFIX, api)
        logger.debug('Calling: ' + url)
        return url

    def _get_header(self, use_token):
        if use_token:
            if not self.token:
                raise('No Token')
            authorization_str = 'token %s' % self.token
            headers = {'content-type': 'application/json',
                       'Authorization': authorization_str}
        else:
            headers = {'Content-Type': 'application/json'}
        return headers

    def get(self, api, use_token):

        url = self._get_url(api)
        headers = self._get_header(use_token)

        r = requests.get(url, headers=headers)

        return r

    def login(self, password, email):
        data = { 'email': email, 'password': password }
        api = 'auth/login'

        r = self.post(api=api, data=data, use_token=False)
        if r.status_code == 200:
            content = json.loads(r.content.decode())
            self.token = content['token']
            self.username = content['username']
            logger.info('Welcome @{0} (token: {1})'.format(self.username, self.token))
            return True
        else:
            logger.error('Login failed: ' + str(r.status_code) + ' ' + r.content.decode())
            return False

    def logout(self):
        api = 'auth/logout'
        r = self.post(api=api, use_token=True)
        if r.status_code == 204:
            logger.info('Goodbye @{0}'.format(self.username))
            self.username = None
            self.token = None
        else:
            logger.error('Logout failed: ' + str(r.status_code) + ' ' + r.content.decode())

if __name__ == '__main__':
    # Test
    # Logger Format
    from logging import StreamHandler, Formatter
    FORMAT = '[%(asctime)-15s] %(levelname)-6s %(message)s'
    DATE_FORMAT = '%d/%b/%Y %H:%M:%S'
    formatter = Formatter(fmt=FORMAT, datefmt=DATE_FORMAT)
    handler = StreamHandler()
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)

    c = Connection('http://127.0.0.1:8000')
    c.login(email='user1@test.com', password='user1')
    r = c.get(api='auth/aws', use_token=True)
    logger.info(r.status_code)
    if r.status_code == 200:
        content = json.loads(r.content.decode())

        # Given the resp from the server, with an IdentityId and a Token, the client can directly
        # get credentials from Cognito, which should be based on a given IAM Role
        # (One that only allows access to wahtever services you want the client script to access. e.g. S3 uploads)
        client = boto3.client('cognito-identity', AWS_REGION)
        resp = client.get_credentials_for_identity(IdentityId=content['IdentityId'],
                                                   Logins={'cognito-identity.amazonaws.com': content['Token']})

        # The resp contains the actual temporary AWS secret/access codes and a session token, to be
        # used with the rest of the AWS APIs
        secretKey = resp['Credentials']['SecretKey']
        accessKey = resp['Credentials']['AccessKeyId']
        sessionToken = resp['Credentials']['SessionToken']

        # Now you can use Boto3 like you would if you were using your own secret keys
        # what you will see in any Boto3 example on the web
        session = Session(aws_access_key_id=accessKey,
                          aws_secret_access_key=secretKey,
                          aws_session_token=sessionToken,
                          region_name=AWS_REGION)

        s3_resource = session.resource('s3')
        bucket = s3_resource.Bucket('bucket-with-role-permision')
        print('\n' + str(bucket))
        for key in bucket.objects.all():
            logger.info(key.key)

    logger.info("-------------")

    c.logout()

cognito-django-based-server.py

__author__ = 'dkarchmer'
'''
This file demonstrates how to create a Django Rest Framework (DRF) APIView based API to provide
user access to AWS credentials via Cognito (In the script above, this is defined as http://127.0.0.1:8000/auth/aws)

NOTE: This is NOT intended to be fully working code. It may not, for example, contain all required imports
      It also does not explain how to hook this with Django and DRF. It assumes you know how to do so
      But the Class and associated functions are in fact fully functional. 
'''
import logging
import boto3
import botocore

from rest_framework.response import Response
from rest_framework.views import APIView

AWS_REGION = 'us-east-1

# Get an instance of a logger
logger = logging.getLogger(__name__)

# Assumes a cognito identity pool has been created.
IDENTITY_POOL_ID = 'us-east-1:XXXXXXXXXX-1234-1234-1234-YYYYYYYYY'
# Change based on what you entered on the AWS Cognito Dashboard (Custom tab)
DEVELOPER_PROVIDED_NAME = 'account.example.com'
TOKEN_DURATION = 3400

def get_aws_open_id_token(username):
    client = boto3.client('cognito-identity', AWS_REGION)

    # This is what should be done by the Server (after proper login).
    # It requires the Role to have access to cognito
    # Given a valid user (per server's authentication), the API should return the Cognito Resp
    # directly to the client
    # See http://docs.aws.amazon.com/cognito/devguide/identity/concepts/authentication-flow/
    logger.info('Requesting open_id from Cognito for: {0}'.format(username))
    try:
        resp = client.get_open_id_token_for_developer_identity(
            IdentityPoolId=IDENTITY_POOL_ID,
            Logins={DEVELOPER_PROVIDED_NAME: username},
            TokenDuration=TOKEN_DURATION
        )
        logger.info("Identity ID: {0}".format(resp['IdentityId']))
        logger.info("Request ID : {0}".format((resp['ResponseMetadata']['RequestId'])))
    except botocore.exceptions.ClientError as e:
        logger.error(str(e))
        resp = None


    return resp
    

class APICognitoViewSet(APIView):
    """
    GET IdentityId/Token that can be used by client to get Temporary AWS credentials
    This is the class that services the /auth/aws GET API in the script above
    """

    def get(self, request, format=None):
        """
        Update IdentityID and Token from Cognito
        """
        if request.user.is_anonymous():
            # User most login before they can get a token
            # This not only ensures the user has registered, and has an account
            # but that the account is active
            return Response('User not recognized.', status=status.HTTP_403_FORBIDDEN)

        data_dic = {}

        resp = get_aws_open_id_token(request.user.username)
        if resp:
            mystatus = resp['ResponseMetadata']['HTTPStatusCode']
            data_dic['IdentityId'] = resp['IdentityId']
            data_dic['Token'] = resp['Token']
        else:
            logger.error('Something wrong with Cognito call')
            mystatus=status.HTTP_500_INTERNAL_SERVER_ERROR

        return Response(data_dic, status=mystatus)


urlpatterns = patterns('',
    ...
    url(r'^auth/aws', APICognitoViewSet.as_view(), name='api-cognito'),

)

Has the client code every worked? In connection() I don't see self.post() defined. What does auth/login do?

@eprparadocs I think the self.post in this case is sending your user credentials to your back-end logic. If the request is successful, then the response return is 200.

This was extremely helpful for me to understand how Cognito works in the client/server, in a typical DRF setup. Thank you!

I was able to adapt the client code to work for me. In my case, my API uses basic auth (and returns a token in the response) so I replaced the self.post() call with a self.do_login() call to this function:

    def do_login(self, api, username, password):

        url = self._get_url(api)
        headers = self._get_header(use_token=False)

        r = requests.post(url, auth=(username, password))

        return r

what is the boto3 function used for logging out the user?