django-boto3-cognito: AWS' Cognito Developer Authenticated Identities Authflow using Django/Python/Boto3

cognito-developer-authenticated-client-example.py

__author__ = 'dkarchmer'
'''
Main fucntionality to manage API calls to ampervue.com
'''
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
        # (e.g. one that only gives access to upload files to one bucket in S3)
        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']

        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

import logging
import boto3
import botocore
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'
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
    """

    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'),

)