Flask RBAC demystified: a developer's guide

Feb 24th, 2022

Eric Goebelbecker

Integration |

RBAC

Flask is one of Python's most popular web frameworks. It's lightweight, easy to master, and extensible. There are few, if any, applications that you can't tackle with Flask, including role-based access control (RBAC).

In this post, we'll look at setting up RBAC with Flask. We'll start with a simple Flask application and roll our own RBAC system.

Let's get started!

What Is RBAC?

Before we get into the code, let's define what RBAC is and how it can help you build an application that enforces user access and security.

RBAC governs user access to an application based on their role. Ideally, you've based those roles on business functions. Each role encapsulates one or more permissions, and a user you assign to a role gains the associated capabilities.

Let's take a look at a simple example. Consider a basic inventory management system.

Managers create items, view them, edit their metadata, delete them, and set their quantities.
Warehouse employees are limited to viewing items and setting their quantities.
Salespeople may only view items.

Let's see how this works with a simple Flask application.

A RESTful API in Flask

Let's start with a Flask application that offers a simple RESTful application programming interface. The source code is here on Github. A working copy without RBAC is at the working_no_rbac tag.

To run the code, you'll need a working Python 3 environment, a tool for sending requests to a RESTful API, and the ability to install Python packages.

Let's look at a few code snippets in that version before moving to Flask RBAC.

To keep the code short and focused on the task at hand, we're storing items in two columns in an SQLite database: one with a name and another with the JSON data. The API implements the REST GET, PUT, POST, and DELETE verbs.

Item names must be unique, which means any attempt to POST an item with an existing name will fail. So, modify an item with the PUT instead.

Even though the initial version doesn't implement RBAC yet, it does enforce HTTP basic authentication. The database has a users table with a username and a hashed password.

Here's the GET request handler.

# Retrieve an item
@app.route('/api/inventory', methods=['GET'])
@auth.login_required
def query_records():
    try:
        name = request.args.get('name')
        with sqlite3.connect("database.db") as con:
            con.row_factory = sqlite3.Row
            cur = con.cursor()
            cur.execute("select data from inventory where name=?", (name, ))
            row = cur.fetchone()
            return row["data"] if row else ("{} not found".format(name), 400)
    except Exception as e:
        abort(500, e)

Since we're storing the data as a JSON column, the app can return it.

Here's the PUT request handler.

# Modify an existing item
@app.route('/api/inventory', methods=['PUT'])
@auth.login_required
def update_record():
    try:
        record = json.loads(request.data)
        with sqlite3.connect("database.db") as con:
            cur = con.cursor()
            cur.execute(
              "INSERT INTO inventory (name, data) VALUES(?, ?) ON CONFLICT(name) DO UPDATE SET data=?",
              (record['name'], request.data, request.data))
            con.commit()
        return jsonify(record)
    except Exception as e:
        abort(500, e)

The POST handler relies on the database to refuse to add an existing name.

# Add a new item
@app.route('/api/inventory', methods=['POST'])
@auth.login_required
def create_record():
    try:
        record = json.loads(request.data)
        print("Putting {}".format(record))
        with sqlite3.connect("database.db") as con:
            cur = con.cursor()
            cur.execute(
              "INSERT INTO inventory (name, data) VALUES(?, ?)",
              (record['name'], request.data))
            con.commit()
        return jsonify(record)
    except Exception as e:
        abort(500, e)

The DELETE handler deletes items using the name.

# Delete an item
@app.route('/api/inventory', methods=['DELETE'])
@auth.login_required
def delete_record():
    try:
        name = request.args.get('name')
        with sqlite3.connect("database.db") as con:
            cur = con.cursor()
            cur.execute(
              "DELETE FROM inventory where name = ?",
              (name, ))
            con.commit()

        return "{} deleted".format(name), 200
    except Exception as e:
        abort(500, e)

Finally, verify_password enforces a login by failing if there's no username or password and checking the password against a hash in the database.

# Verify password against hashed values in the database
@auth.verify_password
def verify_password(username, password):
    if username and password:
        with sqlite3.connect("database.db") as con:
            con.row_factory = sqlite3.Row
            cur = con.cursor()
            cur.execute(
              "select password from users where username=?",
              (username,))
            row = cur.fetchone()
            return check_password_hash(row['password'], password)
    else:
        return False

The database has three users and passwords:

manager: imthemanager
warehouse: imthewarehouse
sales: imsales

You can run this service and test it with any web or RESTful client. Here's an example with cURL.

First, call GET to retrieve an item as the sales user. Pass the username and password with the -u command line flag.

% curl -u sales:imsales -i -X GET "http://localhost:5000/api/inventory?name=X-Men+1+(1991)"
HTTP/1.0 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 41
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sat, 15 Jan 2022 19:17:06 GMT

{"name": "X-Men 1 (1991)", "count": 5000}%

Then, add a new item:

% curl -u sales:imsales --header "Content-Type: application/json" 
--data "{ \"name\": \"Incredible Hulk 180\", \"count\": 2 }" -i -X POST "http://127.0.0.1:5000/api/inventory"
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 41
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sun, 16 Jan 2022 19:50:15 GMT

{"count":2,"name":"Incredible Hulk 180"}

Finally, DELETE the new item.

% curl -u sales:imsales -i -X DELETE "http://127.0.0.1:8081/api/inventory?name=Incredible+Hulk+180"
HTTP/1.0 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 27
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sun, 16 Jan 2022 19:53:10 GMT

Incredible Hulk 180 deleted%

So, we have a working RESTful API server in Flask, but sales can add and delete items. Let's fix that with RBAC.

Adding RBAC to a Flask application

Let's start out by looking at modules.

Flask modules

One of the most significant advantages of working with Flask is that if you need to add a capability to your application, there's probably already a module that does it for you. In this case, there's Flask-Authorize, Flask-RBAC, Flask-ACL, and many more modules that will help you add role-based access control to your Flask app. But each has a slightly different approach, requires a significant amount of scaffolding (such as persistent sessions), and may or may not be well-supported with bug fixes and documentation. In addition, few work well for a simple tutorial with cURL. (At least not without making the command line examples challenging to demonstrate.)

There's also the Aserto Flask API, which you drop into existing code and use to verify access and roles using Aserto's cloud infrastructure.

So, we'll write a few functions that will give this app basic RBAC using the same HTTP basic authentication the app's already using. You can find this code at the working_rbac tag and the head of the main branch.

Defining RBAC roles

The first step in adding role-based access control to your application is defining the roles. So, to keep things simple, we'll assign the three existing users their own roles.

manager can add, delete, retrieve, and modify records in the database.
warehouse can retrieve and modify records.
sales can retrieve records.

Add Flask-login

Even though we're sticking with HTTP basic auth, we need to switch the login module from Flask-HTTPAuth to Flask-Login. Install flask-login with pip and add a few new imports to the top of app.py.

import json
import sqlite3
from flask import Flask, request, jsonify, abort
from werkzeug.security import check_password_hash
from flask_login import LoginManager, login_user, current_user, login_required
import base64
from functools import wraps
from enum import IntEnum

You'll see where you use these names below. The application needs a LoginManager and a secret key. Since we're not using durable sessions, this app won't use the key, but the code won't run without one.
Flask-Login is going to return cookies to cURL. You can turn that off, but we're tight on space here, so I'm not going to add that code.

app = Flask(__name__)
app.secret_key = "ITJUSTDOESNTMATTER"

login_manager = LoginManager()
login_manager.init_app(app)

To enforce roles, we need to associate them with users. So, instead of verifying that a user has passed in a valid username and password combination and returning True or False, the login process has to retrieve the user's role and store it so Flask can verify that the user is allowed access to the route they called.

We need a class for roles and one for users.

The Role class uses IntEnum. This allows you to compare roles using greater than and less than.

Flask-Login expects the User class to implement a few methods. In addition to those, we read the role name out of the user database table and convert it into the correct Role enum and store it in access_level. Then at the bottom of the class, we compare the requested Role to self.access_level.

class Role(IntEnum):
    MANAGER = 3
    WAREHOUSE = 2
    SALES = 1


class User:
    def __init__(self, name, role):
        self.name = name
        self.access_level = Role.SALES
        if role == 'manager':
            self.access_level = Role.MANAGER
        elif role == 'warehouse':
            self.access_level = Role.WAREHOUSE

    @staticmethod
    def is_authenticated():
        return True

    @staticmethod
    def is_active():
        return True

    @staticmethod
    def is_anonymous():
        return False

    def get_id(self):
        return self.name

    def get_role(self):
        return self.access_level

    def allowed(self, access_level):
        return self.access_level >= access_level

Now that we have a way to describe a role and associate it with a user, we need to change how we load users.

Remove verify_password and replace it with a new handler called by the LoginManager. This handler does the following:

Examine the Authorization header.
Extract the username and password.
Retrieve the user information from the database.
Check the password.
If the password is correct, create a user object.
Pass it to the login manager with login_user().
Return the new user object.

@login_manager.request_loader
def load_user_from_request(request_info):

    # Try to login using Basic Auth
    auth_header = request_info.headers.get('Authorization')
    if auth_header:
        auth_header = auth_header.replace('Basic ', '', 1)
        try:
            login_info = base64.b64decode(auth_header)
            login_text = login_info.decode()
            login_info = login_text.split(':')

            user_info = retrieve_user(login_info[0])

            if check_password_hash(user_info['password'], login_info[1]):
                user = User(user_info['name'], user_info['role'])
                login_user(user)
                return user

        except TypeError as e:
            print(e)
            return None


def retrieve_user(username):
    with sqlite3.connect("database.db") as con:
        con.row_factory = sqlite3.Row
        cur = con.cursor()
        cur.execute("select * from users where name=?", (username,))
        row = cur.fetchone()
        return row

So, we now have a user object. How do we check roles? A new decorator compares the user's access level to one defined for the Flask route.

This function accepts a Role. It compares the requested Role with the Role association with current_user when it's called.

The LoginManager exports current_user. If the user's Role is greater than the one requested, the route proceeds.

If there's no current_user, which means the request didn't pass an authentication header, the function triggers a 401. This should never happen. (But we know how that goes!)

def check_access(access_level):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.is_authenticated:
                return abort(401)

            if not current_user.allowed(access_level):
                return abort(401)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

Add this decorator to the routes. Here's what PUT should look like.

# Modify an existing item
@app.route('/api/inventory', methods=['PUT'])
@check_access(Role.WAREHOUSE)
def update_record():
    try:
        record = json.loads(request.data)
        with sqlite3.connect("database.db") as con:
            cur = con.cursor()
            cur.execute(
              "INSERT INTO inventory (name, data) VALUES(?, ?) ON CONFLICT(name) DO UPDATE SET data=?",
              (record['name'], request.data, request.data))
            con.commit()
        return jsonify(record)
    except Exception as e:
        abort(500, e)

Testing Flask RBAC

Now we can run some tests.

Sales can still retrieve items.

% curl -u sales:imsales -i 
-X GET "http://localhost:5000/api/inventory?name=X-Men+1+(1991)"

HTTP/1.0 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 41
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sat, 15 Jan 2022 19:17:06 GMT

{"name": "X-Men 1 (1991)", "count": 5000}%

But they can't add a new one:

 % curl -u sales:imsales --header "Content-Type: application/json" 
 --data "{ \"name\": \"Incredible Hulk 2\", \"count\": 2 }" -i -X POST "http://127.0.0.1:5000/api/inventory"

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 343
Vary: Cookie
Set-Cookie: session=.eJwlzj0KwzAMQOG7eO6gP8tWLlNkS6aFTkkzld69ga4PHnyfcl97Ho-yvfczb-X-jLIVkRyDEdQnC2g1W7R655aBuCoTNzdDEZI2wxgWCxoYQTYIpEEdeVaarhzj-sS7Zo8ru3YxUe4OrpQ4VCogZMVAUWgOAFEuyHnk_tcc_sqjfH-kTy5g.YeSDwQ.c6FTiu00GzM9X2kjU_QEac4wWlg; HttpOnly; Path=/
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sun, 16 Jan 2022 20:44:49 GMT

Only the manager can add a new item.

% curl -u manager:imthemanager --header "Content-Type: application/json" 
--data "{ \"name\": \"Incredible Hulk 2\", \"count\": 2 }" -i -X POST "http://127.0.0.1:5000/api/inventory"

HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 41
Vary: Cookie
Set-Cookie: session=.eJwlzjEOwjAMQNG7ZGawY8eJe5nKiR1ggCGFCXF3KrF-6Uvvk_a54ril7bXecUn73dOWmKN3QhAbxCBFdebZGtVwxFkoUzVVZM5chyvBJEYFzRAVHHPPDWmUPEzI-_mxNYnmZzZprCzUDExyYBcugBAFHVmgGgB4OiHvI9Zf87CnXWOl7w8CTi8j.YeSECg.jMwtuhgKocaX_0JaDz1cKOZPslw; HttpOnly; Path=/
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sun, 16 Jan 2022 20:46:02 GMT

{"count":2,"name":"Incredible Hulk 2"}

They can alter the quantity, too, even though the route is configured for the warehouse role.

% curl -u manager:imthemanager --header "Content-Type: application/json" 
--data "{ \"name\": \"Incredible Hulk 2\", \"count\": 1 }" -i -X PUT "http://127.0.0.1:5000/api/inventory"

HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 41
Vary: Cookie
Set-Cookie: session=.eJwlzjEOwjAMQNG7ZGawY8eJe5nKiR1ggCGFCXF3KrF-6Uvvk_a54ril7bXecUn73dOWmKN3QhAbxCBFdebZGtVwxFkoUzVVZM5chyvBJEYFzRAVHHPPDWmUPEzI-_mxNYnmZzZprCzUDExyYBcugBAFHVmgGgB4OiHvI9Zf87CnXWOl7w8CTi8j.YeSEZQ.H48ne9M7RMYC_INiPbm7SsqFGaI; HttpOnly; Path=/
Server: Werkzeug/2.0.2 Python/3.8.9
Date: Sun, 16 Jan 2022 20:47:33 GMT

{"count":1,"name":"Incredible Hulk 2"}

RBAC and Flask

In this tutorial, we implemented RBAC in a simple proof-of-concept Flask application. We covered the basics of a role-based hierarchy and then used the application to illustrate it in action.

If you want to quickly get your Flask app up and running with RBAC, check out Aserto's Python SDK on Github and sign up today. We also has a helpful article on RBAC best practices. Until next time!

This post was written by Eric Goebelbecker. Eric has worked in the financial markets in New York City for 25 years, developing infrastructure for market data and financial information exchange (FIX) protocol networks. He loves to talk about what makes teams effective (or not so effective!).