Part 2: Database Models, Migrations and JWT Setup

Project Structure

The chart below shows the folder structure that was created in Part 1. In this post, we will work on all files marked as NEW CODE. Files that contain code from Part 1 but will not be modified in this post are marked as NO CHANGES.

. (project root folder)
|- src
|   |- flask_api_tutorial
|       |- api
|       |   |- auth
|       |   |   |- __init__.py
|       |   |
|       |   |- widgets
|       |   |   |- __init__.py
|       |   |
|       |   |- __init__.py
|       |
|       |- models
|       |   |- __init__.py
|       |   |- user.py
|       |
|       |- util
|       |   |- __init__.py
|       |   |- datetime_util.py
|       |   |- result.py
|       |
|       |- __init__.py
|       |- config.py
|
|- tests
|   |- __init__.py
|   |- conftest.py
|   |- test_config.py
|   |- test_user.py
|   |- util.py
|
|- .env
|- .gitignore
|- .pre-commit-config.yaml
|- pyproject.toml
|- pytest.ini
|- README.md
|- run.py
|- setup.py
|- tox.ini


KEY:
FOLDER
NEW CODE
NO CHANGES
EMPTY FILE

Database Models and Migrations with Flask-SQLAlchemy

If you have never used SQLAlchemy (or any ORM) before, the basic concept is simple: ORMs allow you to interact with data stored in a database as high-level abstractions such as classes, instances of classes and methods rather than writing raw SQL (the ORM translates your application code into SQL commands and queries).

A common task that I rarely see covered in tutorials like this is how to manage changes that are made to a database. Changes to the database schema will usually require changing data that is already stored in the database, or migrating the existing data. We will use the Flask-Migrate extension to handle this task and explain how to setup a migration system and how to create and apply migrations.

User DB Model

In Part 1, we created an instance of the Flask-SQLAlchemy extension with the name db in the src/flask_api_tutorial/__init__.py file and initialized it in the create_app method. The db object contains functions and classes from sqlalchemy and sqlalchemy.orm.

Whenever we need to declare a new database model (i.e., create a new database table), we create a class that subclasses db.Model. Since we are creating an API that performs user authentication, our first SQLAlchemy model will be a User class that stores login credentials and metadata for registered users.

Create a new file user.py in the src/flask_api_tutorial/models folder and add the content below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

"""Class definition for User model."""
from datetime import datetime, timezone
from uuid import uuid4

from flask import current_app
from sqlalchemy.ext.hybrid import hybrid_property

from flask_api_tutorial import db, bcrypt
from flask_api_tutorial.util.datetime_util import (
    utc_now,
    get_local_utcoffset,
    make_tzaware,
    localized_dt_string,
)


class User(db.Model):
    """User model for storing logon credentials and other details."""

    __tablename__ = "site_user"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    email = db.Column(db.String(255), unique=True, nullable=False)
    password_hash = db.Column(db.String(100), nullable=False)
    registered_on = db.Column(db.DateTime, default=utc_now)
    admin = db.Column(db.Boolean, default=False)
    public_id = db.Column(db.String(36), unique=True, default=lambda: str(uuid4()))

    def __repr__(self):
        return (
            f"<User email={self.email}, public_id={self.public_id}, admin={self.admin}>"
        )

    @hybrid_property
    def registered_on_str(self):
        registered_on_utc = make_tzaware(
            self.registered_on, use_tz=timezone.utc, localize=False
        )
        return localized_dt_string(registered_on_utc, use_tz=get_local_utcoffset())

    @property
    def password(self):
        raise AttributeError("password: write-only field")

    @password.setter
    def password(self, password):
        log_rounds = current_app.config.get("BCRYPT_LOG_ROUNDS")
        hash_bytes = bcrypt.generate_password_hash(password, log_rounds)
        self.password_hash = hash_bytes.decode("utf-8")

    def check_password(self, password):
        return bcrypt.check_password_hash(self.password_hash, password)

    @classmethod
    def find_by_email(cls, email):
        return cls.query.filter_by(email=email).first()

    @classmethod
    def find_by_public_id(cls, public_id):
        return cls.query.filter_by(public_id=public_id).first()

The User class demonstrates several important concepts for creating database models in SQLAlchemy:

Did you notice that we never defined a __init__ method? That’s because SQLAlchemy adds an implicit constructor to all model classes which accepts keyword arguments for all its columns and relationships. If you decide to override the constructor for any reason, make sure to keep accepting **kwargs and call the super constructor with those **kwargs to preserve this behavior:

class User(db.Model):
    # ...
    def __init__(self, **kwargs):
        super(User, self).__init__(**kwargs)
        # do custom stuff

Creating The First Migration

Whenever we make a change to our database schema (e.g., add new table, change column name, change foreign key dependencies, etc.) we need to create a migration with the Flask-Migrate extension. Each migration is stored in the migration database (which is actually just a folder named migrations), which also keeps track of the order that the migrations occurred. This allows us to either upgrade or downgrade our database all the way back to the database’s initial state.

The Flask-Migrate extension adds a new set of commands to the Flask CLI grouped under flask db. In order to create the migration database, we must run flask db init:

(flask-api-tutorial) flask-api-tutorial $ flask db init
  Creating directory /Users/aaronluna/Projects/flask-api-tutorial/migrations ...  done
  Creating directory /Users/aaronluna/Projects/flask-api-tutorial/migrations/versions ...  done
  Generating /Users/aaronluna/Projects/flask-api-tutorial/migrations/script.py.mako ...  done
  Generating /Users/aaronluna/Projects/flask-api-tutorial/migrations/env.py ...  done
  Generating /Users/aaronluna/Projects/flask-api-tutorial/migrations/README ...  done
  Generating /Users/aaronluna/Projects/flask-api-tutorial/migrations/alembic.ini ...  done
  Please edit configuration/connection/logging settings in '/Users/aaronluna/Projects/flask-api-tutorial/migrations/alembic.ini' before proceeding.

In order for Flask-Migrate to detect the User model, we must import it in the run.py module. Open run.py in the project root folder and make the changes highlighted below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

"""Flask CLI/Application entry point."""
import os

from flask_api_tutorial import create_app, db
from flask_api_tutorial.models.user import User

app = create_app(os.getenv("FLASK_ENV", "development"))


@app.shell_context_processor
def shell():
    return {"db": db, "User": User}

Ok, after making the changes to run.py we are ready to create our first migration. To do so, we use the flask db migrate command. Also, I recommend adding a message describing the schema changes that will be made, as shown below:

(flask-api-tutorial) flask-api-tutorial $ flask db migrate --message "add User model"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'site_user'
  Generating /Users/aaronluna/Projects/flask-api-tutorial/migrations/versions/eb6c2faa0708_add_user_model.py ...  done

The flask db migrate command creates the migration script but does not apply the changes to the database. To upgrade the database and execute the migration script you must run the flask db upgrade command:

(flask-api-tutorial) flask-api-tutorial $ flask db upgrade
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> eb6c2faa0708, add User model

We can verify that the site_user table has been created using flask shell and the sqlite3 module:

(flask-api-tutorial) flask-api-tutorial $ flask shell
Python 3.7.6 (default, Jan 19 2020, 06:08:58)
[Clang 11.0.0 (clang-1100.0.33.8)] on darwin
App: app [development]
Instance: /Users/aaronluna/Projects/flask_api_tutorial/instance
>>> import sqlite3
>>> from pathlib import Path
>>> DATABASE_URL = app.config.get("SQLALCHEMY_DATABASE_URI")
>>> db = sqlite3.connect(Path(DATABASE_URL).name)
>>> cursor = db.cursor()
>>> results = cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
>>> for row in results:
...     print(row)
...
('alembic_version',)
('site_user',)
>>> exit()

The query we executed retrieves all rows from the sqlite_master table where the type column contains table. This returned two rows: alembic_version and site_user. The latter confirms that running the upgrade successfully created the table we specified in user.py.

The alembic_version table contains a single column named version_num and a single row containing the version number of the most recent migration that has been executed on this database. The version number can be traced to the migration script files in the migrations folder if you are interested in discovering the low-level details of how database migrations are created.

JSON Web Token Authentication

The user authentication process that we are implementing utilizes JSON Web Tokens (JWT), which were defined in detail in Part 1. The general workflow is given below:

Depending on the environment, the access token will expire after a set amount of time. When this happens, the client must login with their email and password to obtain a new token.

encode_access_token Function

Let’s start implementing our workflow by creating the method that will be used to generate access tokens. Since the token will be generated using attributes from a User instance, we will create the method as a member of the User class.

First, update the import statements in user.py to include datetime.timedelta and the jwt package (Line 2 and Line 5 below):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

"""Class definition for User model."""
from datetime import datetime, timedelta, timezone
from uuid import uuid4

import jwt
from flask import current_app
from sqlalchemy.ext.hybrid import hybrid_property

from flask_api_tutorial import db, bcrypt
from flask_api_tutorial.util.datetime_util import (
    utc_now,
    get_local_utcoffset,
    make_tzaware,
    localized_dt_string,
)

Then, add the encode_access_token method to user.py:

55
56
57
58
59
60
61
62
63
64

def encode_access_token(self):
    now = datetime.now(timezone.utc)
    token_age_h = current_app.config.get("TOKEN_EXPIRE_HOURS")
    token_age_m = current_app.config.get("TOKEN_EXPIRE_MINUTES")
    expire = now + timedelta(hours=token_age_h, minutes=token_age_m)
    if current_app.config["TESTING"]:
        expire = now + timedelta(seconds=5)
    payload = dict(exp=expire, iat=now, sub=self.public_id, admin=self.admin)
    key = current_app.config.get("SECRET_KEY")
    return jwt.encode(payload, key, algorithm="HS256")

Let’s breakdown how this method generates the access token:

Global Test Fixtures: conftest.py

In order to test the encode_access_token method, we will need a User object (which requires a database connection, which requires a Flask application instance, etc). In the unittest framework, the correct way to do this would involve creating a base test class with a setup method that creates the application instance and initializes the database. We would then create classes that inherit from the base test class, making the setup method available without duplicating the same code in each test class.

There’s nothing wrong with this approach, but it’s not (IMO) the best or the simplest solution. With pytest, we can eliminate all of the boilerplate code required for class inheritance – we can even eliminate classes entirely. How? With fixtures, of course!

Fixtures are functions that construct any type of object needed by a test. We can utilize them by declaring our test functions with a parameter whose name is the same as a fixture. When the name of a parameter and the name of a fixture coincide, pytest will execute the fixture function and pass the result to the test function.

First, create a file named util.py in the tests folder and add the content below:

"""Shared functions and constants for unit tests."""

EMAIL = "new_user@email.com"
PASSWORD = "test1234"

It may seem silly to create a file for just these two string values, but as we develop more test cases you will see why I chose to do so.

Next, create a new file conftest.py in the tests folder and add the content below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

"""Global pytest fixtures."""
import pytest

from flask_api_tutorial import create_app
from flask_api_tutorial import db as database
from flask_api_tutorial.models.user import User
from tests.util import EMAIL, PASSWORD


@pytest.fixture
def app():
    app = create_app("testing")
    return app


@pytest.fixture
def db(app, client, request):
    database.drop_all()
    database.create_all()
    database.session.commit()

    def fin():
        database.session.remove()

    request.addfinalizer(fin)
    return database


@pytest.fixture
def user(db):
    user = User(email=EMAIL, password=PASSWORD)
    db.session.add(user)
    db.session.commit()
    return user

conftest.py is a special filename that pytest automatically looks for and loads test fixtures from, making the fixtures available to all functions in the same folder (and sub-folders) where conftest.py is located. You do not need to add an import statement to any test function in order to use the app or db fixtures we have defined.

Also, the app fixture is a special case. The pytest-flask extension looks for the app fixture and automatically creates the client fixture which returns an instance of the Flask test client that we will use to test API calls. You can read more about pytest-flask in the official docs.

Here a few more things to note about the fixtures we defined in conftest.py:

Unit Test: test_encode_access_token

We are finally ready to write test code that verifies the encode_access_token method. Create a new file in the tests folder named test_user.py and add the following content (ensure that test_user.py and conftest.py are in the same folder):

"""Unit tests for User model class."""


def test_encode_access_token(user):
    access_token = user.encode_access_token()
    assert isinstance(access_token, bytes)

Run the tox command and verify that all tests pass:

(flask-api-tutorial) flask-api-tutorial $ tox
GLOB sdist-make: /Users/aaronluna/Projects/flask-api-tutorial/setup.py
py37 create: /Users/aaronluna/Projects/flask-api-tutorial/.tox/py37
py37 installdeps: black, flake8, pydocstyle, pytest, pytest-black, pytest-clarity, pytest-dotenv, pytest-flake8, pytest-flask
py37 inst: /Users/aaronluna/Projects/flask-api-tutorial/.tox/.tmp/package/1/flask-api-tutorial-0.1.zip
py37 installed: alembic==1.4.0,aniso8601==8.0.0,appdirs==1.4.3,attrs==19.3.0,bcrypt==3.1.7,black==19.10b0,certifi==2019.11.28,cffi==1.14.0,chardet==3.0.4,Click==7.0,entrypoints==0.3,flake8==3.7.9,Flask==1.1.1,flask-api-tutorial==0.1,Flask-Bcrypt==0.7.1,Flask-Cors==3.0.8,Flask-Migrate==2.5.2,flask-restx==0.1.1,Flask-SQLAlchemy==2.4.1,idna==2.9,importlib-metadata==1.5.0,itsdangerous==1.1.0,Jinja2==2.11.1,jsonschema==3.2.0,Mako==1.1.1,MarkupSafe==1.1.1,mccabe==0.6.1,more-itertools==8.2.0,packaging==20.1,pathspec==0.7.0,pluggy==0.13.1,py==1.8.1,pycodestyle==2.5.0,pycparser==2.19,pydocstyle==5.0.2,pyflakes==2.1.1,PyJWT==1.7.1,pyparsing==2.4.6,pyrsistent==0.15.7,pytest==5.3.5,pytest-black==0.3.8,pytest-clarity==0.3.0a0,pytest-dotenv==0.4.0,pytest-flake8==1.0.4,pytest-flask==0.15.1,python-dateutil==2.8.1,python-dotenv==0.11.0,python-editor==1.0.4,pytz==2019.3,regex==2020.2.20,requests==2.23.0,six==1.14.0,snowballstemmer==2.0.0,SQLAlchemy==1.3.13,termcolor==1.1.0,toml==0.10.0,typed-ast==1.4.1,urllib3==1.25.8,wcwidth==0.1.8,Werkzeug==0.16.1,zipp==3.0.0
py37 run-test-pre: PYTHONHASHSEED='1533942126'
py37 run-test: commands[0] | pytest
================================================= test session starts ==================================================
platform darwin -- Python 3.7.6, pytest-5.3.5, py-1.8.1, pluggy-0.13.1 -- /Users/aaronluna/Projects/flask-api-tutorial/.tox/py37/bin/python
cachedir: .tox/py37/.pytest_cache
rootdir: /Users/aaronluna/Projects/flask-api-tutorial, inifile: pytest.ini
plugins: clarity-0.3.0a0, black-0.3.8, dotenv-0.4.0, flask-0.15.1, flake8-1.0.4
collected 38 items

run.py::FLAKE8 PASSED                                                                                            [  2%]
run.py::BLACK PASSED                                                                                             [  5%]
setup.py::FLAKE8 PASSED                                                                                          [  7%]
setup.py::BLACK PASSED                                                                                           [ 10%]
src/flask_api_tutorial/__init__.py::FLAKE8 PASSED                                                                [ 13%]
src/flask_api_tutorial/__init__.py::BLACK PASSED                                                                 [ 15%]
src/flask_api_tutorial/config.py::FLAKE8 PASSED                                                                  [ 18%]
src/flask_api_tutorial/config.py::BLACK PASSED                                                                   [ 21%]
src/flask_api_tutorial/api/__init__.py::FLAKE8 PASSED                                                            [ 23%]
src/flask_api_tutorial/api/__init__.py::BLACK PASSED                                                             [ 26%]
src/flask_api_tutorial/api/auth/__init__.py::FLAKE8 PASSED                                                       [ 28%]
src/flask_api_tutorial/api/auth/__init__.py::BLACK PASSED                                                        [ 31%]
src/flask_api_tutorial/api/widgets/__init__.py::FLAKE8 PASSED                                                    [ 34%]
src/flask_api_tutorial/api/widgets/__init__.py::BLACK PASSED                                                     [ 36%]
src/flask_api_tutorial/models/__init__.py::FLAKE8 PASSED                                                         [ 39%]
src/flask_api_tutorial/models/__init__.py::BLACK PASSED                                                          [ 42%]
src/flask_api_tutorial/models/user.py::FLAKE8 PASSED                                                             [ 44%]
src/flask_api_tutorial/models/user.py::BLACK PASSED                                                              [ 47%]
src/flask_api_tutorial/util/__init__.py::FLAKE8 PASSED                                                           [ 50%]
src/flask_api_tutorial/util/__init__.py::BLACK PASSED                                                            [ 52%]
src/flask_api_tutorial/util/datetime_util.py::FLAKE8 PASSED                                                      [ 55%]
src/flask_api_tutorial/util/datetime_util.py::BLACK PASSED                                                       [ 57%]
src/flask_api_tutorial/util/result.py::FLAKE8 PASSED                                                             [ 60%]
src/flask_api_tutorial/util/result.py::BLACK PASSED                                                              [ 63%]
tests/__init__.py::FLAKE8 PASSED                                                                                 [ 65%]
tests/__init__.py::BLACK PASSED                                                                                  [ 68%]
tests/conftest.py::FLAKE8 PASSED                                                                                 [ 71%]
tests/conftest.py::BLACK PASSED                                                                                  [ 73%]
tests/test_config.py::FLAKE8 PASSED                                                                              [ 76%]
tests/test_config.py::BLACK PASSED                                                                               [ 78%]
tests/test_config.py::test_config_development PASSED                                                             [ 81%]
tests/test_config.py::test_config_testing PASSED                                                                 [ 84%]
tests/test_config.py::test_config_production PASSED                                                              [ 86%]
tests/test_user.py::FLAKE8 PASSED                                                                                [ 89%]
tests/test_user.py::BLACK PASSED                                                                                 [ 92%]
tests/test_user.py::test_encode_access_token PASSED                                                              [ 94%]
tests/util.py::FLAKE8 PASSED                                                                                     [ 97%]
tests/util.py::BLACK PASSED                                                                                      [100%]

================================================== 38 passed in 6.33s ==================================================
_______________________________________________________ summary ________________________________________________________
  py37: commands succeeded
  congratulations :)

decode_access_token Function

Let’s move on to the obvious next step in our authorization workflow: decoding tokens. We need to update the import statements in user.py to include the Result class (Line 16 below):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

"""Class definition for User model."""
from datetime import datetime, timedelta, timezone
from uuid import uuid4

import jwt
from flask import current_app
from sqlalchemy.ext.hybrid import hybrid_property

from flask_api_tutorial import db, bcrypt
from flask_api_tutorial.util.datetime_util import (
    utc_now,
    get_local_utcoffset,
    make_tzaware,
    localized_dt_string,
)
from flask_api_tutorial.util.result import Result

Next, add the decode_access_token method to the user.py file:

67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

@staticmethod
def decode_access_token(access_token):
    if isinstance(access_token, bytes):
        access_token = access_token.decode("ascii")
    if access_token.startswith("Bearer "):
        split = access_token.split("Bearer")
        access_token = split[1].strip()
    try:
        key = current_app.config.get("SECRET_KEY")
        payload = jwt.decode(access_token, key, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        error = "Access token expired. Please log in again."
        return Result.Fail(error)
    except jwt.InvalidTokenError:
        error = "Invalid token. Please log in again."
        return Result.Fail(error)

    user_dict = dict(
        public_id=payload["sub"],
        admin=payload["admin"],
        token=access_token,
        expires_at=payload["exp"],
    )
    return Result.Ok(user_dict)

There are several important things to note about this method:

Unit Tests: Decode Access Token

We want to verify the three possible results of the decode_access_token method: token is valid, token is expired and token is invalid. Before we start writing any tests, place the import statements below at the top of test_user.py:

1
2
3
4
5
6

"""Unit tests for User model class."""
import json
import time
from base64 import urlsafe_b64encode, urlsafe_b64decode

from flask_api_tutorial.models.user import User

The first test will verify the expected behavior for a valid access token. Add the test_decode_access_token_success method to test_user.py:

14
15
16
17
18
19
20

def test_decode_access_token_success(user):
    access_token = user.encode_access_token()
    result = User.decode_access_token(access_token)
    assert result.success
    user_dict = result.value
    assert user.public_id == user_dict["public_id"]
    assert user.admin == user_dict["admin"]

Please note the following verifications we are performing in this method:

Next, let’s create a test that verifies what happens when we attempt to decode an expired access token. Add the test_decode_access_token_expired method to test_user.py:

23
24
25
26
27
28

def test_decode_access_token_expired(user):
    access_token = user.encode_access_token()
    time.sleep(6)
    result = User.decode_access_token(access_token)
    assert not result.success
    assert result.error == "Access token expired. Please log in again."

Let’s go through this test and explain how we achieved the desired result:

The last test is by far the most interesting of the three. It is trivially easy to modify part of a JWT and send it in place of the token that was generated by the server. For example, what if a user changed the admin claim in their token from False to True? Would they be able to access resources that require administrator access, even though they have not been granted the required access? Let’s try it out!

Add the test_decode_access_token_invalid method to test_user.py:

31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

def test_decode_access_token_invalid(user):
    access_token = user.encode_access_token()
    split = access_token.split(b".")
    payload_base64 = split[1]
    pad_len = 4 - (len(payload_base64) % 4)
    payload_base64 += pad_len * b"="
    payload_str = urlsafe_b64decode(payload_base64)
    payload = json.loads(payload_str)
    assert not payload["admin"]
    payload["admin"] = True
    payload_mod = json.dumps(payload)
    payload_mod_base64 = urlsafe_b64encode(payload_mod.encode())
    split[1] = payload_mod_base64.strip(b"=")
    access_token_mod = b".".join(split)
    assert not access_token == access_token_mod
    result = User.decode_access_token(access_token_mod)
    assert not result.success
    assert result.error == "Invalid token. Please log in again."

Rather than explain this test case line-by-line as done previously, I think it’s easier to execute the test in the flask shell interpreter and print out the value of several important variables:

(flask-api-tutorial) flask-api-tutorial $ flask shell
Python 3.7.6 (default, Jan 19 2020, 06:08:58)
[Clang 11.0.0 (clang-1100.0.33.8)] on darwin
App: flask-api-tutorial [development]
Instance: /Users/aaronluna/Projects/flask_api_tutorial
>>> import json
>>> from base64 import urlsafe_b64encode, urlsafe_b64decode
>>> from flask_api_tutorial import db
>>> from flask_api_tutorial.models.user import User
>>> USER_EMAIL = "new_user@email.com"
>>> USER_PASSWORD = "test1234"
>>> user = User(email=USER_EMAIL, password=USER_PASSWORD)
>>> db.session.add(user)
>>> db.session.commit()
>>> access_token = user.encode_access_token()
>>> split = access_token.split(b'.')
>>> print(f'access_token (original):\n  header: {split[0]}\n  payload: {split[1]}\n  signature: {split[2]}')
access_token (original):
  header: b'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9'
  payload: b'eyJleHAiOjE1NTc5NDA3MDAsImlhdCI6MTU1NzkzOTc5NSwic3ViIjoiMzhjMDMwYTAtNTdhNC00NmRjLWFjOWYtZTcwZDA0OWUzMDE2IiwiYWRtaW4iOmZhbHNlfQ'
  signature: b'EvNgTtgbgxpEmJedAOwLuxf6YSq09N2GCRmQOxF2REs'

The highlighted value above is the base64-encoded payload from the original access token that was generated by the server. Next, we will decode the payload to reveal the values for the user claims.

>>> payload_base64 = split[1]
>>> pad_len = 4 - (len(payload_base64) % 4)
>>> payload_base64 += pad_len * b'='
>>> payload_str = urlsafe_b64decode(payload_base64)
>>> payload = json.loads(payload_str)
>>> print(f'payload (original):\n {json.dumps(payload, indent=2)}')
payload (original):
 {
  "exp": 1557940700,
  "iat": 1557939795,
  "sub": "38c030a0-57a4-46dc-ac9f-e70d049e3016",
  "admin": false
}

Please note that anybody can decode the access token’s payload with the urlsafe_b64decode function as shown above. This is why you must never store sensitive user data (especially passwords) in a JWT.

The values stored in this token are common to most applications: the time when the token was created, the time when the token expires, a value used to identify the user, and a flag indicating the role/access level of the user.

Since anybody can edit the value of the payload before sending it back to the server, we want to ensure that the token is rejected if it has been modified. It is especially important to ensure that if the “admin” value is flipped to “True”, the user is not given access to protected resources.

>>> payload['admin'] = True
>>> print(f'payload (modified):\n {json.dumps(payload, indent=2)}')
payload (modified):
 {
  "exp": 1557940700,
  "iat": 1557939795,
  "sub": "38c030a0-57a4-46dc-ac9f-e70d049e3016",
  "admin": true
}

We have changed the “admin” value to True and printed the modified payload value to show that the value is now flipped.

>>> payload_mod = json.dumps(payload)
>>> payload_mod_base64 = urlsafe_b64encode(payload_mod.encode())
>>> split[1] = payload_mod_base64.strip(b'=')
>>> print(f'access_token (modified):\n  header: {split[0]}\n  payload: {split[1]}\n  signature: {split[2]}')
access_token (modified):
  header: b'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9'
  payload: b'eyJleHAiOiAxNTU3OTQwNzAwLCAiaWF0IjogMTU1NzkzOTc5NSwgInN1YiI6ICIzOGMwMzBhMC01N2E0LTQ2ZGMtYWM5Zi1lNzBkMDQ5ZTMwMTYiLCAiYWRtaW4iOiB0cnVlfQ'
  signature: b'EvNgTtgbgxpEmJedAOwLuxf6YSq09N2GCRmQOxF2REs'

We use the urlsafe_b64encode function to encode the modified payload and then we replace the original encoded payload with this new value. Note that the header and signature portions are the same as the original access token that was generated by the server.

>>> access_token_mod = b'.'.join(split)
>>> result = User.decode_access_token(access_token_mod)
>>> result.success
False
>>> result.error
'Invalid token. Please log in again.'
>>> exit()

As expected, the modified access token is not decoded successfully and the error message indicates that the token is invalid, which is expected if the contents of the token have been modified in any way. Therefore, our hypothetical malicious user would not be able to give himself admin access by modifying the JWT. Crisis averted!

Let’s run tox and make sure that all test cases pass:

(flask-api-tutorial) flask-api-tutorial $ tox
GLOB sdist-make: /Users/aaronluna/Projects/flask-api-tutorial/setup.py
py37 create: /Users/aaronluna/Projects/flask-api-tutorial/.tox/py37
py37 installdeps: black, flake8, pydocstyle, pytest, pytest-black, pytest-clarity, pytest-dotenv, pytest-flake8, pytest-flask
py37 inst: /Users/aaronluna/Projects/flask-api-tutorial/.tox/.tmp/package/1/flask-api-tutorial-0.1.zip
py37 installed: alembic==1.4.0,aniso8601==8.0.0,appdirs==1.4.3,attrs==19.3.0,bcrypt==3.1.7,black==19.10b0,certifi==2019.11.28,cffi==1.14.0,chardet==3.0.4,Click==7.0,entrypoints==0.3,flake8==3.7.9,Flask==1.1.1,flask-api-tutorial==0.1,Flask-Bcrypt==0.7.1,Flask-Cors==3.0.8,Flask-Migrate==2.5.2,flask-restx==0.1.1,Flask-SQLAlchemy==2.4.1,idna==2.9,importlib-metadata==1.5.0,itsdangerous==1.1.0,Jinja2==2.11.1,jsonschema==3.2.0,Mako==1.1.1,MarkupSafe==1.1.1,mccabe==0.6.1,more-itertools==8.2.0,packaging==20.1,pathspec==0.7.0,pluggy==0.13.1,py==1.8.1,pycodestyle==2.5.0,pycparser==2.19,pydocstyle==5.0.2,pyflakes==2.1.1,PyJWT==1.7.1,pyparsing==2.4.6,pyrsistent==0.15.7,pytest==5.3.5,pytest-black==0.3.8,pytest-clarity==0.3.0a0,pytest-dotenv==0.4.0,pytest-flake8==1.0.4,pytest-flask==0.15.1,python-dateutil==2.8.1,python-dotenv==0.11.0,python-editor==1.0.4,pytz==2019.3,regex==2020.2.20,requests==2.23.0,six==1.14.0,snowballstemmer==2.0.0,SQLAlchemy==1.3.13,termcolor==1.1.0,toml==0.10.0,typed-ast==1.4.1,urllib3==1.25.8,wcwidth==0.1.8,Werkzeug==0.16.1,zipp==3.0.0
py37 run-test-pre: PYTHONHASHSEED='2592492654'
py37 run-test: commands[0] | pytest
================================================= test session starts ==================================================
platform darwin -- Python 3.7.6, pytest-5.3.5, py-1.8.1, pluggy-0.13.1 -- /Users/aaronluna/Projects/flask-api-tutorial/.tox/py37/bin/python
cachedir: .tox/py37/.pytest_cache
rootdir: /Users/aaronluna/Projects/flask-api-tutorial, inifile: pytest.ini
plugins: clarity-0.3.0a0, black-0.3.8, dotenv-0.4.0, flask-0.15.1, flake8-1.0.4
collected 41 items

run.py::FLAKE8 PASSED                                                                                            [  2%]
run.py::BLACK PASSED                                                                                             [  4%]
setup.py::FLAKE8 PASSED                                                                                          [  7%]
setup.py::BLACK PASSED                                                                                           [  9%]
src/flask_api_tutorial/__init__.py::FLAKE8 PASSED                                                                [ 12%]
src/flask_api_tutorial/__init__.py::BLACK PASSED                                                                 [ 14%]
src/flask_api_tutorial/config.py::FLAKE8 PASSED                                                                  [ 17%]
src/flask_api_tutorial/config.py::BLACK PASSED                                                                   [ 19%]
src/flask_api_tutorial/api/__init__.py::FLAKE8 PASSED                                                            [ 21%]
src/flask_api_tutorial/api/__init__.py::BLACK PASSED                                                             [ 24%]
src/flask_api_tutorial/api/auth/__init__.py::FLAKE8 PASSED                                                       [ 26%]
src/flask_api_tutorial/api/auth/__init__.py::BLACK PASSED                                                        [ 29%]
src/flask_api_tutorial/api/widgets/__init__.py::FLAKE8 PASSED                                                    [ 31%]
src/flask_api_tutorial/api/widgets/__init__.py::BLACK PASSED                                                     [ 34%]
src/flask_api_tutorial/models/__init__.py::FLAKE8 PASSED                                                         [ 36%]
src/flask_api_tutorial/models/__init__.py::BLACK PASSED                                                          [ 39%]
src/flask_api_tutorial/models/user.py::FLAKE8 PASSED                                                             [ 41%]
src/flask_api_tutorial/models/user.py::BLACK PASSED                                                              [ 43%]
src/flask_api_tutorial/util/__init__.py::FLAKE8 PASSED                                                           [ 46%]
src/flask_api_tutorial/util/__init__.py::BLACK PASSED                                                            [ 48%]
src/flask_api_tutorial/util/datetime_util.py::FLAKE8 PASSED                                                      [ 51%]
src/flask_api_tutorial/util/datetime_util.py::BLACK PASSED                                                       [ 53%]
src/flask_api_tutorial/util/result.py::FLAKE8 PASSED                                                             [ 56%]
src/flask_api_tutorial/util/result.py::BLACK PASSED                                                              [ 58%]
tests/__init__.py::FLAKE8 PASSED                                                                                 [ 60%]
tests/__init__.py::BLACK PASSED                                                                                  [ 63%]
tests/conftest.py::FLAKE8 PASSED                                                                                 [ 65%]
tests/conftest.py::BLACK PASSED                                                                                  [ 68%]
tests/test_config.py::FLAKE8 PASSED                                                                              [ 70%]
tests/test_config.py::BLACK PASSED                                                                               [ 73%]
tests/test_config.py::test_config_development PASSED                                                             [ 75%]
tests/test_config.py::test_config_testing PASSED                                                                 [ 78%]
tests/test_config.py::test_config_production PASSED                                                              [ 80%]
tests/test_user.py::FLAKE8 PASSED                                                                                [ 82%]
tests/test_user.py::BLACK PASSED                                                                                 [ 85%]
tests/test_user.py::test_encode_access_token PASSED                                                              [ 87%]
tests/test_user.py::test_decode_access_token_success PASSED                                                      [ 90%]
tests/test_user.py::test_decode_access_token_expired PASSED                                                      [ 92%]
tests/test_user.py::test_decode_access_token_invalid PASSED                                                      [ 95%]
tests/util.py::FLAKE8 PASSED                                                                                     [ 97%]
tests/util.py::BLACK PASSED                                                                                      [100%]

================================================= 41 passed in 12.27s ==================================================
_______________________________________________________ summary ________________________________________________________
  py37: commands succeeded
  congratulations :)

Checkpoint

I promise that the pace will pick up, since we have again made very little progress on the API requirements. I believe it’s fair to say that one item is completely implemented: JWT contains the following claims: time the token was issued, time the token expires, a value that identifies the user, and a flag that indicates if the user has administrator access. I think we can also claim partial credit on two items: Requests must be rejected if JWT has been modified and Requests must be rejected if JWT is expired.

As always, leave your feedback and questions in the comments!