JWT authorization in Flask

JSON Web Tokens (JWT) are very popular nowadays. Modern web-development is aimed at building Single Page Applications (SPA) using latest JavaScript libraries such as Angular, React or Vue. Because of that reason, JWT becomes a standard of authorization and communication between SPAs and web servers. In this article, I want to build a Flask web server with JWT authorization.

Full source code + code for each part you can find here:

https://github.com/oleg-agapov/flask-jwt-auth

Attention! [edit: Apr 2021]

The code in this article is working, but uses old versions of the packages. If you want to follow along you need to install versions listed here.

If you want to use newer versions (for example Flask-JWT-Extended v4.x) please see this PR with updated code.

Step 1. Project scaffolding

Let’s start with setting up Flask and all necessary extensions.

$ mkdir flask-jwt
$ cd flask-jwt
$ virtualenv -p python3 venv
# following command will activate virtual environment on macOs/Linux
$ source venv/bin/activate
# on Windows run next 
# (see here https://virtualenv.pypa.io/en/stable/userguide/)
# \venv\Scripts\activate

After activation of virtual environment install following packages:

(venv) pip install flask flask-restful flask-jwt-extended passlib flask-sqlalchemy

Here we have:

• Flask-RESTful — to create API endpoints
• Flask-JWT-Extended — to generate and check JWT
• passlib — to generate secure password hashes
• Flask-SQLAlchemy — Object-relational mapping to database

For the sake of simplicity, I’ll use a flat structure of the project. So, inside /flask-jwt folder create 4 files:

flask-jwt
├── views.py     # views of the server
├── models.py    # database models
├── resources.py # API endpoints
└── run.py       # main script to start the server

Let’s ensure that Flask is installed correctly. Add the next code to run.py:

from flask import Flask
app = Flask(__name__)
import views, models, resources

And the next code to views.py:

from run import app
from flask import jsonify
@app.route('/')
def index():
    return jsonify({'message': 'Hello, World!'})

It’s a very basic “Hello World” application. Files models.py and resources.py are empty for now. Start a server with the following command:

(venv) FLASK_APP=run.py FLASK_DEBUG=1 flask run

You should see a standard message saying that server is running:

 * Serving Flask app “run”
 * Forcing debug mode on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 268–350–916

Now open localhost:5000 in the browser. You should get a JSON response with a message “Hello World”.

For better testing of my application I’ll use Postman. It’s an application which allows performing different kinds of requests to the server. Feel free to download and install it.

Step 2. Setting up endpoints

Next logical step in building our application would be setting up API endpoints. They will not do anything meaningful for now, but they will serve as a good starting point in understanding the overall app structure.

Open resources.py and add the following code:

At the very top of the file we imported Resource class from Flask-RESTful extension. Flask-RESTful gives us a handy way to build RESTful APIs. Each endpoint in REST API is called a resource. Above mentioned extension has a class Resource, which we inherit and get all the features of API endpoint.

As you can see I created 7 resources. Among them are:

• user registration and login
• logout for access and refresh tokens (I’ll explain this a bit later)
• token refresh (this also I’ll explain later)
• users resource for getting a list of registered users (only for testing purposes)
• secret endpoint (to test access with tokens)

All of the resources except the last two will accept POST requests, AllUsers resource will have GET and DELETE, Secret resource accept only GET requests. If you are familiar with developing pure Flask APIs you should know that server response should be wrapped with jsonify() function. Flask-RESTful does it for you so you don’t have to jsonify the response.

The next step is to register our endpoints inside our application. Open run.py and add the following code (I give the full source code of the file):

Here we imported Api class from flask-restful to initialize the APIs. Then we create a new api object and attach our app to it. And finally, we add resources to corresponding endpoints.

You can test them now. Try to send a POST request to localhost:5000/registration in Postman. You should get a response message ‘User registration’. You can try other resources as well.

Now let’s add parsing of incoming data inside POST requests. Flask-RESTful comes with a built-in parser. Add the following code to the top of resources.py:

From flask-restful we imported method reqparse which does all the magic. Here we first initialize parser with reqparse.RequestParser(). Then we add parameters to parse: username and password. They are both required parameters. See the documentation for more options.

To use the parser you have to refer it inside your resource. For example like this:

Now every call to /registration or /login endpoint will require two parameters in the body. Let’s check it in Postman. If you try to access these resources without passing a data in the body you’ll see an error.

Now let’s fix this. In Postman add new Header with key Content-Type and value application/json. Next, in Body add raw JSON:

{
    "username": "test",
    "password": "test"
}

If you try to make a call once again it should return the data you sent in Body.

Step 3. User registration and login

In this part we will add support of database and simulate user registration and user login. Let’s begin with the database. I’ll use SQLAlchemy extension to communicate with SQLite database. With some simple manipulations you’ll be able to switch to MySQL or PostgreSQL if you like. Open run.py and add the following:

So, we’ve added SQLAlchemy import to the top. Then somewhere after app initialization add SQLAlchemy configs and create a db object which will connect database to our app. db.create_all() method will create all necessary tables for us.

We have now configured a database. Let’s create a user model. In models.py:

On the top of the file we imported db object from the file where we initialized the database connection. Then we declare UserModel class. For now it has the primary key id, username (which is unique) and password (cannot be a Null). To start using the model we’ve added method save_to_db() which will save the instance of the class to database. Let’s see how it works. Open resources.py and modify UserRegistration resource:

This endpoint works in next way. First of all, we importing UserModel from models.py. Then, after receiving a request with the credentials we create a new user using UserModel class and passing username and password from the request. Then we try to apply save_to_db() method to it and if it works we return a message that a new user was created. Otherwise, we raise 500 status code with message ‘Something went wrong’.

Try to make a request to /registration endpoint with some test data in the body. You should receive a message that user was created.

And if you try to send a request with the same credentials again you will get an error ‘Something went wrong’. This happens because we already have a user with such username. Let’s add a more meaningful answer in this case.

First of all, add this method to the UserModel class in models.py:

@classmethod
def find_by_username(cls, username):
   return cls.query.filter_by(username = username).first()

This method will return a user’s data if there is match by username. Now change UserRegistration resource (resources.py):

Here we first do a check, if the user with given username already exists. If it’s true — we return a message that we already have that username. Now it’s time to implement UserLogin endpoint:

This endpoint will do the following:

• first, we parse the request parameters
• then we do a lookup by username
• if a user with such username doesn’t exist we return the error message
• if a user exists we check his password
• if passwords matched we saying to the user that he is logged in
• otherwise, return the error

Create a new request to /login path in Postman and make a POST request with credentials you posted to /registration route. You should get a message that you are logged in.

If you recall the endpoints we setup earlier, we have a resource called AllUsers. I want to use it in testing purposes to look what we have in users table. Let’s make it works. In resources.py change AllUsers class to:

The idea behind this endpoint will be following. If you send GET request to /users resource you should get a list of all registered users; if you send DELETE request to the same resource all users should be deleted. Open models.py and add return_all() and delete_all() methods:

Method return_all() will return an object with the list of all registered users. Method delete_all() will delete all users and return a number of deleted rows.

We almost finished with registration/login processes. Our app has one flaw — we store raw user’s passwords. It could be a problem if someone gets access to users table. Let’s use passlib extension to encode passwords.

Open models.py and add:

from passlib.hash import pbkdf2_sha256 as sha256
class UserModel(db.Model):
    ...
    @staticmethod
    def generate_hash(password):
        return sha256.hash(password)
    @staticmethod
    def verify_hash(password, hash):
        return sha256.verify(password, hash)

Method generate_hash() will generate a hashed string which we will store in the database. Method verify_hash() we will use to check the given password.

Modify UserRegistration in resources.py to store hashed password:

new_user = UserModel(
    username = data['username'],
    password = UserModel.generate_hash(data['password'])
)

And add verifying of password with hash in UserLogin:

if UserModel.verify_hash(data['password'], current_user.password):
    ...
else:
    ...

Now user’s password are encoded and protected:

Step 4. Adding JWT

In this part, we’ll actually add JSON Web Token authorization to our app. Let’s begin with run.py:

from flask_jwt_extended import JWTManager
app.config['JWT_SECRET_KEY'] = 'jwt-secret-string'
jwt = JWTManager(app)

Add import of JWTManager class from flask-jwt-extended library to the import section. Next, after initializing SQLAlchemy add JWT secret key constant and initialize JWT by passing our app instance to JWTManager class.

Add the following import to resources.py:

from flask_jwt_extended import (create_access_token, create_refresh_token, jwt_required, jwt_refresh_token_required, get_jwt_identity, get_raw_jwt)

Here we import all necessary methods to work with tokens. Let’s look how they work. Change UserRegistration and UserLogin resources to return the tokens in case of successful registration or login:

So, in case of successful registration or login, we will return to user two tokens: access token and refresh token. For tokens generation we use two functions: create_access_token() and create_refresh_token(). Each function accepts at least one argument — identity. I simply passed username as identity, but you can pass even complex objects. Access token we need to access protected routes. Refresh token we need to reissue access token when it will expire.

To create a protected resource you simply need to add a @jwt_required decorator to it. Like this:

class SecretResource(Resource):
    @jwt_required
    def get(self):
        return {
            'answer': 42
        }

Now to access this resource you need to add a header to your request in format Authorization: Bearer <JWT>. Go and login once again and copy access token from the response. Then create a request to /secret and add access token to Authorization header:

As I mentioned earlier, tokens have an expiration date. By default, access tokens have 15 minutes lifetime, refresh tokens — 30 days. In order not to ask users to log in too often after access token expiration you can reissue new access token using refresh token. This is usually a separate endpoint, and we have it. Let’s add functionality to reissue access token with refresh token:

class TokenRefresh(Resource):
    @jwt_refresh_token_required
    def post(self):
        current_user = get_jwt_identity()
        access_token = create_access_token(identity = current_user)
        return {'access_token': access_token}

So, first of all, this resource has jwt_refresh_token_required decorator, which means that you can access this path only using refresh token. By the way, you cannot access jwt_required endpoints using refresh token, and you cannot access jwt_refresh_token_required endpoints using access token. This adds an additional layer of security. To identify user we use helper function get_jwt_identity() which extract identity from refresh token. Then we use this identity to generate a new access token and return it to the user.

And actually, that is all you need to support JWT authorization!

Step 5. Logout and token revoking

Adding logout functionality will require a bit more coding. In contrast to session authorization, we cannot just delete tokens on the client side because these tokens still will be valid (as long as they don’t expire). Thus in case of user logout, we need to add these tokens to the blacklist. Then you need to check all incoming tokens against the blacklist and if there is a match — disallow access.

Here is the simplest implementation of logout functionality. In models.py add revoked token model:

It is a simple model which stores a primary key id and jti — unique identifier of the token. It has also add() method which adds token to the database and is_jti_blacklisted() method which do a check if the token is revoked.

Now in run.py add support of token blacklisting:

app.config['JWT_BLACKLIST_ENABLED'] = True
app.config['JWT_BLACKLIST_TOKEN_CHECKS'] = ['access', 'refresh']

@jwt.token_in_blacklist_loader
def check_if_token_in_blacklist(decrypted_token):
    jti = decrypted_token['jti']
    return models.RevokedTokenModel.is_jti_blacklisted(jti)

First thing is to enable blacklisting in configuration. Then you specify what kind of token you want to check, we have chosen both access and refresh. Next you can see a function check_if_token_in_blacklist() with @jwt.token_in_blacklist_loader decorator. A decorator is a callback and it is called every time client try to access secured endpoints. Function under the decorator should return True or False depending on if the passed token is blacklisted.

The last thing will be implementing logout endpoints in resources.py:

Because we have two tokens — access and refresh — we have two logout endpoint for each. They work pretty simply. We need to extract the unique identifier from the passed token and add it to the database of revoked tokens.

Final words

In this tutorial, I tried to show that adding JWT authorization is not a big deal if you use the right tools. All the source code of this example you can find here:

https://github.com/oleg-agapov/flask-jwt-auth

Still this example can be improved. Possible improvements are:

• Returning tokens in HttpOnly cookies with additional CSRF protection
• A bit different revoked tokens model (see GitHub example)
• Redefining the standard behavior of Flask-JWT-Extended extension using configuration constants (custom tokens expiration date, a custom format of authorization header)

If you like this tutorial and would like to donate me a few bucks💰 you can do it on my paypal account.