Flask API & Cognito

Our API uses API Gateway and Lambda functions with Python and a library called Flask. It’s a great match for our uses and it helps us speed up the development process for our API business logic.

We will use Cognito to protect our API endpoints; and thankfully, there are many resources that guide us in this endeavor.

Requirements

The python requirements are organized in three environments:

requirements/
├── dev.txt
├── production.txt
└── staging.txt

** TO DEVELOP LOCALLY YOU MUST CHOOSE DEV **

Install the requirements in your machine, run these in order:

# 1) First create a virutal environment in the API root folder:
$ virtualenv venv

# 2) Then activate the environment
$ source venv/bin/activate

# 3) Now you are ready to install the requirements
$ pip install -r requirements/dev.txt

This particular requirements file includes tools such as pytest that make development and unit testing a lot easier, but it also makes the api bulky. Do not bother in installing the production or staging requirement files, those are only meant for cloud deployments.

Running the API (with hot-reload)

Once the installation of the requirements is done, you are ready to launch the application using this command:

# Run Flask in development mode:
$ FLASK_ENV=development flask run

You may have noticed the FLASK_ENV=development bash variable, this is passed to the flask command and it will initialize the application in app.py and enabled hot-reload, meaning that any changes you make to the code will be automatically reloaded for you (without you having to restart the API for every change).

Blueprint Architecture

We will adhere to a blueprint architecture as it is stipulated in their documentation: https://flask.palletsprojects.com/en/1.1.x/blueprints/#blueprints

This is going to help scale large amounts of code into our API, it should also help with modularity and code re-use and our testing strategies. Please refer to the architecture notes below for details on how blueprints work.

Test-driven development

To enable test-driven development patterns in our API I have created a tests folder with a sample test.

In the API root directory, you can use these commands to run your tests:

# Run all tests
$ pytest -vs

# Run a specific test file
$ pytest -vs tests/your_test.py

# Run a specific test in a file:
$ pytest -v tests/your_tests.py::TestClass::test_method

Creating a new test file

You should look at a file called ./tests/test_app.py and copy it into a new file. Inside the test_app.py file you will see this syntax:

First you need to make sure you import the Flask application:

#!/usr/bin/env python
import json, pdb
from unittest.mock import patch

# Imports the Flask application
from app import app

Then, you create a test class, it must begin with the prefix Test in order to be valid, here we use TestApp but if you were to create a new test file for say the authentication blueprint, you could name the test class TestAuth so on and so forth:

class TestApp:
    @classmethod
    def setup_class(cls):
        # Gives us access to the app class
        cls.app = app
        cls.app.config["TESTING"] = True
        # Allows us to have a client for every test we make via self
        cls.client = cls.app.test_client()
        print("Beginning tests for: TestApp")

    @classmethod
    def teardown_class(cls):
        # Discards the app instance we have
        cls.app = None
        cls.client = None
        print("\n\nAll tests finished for: TestApp")

And your first test could be something like this:

    @staticmethod
    def parse_response(response: bytes) -> dict:
        """
        Parses a response from Flask into a JSON dict
        :param bytes response: The response bytes string
        :return dict:
        """
        return json.loads(response.decode('utf-8'))

    def test_app_initializes(self):
        """Start with a blank database."""
        response = self.client.get('/')
        response_dict = self.parse_response(response.data)

        assert isinstance(response_dict, dict)
        assert "message" in response_dict
        assert "MOPED API Available" in response_dict.get("message", "")

Parsing the JWT token within the API

Parsing JWT tokens provided by AWS Cognito is done with the help of the flask-cognito library (see references at the bottom) Take a look at the ./auth/auth.py file in the API, you will notice a few interesting lines:

First we import two helper methods:

from flask_cognito import cognito_auth_required, current_cognito_jwt

cognito_auth_required: This is a decorator that populates the value of a global thread variable (local) with the decoded JWT token.
current_cognito_jwt: This is a helper function, basically it can safely access the place in memory where the JWT token is stored. This is a lambda function that returns a class of type LocalProxy, which wraps the dictionary value we are looking for. To access the decoded token value as a dictionary, use the _get_current_objec() method.

Example:

#
# In order to retrieve the current_cognito_jwt object,
# we need to call the @cognito_auth_required decorator.
#
@auth_blueprint.route('/example')
@cognito_auth_required
def auth_example() -> str:
    """
    Shows the current user payload data
    :return str:
    """
    
    # Use the _get_current_object method to get the dictionary containing our token:
    decoded_jwt = current_cognito_jwt._get_current_object()
    
    # Now we output our token:
    return jsonify({
        "decoded_jwt": decoded_jwt
    })

Custom Decorators

One thing I have discovered with Hasura and Cognito, is that the Hasura claims are not a normal JSON document, in fact, the JWT token wraps the Hasura claims in a nested JSON (JSON within a JSON), which can be inconvenient, but it is the way it works according to their documentation.

This is an example of a decoded JWT token, notice the Hasura claims in the https://hasura.io/jwt/claims key:

{
    "aud": "3u9n9373e37v603tbp25gs5fdc",
    "auth_time": 1601060440,
    "cognito:username": "96051f62-7897-4264-ad97-34be981bfcc9",
    "email": "sergio.garcia@austintexas.gov",
    "email_verified": true,
    "event_id": "ee9d7f1b-c2d9-425b-ae2d-9f1547142bb8",
    "exp": 1601064040,
    "https://hasura.io/jwt/claims": "{\"x-hasura-default-role\": \"user\", \"x-hasura-allowed-roles\": [\"user\"], \"x-hasura-user-id\": \"96051f62-7897-4264-ad97-34be981bfcc9\"}",
    "iat": 1601060440,
    "iss": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_U2dzkxfTv",
    "sub": "96051f62-7897-4264-ad97-34be981bfcc9",
    "token_use": "id"
},

To help with this, I’ve created a couple decorators and methods in the ./claims.py file, which can help with obtaining a normalized version of the token. One of them is called @normalize_claims , here is an example on how to use it:

#
# You may also use the normalize_claims decorator
# along with the claims parameter to have a fully parsed claims dict
#
@auth_blueprint.route('/example')
@cognito_auth_required
@normalize_claims
def auth_example(claims) -> str:
    """
    Shows the current user payload data
    :return str:
    """
    return jsonify({
        "cognito:username": claims["cognito:username"],
        "email": claims["email"],
        "hasura_claims": claims["https://hasura.io/jwt/claims"]
    })

You will notice there will not be any nested JSON strings. Feel free to implement your own decorators or helper methods.

Diagram owner

Data & Technology Services (Sergio Garcia)

Team

Sergio Garcia
@ Team member

Informed

@ Stakeholder
@ Stakeholder

Status

/ / /

Last date updated

e.g.,24 Sep 2020

On this page

Name

Description

Operational Excellence

The ability to run and monitor systems to deliver business value and to continually improve supporting processes and procedures.

Security

The ability to protect information, systems, and assets while delivering business value through risk assessments and mitigation strategies.

Reliability

The ability of a system to recover from infrastructure or service disruptions, dynamically acquire computing resources to meet demand, and mitigate disruptions such as misconfigurations or transient network issues.

Performance Efficiency

The ability to use computing resources efficiently to meet system requirements, and to maintain that efficiency as demand changes and technologies evolve

Cost Optimization

The ability to run systems to deliver business value at the lowest price point.

AWS Well Architected Framework PDF

note

Goals

Secure the API to Cognito users only.
Modular and scalable development of the API with Flask Blueprints
Test-driven development
Access to JWT tokens.
Secure the API to Cognito users only.
Modular and scalable development of the API with Flask Blueprints
Test-driven development
Access to JWT tokens.

Architecture

In terms of infrastructure, this is the way the architecture looks like. The API Gateway, CloudFront distribution, and Lambda functions are deployed by Zappa, along with additional IAM policies. The Cognito user pool has already been deployed separately. The GitHub actions will deploy API for staging and production branches only.

It must be noted that currently PRs are not supported, this is because a branch would need to be deployed, and each branch will require specific configurations and resources in AWS to work properly. It may be possible to group together all the resources zappa creates, and then produce a single configuration for all branches and make endpoints using common permissions and configurations. Again, at the moment PRs are not yet supported.

Architecture & Blueprints

To implement the API we will strictly follow the blueprint guidelines in the Flask documentation: https://flask.palletsprojects.com/en/1.1.x/blueprints/#blueprints

These are the main takeaways about our architecture:

app.py, this is the entrypoint for all API calls, and in a way, it is the main router. It’s job is to instantiate the main application, and it is here where we register all the other sections (blueprints) we will implement in the future.
config.py contains any configuration needed by app.py
claims.py currently contains helpers that can help decoding JWT tokens, and its purpose it to implement decorators that can be re-used throughout the project.
tests contains all the tests for this api following pytest patterns.
requriements contains the python libraries needed to run locally (dev) or remotely (staging, production).
static contains any static files served by app.py
templates contains any jinja templates used by app.py
<blueprints> each blueprint will be hosted in a folder

Take a look at the root directory of the API, it should look a little bit like this:

.
├── README.md
├── app.py
├── auth
│   ├── __init__.py
│   ├── auth.py
│   └── templates
│       └── auth
├── claims.py
├── config.py
├── requirements
│   ├── dev.txt
│   ├── production.txt
│   └── staging.txt
├── static
├── templates
├── tests
└── zappa_settings.json

Blueprints

When creating a new blueprint, you will need to create a new folder. Take a look at auth for example, at the time of writing this, it looks a bit like this:

├── auth
│   ├── __init__.py
│   ├── auth.py
│   └── templates

The pattern should be like this:

<blueprint name as folder>/<blueprint name as python file> => auth/auth.py

Every blueprint directory should contain a templates directory that contains any jinja templates that are used specifically by that blueprint. Any other code related to the blueprint should be contained in the same directory.

Deployment strategy

The deployments are taken care of using GitHub actions and the Zappa framework. Currently we will only support deploying staging and production, but not PRs. PRs will take a bit more work and require more research.

Rest API endpoints

The REST API routes and functionality will be described later on in the documentation; for now, these are the endpoint URL addresses.

Staging: https://moped-api-staging.austinmobility.io

Production: https://moped-api.austinmobility.io

Making a test request:

The first thing you will need to make a request is to get a valid JWT token, if you are in Staging you will need a staging token, if you are going to ping the production endpoint you will need a production JWT.

To get a token, log in to the site (staging):

https://moped.austinmobility.io/moped/session/signin

Once you are logged in, get the jwt_token variable from local storage. In the console you can type localStorage.getItem("jwt_token"), the result will output your token:

Copy that token and paste it in curl like this:

curl --location \
   --request GET 'https://moped-api-staging.austinmobility.io/auth/current_user' \
   --header 'Authorization: Bearer <your_token_here>'

You can see the result is something like this:

Alternatively you can also use Postman:

Action Items

Action

Description

Owner

Due date

ZenHub ticket

Implement support for PR for APIs

Eventually, we will need PRs for staging, we need to architect this eventually.

Sergio Garcia

e.g.,24 Sep 2020

Add your Jira ticket by typing / Jira.

References and documentation

Zappa Framework https://github.com/Miserlou/Zappa
Flask:
https://flask.palletsprojects.com/en/1.1.x/
Flask Blueprints: https://flask.palletsprojects.com/en/1.1.x/blueprints/#blueprints https://realpython.com/flask-blueprint/
Flask Cognito: https://github.com/jetbridge/flask_cognito

PreviousReact & Cognito NextSingle Sign-On with CTM

Last updated 4 years ago