Create Robust Access Control in Your Flask Application using Flask-login & Permify

FAYEMI BOLUWATIFE — Fri, 24 Feb 2023 19:43:31 +0000

Introduction

Access control is a popular mechanism used to protect and restrict access to resources and data in applications, networks, or organizations. As a result, it’s a really important concept in application security and privacy.

Having been aware of the main purpose of access control, it is important to note that access control cannot be implemented without authentication and authorization. In other words, if resources and data need to be protected, a way must be devised in order to identify who is going to access the data or resources (i.e Authentication). Also, if certain resources and data need to be accessed by a certain person, a mechanism must be devised to permit certain users (i.e authorization).

There are different models of access control.

Role-Based Access Control (short for RBAC)
Relational-Based Access Control (short for ReBAC)
Attribute-Based Access Control (short for ABAC)

In this article, we are going to focus on implementing RBAC using Flask-Login & Permify in a Blog Application. By the end of this article, you would have completed the following:

Create a Flask Server
Handle Authentication with Flask-login
Build Authorization model with Permify
Set up Permify and connect it to a Flask app
Build blog endpoints
Add dummy organizations & resources and store them in Flask app
Test the endpoints with the dummy resources out

Overview of the Blog Application

Before we dive into building the blog application with Flask, let’s first analyze the blog application requirements and see how the RBAC model can be applied.

The Blog Application is a simple application where users can create posts. The users of this blog application will be grouped into three, based on common responsibilities. The purpose of having three different types of users is so that the actions different users can perform can be managed appropriately (i.e roles). As you would see in the description for the user type below.

Member: Member of an organization, which can have multiple blog posts inside it.
Administrator: Administrator in an organization, can view, edit and delete the posts. This is the superior user type.
Blog Owner: The owner of the blog posts, can also edit and delete the blog post.

The users type listed above can likened to be the roles while the permissions assigned to the roles are defined in the description of each user type. In other words, there are three roles in the blog application.

Building the Blog Application

Note: The source code for this project is available on GitHub.

Prerequisites:

In order to follow along with building the Blog Application, you need to have the following:

Python installed.
Basic Knowledge of Flask.
Docker installed.
API Design Platform (Postman will be used in this guide).

Let’s start to build the Blog Application using Flask and the authorization model using Permify.

Step 1: Create the Flask server

Create a virtual environment and install these packages: Flask, Flask_login, flask_sqlalchemy, python-dotenv, requests.

Create a .env file and store a secret key as an environment variable.

Create a folder named src and then create __init__.py and blog.py files in it.

Copy the following codes into the __init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os
from dotenv import load_dotenv
from .blog import blog

load_dotenv()
db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///db.sqlite"
    db.init_app(app)

    create_database(app)

    app.register_blueprint(blog, url_prefix="")
    return app

def create_database(app):
    if not os.path.exists("instance/" + "db.sqlite"):
        with app.app_context():
            db.create_all()
        print("Database created")

Copy the following codes into the blog.py.

from flask import Blueprint, jsonify

blog = Blueprint("views", __name__)

@blog.get("/")
def status():
    return jsonify({
        "status": "Up and running"
    }), 200

Create a new file named app.py in the project root directory (note inside src/ folder) and copy the following codes into it.

from src import create_app

if __name__ == "__main__":
    app = create_app()
    app.run()

Run python app.py command in the root directory to start the server and send a get request to http://127.0.0.1:5000/ to see the status response.

Now that you’ve created the Flask server, let’s proceed to set up the authentication.

Step 2: Handle Authentication with Flask Login

We will create a simple login & register endpoint.

Create a models.py file in the src/ folder and copy the following codes into it.

from . import db
from flask_login import UserMixin
from sqlalchemy.sql import func

class Organization(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, unique=True, nullable=False)
    creator_id = db.Column(
        db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"), nullable=True
    )

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String, unique=True, nullable=False)
    username = db.Column(db.String(30), unique=True)
    password = db.Column(db.String(30))
    organization_id = db.Column(
        db.Integer, db.ForeignKey("organization.id", ondelete="CASCADE"), nullable=True
    )
    created_organizations = db.relationship(
        "Organization", backref="admin", primaryjoin="User.id==Organization.creator_id"
    )
    organization = db.relationship(
        "Organization", backref="members", foreign_keys=[organization_id]
    )

Create auth.py file in src/ folder and copy the following code into it.

from flask import Blueprint, request, jsonify
from flask_login import login_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from .models import User, Organization
from sqlalchemy.exc import IntegrityError
from . import db
from . import permify


auth = Blueprint("auth", __name__)


@auth.post("/login")
def login():
    data = request.json
    email = data.get("email")
    password = data.get("password")
    user = User.query.filter_by(email=email).first()
    if user:
        if check_password_hash(user.password, password):
            login_user(user, remember=True)
            return (
                jsonify(
                    {
                        "status": "SUCCESS",
                        "message": "User logged in successfully",
                    }
                ),
                200,
            )
        else:
            return (
                jsonify(
                    {
                        "status": "error",
                        "message": "Incorrect password",
                    }
                ),
                400,
            )
    else:
        return (
            jsonify(
                {
                    "status": "error",
                    "message": "User does not exist",
                }
            ),
            400,
        )


@auth.get("/me")
def me():
    if current_user.is_authenticated:
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": "User logged in successfully",
                    "data": {"username": current_user.username},
                }
            ),
            200,
        )
    else:
        return (
            jsonify(
                {
                    "status": "ERROR",
                    "message": "Not authenticated",
                }
            ),
            400,
        )


@auth.post("/register")
def register():
    data = request.json
    username = data.get("username")
    email = data.get("email")
    password_1 = data.get("password-1")
    password_2 = data.get("password-2")
    if password_1 != password_2:
        return (
            jsonify(
                {
                    "status": "error",
                    "message": "Password does not match",
                }
            ),
            400,
        )
    else:
        user = User(
            email=email,
            username=username,
            password=generate_password_hash(password_1, "sha256"),
        )
        try:
            db.session.add(user)
            db.session.commit()
        except IntegrityError as e:
            print(f"An Error occurred {e}")
        else:
            login_user(user, remember=True)
            return (
                jsonify(
                    {
                        "status": "SUCCESS",
                        "message": "User logged in successfully",
                    }
                ),
                200,
            )

Configuring Flask-Login to handle Authentication

Before the login and register endpoint could work without throwing an error, you need to configure flask_login by setting the flask app's secret key and configuring the login manager.

Update the __init__.py file in src/ folder with the code below

from flask_login import LoginManager

def create_app():
    ...  # some omitted code 
    app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
    ...  #  some omitted code
    login_manager = LoginManager() # handles session for users 
    login_manager.login_view = "auth.login"
    login_manager.init_app(app)

    @login_manager.user_loader
    def load_user(id):
        with app.app_context():
            return User.query.get(int(id))

    app.register_blueprint(blog, url_prefix="")
    app.register_blueprint(auth, url_prefix="/auth")
    return app

Step 3: Build Authorization Model with Permify

We will use Permify schema to model our authorization logic. It will be used to define entities, relations between them, and access control decisions.

We will be making use of the permify playground to build the authorization model. As it offers more flexibility and convenience.

Note: Authorization model can be created outside of the permify playground. As a matter of fact, it can be created in any IDE or text editor such as VS Code.

It’s important to mention that you will realize the benefit of creating the model in the playground in a bit.

There are only three entities in our blog application, which are:

User
Organization
Post

The very first step to building Permify Schema is creating Entities.

An entity is an object that defines your resources that hold the role in your permission system.

Think of entities as tables in relational databases or as collections in non-relational databases.

Note: It’s advisable to name entities the same as your database table name that it corresponds to. In that way, you can easily model and reason your authorization as well as to eliminate the possibility of creating errors.

You can create entities using the entity keyword.

entity user {}

entity organization {
   relation admin  @user
   relation member @user

   action create_post = admin or member
   action delete = admin
}

entity post {

   relation author @user
   relation organization @organization

   action edit = author or organization.admin
   action delete = author or organization.admin

}

You can copy the above schema snippet into your playground.

What does the above code snippet do:

We created three different entities using the entity keyword
An Entity has 2 different attributes which are: Relations and Actions.
We made use of the relation keyword to create relationships between the previously created entities.
We also made use of the action keyword to describe what relations or relation’s relations can do.

Let's take a look at some of the actions:

For organization:

action create_post= admin or member indicates that only the admin or member has permission to create posts in the organization.
action delete = admin indicates that only the admin can delete the organization.
For post:
action edit = author or organization.admin indicates that only the author or organization admin has permission to edit posts.
action delete = author or organization.admin indicates that only the author or organization admin can delete posts.

Now that we have the authorization model setup with Permify Schema, let’s start Permify locally and connect the flask app to it.

Step 4: Set Up Permify and connect with the Flask app

There are several options to set up Permify service. However, in this article, I’ll use the docker option by running it via docker container.

Start Permify service by simply running the docker command below in your terminal or CLI.

docker run -p 3476:3476 -p 3478:3478  ghcr.io/permify/permify serve

The above command will download the permify image if you don’t have it locally i.e if you haven’t used permify before. Once it's downloaded, it will be served at the two ports specified in the command.

If the command runs successfully, you should get a message similar to the one in the image below.

Note: If you have docker desktop installed, you can open it and find the running container and click on the Open in Browser icon button.

In case you don’t have docker desktop, you can go to localhost:3476/healthz using postman or your browser to access the permify Rest API.

Hurray! You have successfully set up a permify service on your local computer.

You might be wondering about how you are going to connect the flask app with the Permify service. That’s very easy! Your flask app will simply communicate with Permify service via the REST API provided.

That’s what APIs are for, right?

It’s now time to make use of the authorization model we created earlier on.

As earlier mentioned, we made use of the playground to create the model for a reason. Here is why:

“The Permify Schema needs to be configured and sent to the Permify API in a string format. Therefore, the created model should be converted to a string.
Although it could easily be done programmatically, it could be a little challenging to do it manually. To help with that, we have a button on the playground to copy the created model to the clipboard as a string, so you get your model in string format easily.”

Configuring Authorization Model on Permify API

Click on the Copy button on the playground to copy the authorization model.

Send the copied Permify Schema (i.e authorization model) in the request body of Post request to API endpoint /v1/tenants/t1/schemas/write.

Tip: In order to make testing and communicating with Permify API endpoints easy and seamless, you can fork the Permify API collection. You get access to predefined sample body requests, request methods and even data types of the body to send to the API.

After sending the Post request, you will get a response JSON that contains schema_version.

You can see the Permify API collection in the box labeled 1.

Note: Keep the schema_version returned by the API, it will be used in a later section of this tutorial. Add it as an environment variable to the .env file created in the previous section

We have successfully completed the configuration of the authorization model via Permify Schema. It's now time to add authorization data to see Permify in action.

Creating Relational Tuples

Relational tuples represent authorization data. They are the data used to determine whether a user is authorized on an access control check request.

You can create relational tuples by using /v1/relationships/write endpoint.

According to our authorization model defined in the earlier section, after an organization is created we need to send a post request to permify write endpoint to make the user an admin of the organization i.e we need to create this relational tuple:

organization:<id>#admin@user:<id>

Let’s update the create organization endpoint to include the creation of relational tuple. But before we do that let’s write functions that will be used to communicate with Permify API.

Store the schema version returned when creating the authorization in the .env file created in the previous section.

Create a constants.py file in the src/ folder and copy the following code into it.

from dotenv import load_dotenv
import os 

load_dotenv()
schema_version = os.getenv("PERMIFY_SCHEMA_VERSION")

PERMIFY_BASE_URL = "http://localhost:3476"
PERMIFY_RELATIONAL_TUPLE_URL = f"{PERMIFY_BASE_URL}/v1/tenants/t1/relationships/write"
PERMIFY_CHECK_PERMISSION_URL = f"{PERMIFY_BASE_URL}/v1/tenants/t1/permissions/check"

Create a permify.py file in the src/ folder and copy the following code into it.

import requests
from . import constants
from typing import Dict


def create_relational_tuple(
    entity: Dict,
    relation: str,
    subject: Dict,
    schema_version: str = constants.schema_version,
):
    body = {
        "metadata": {"schema_version": schema_version},
        "tuples": [
            {
                "entity": {"type": entity.get("type"), "id": entity.get("id")},
                "relation": relation,
                "subject": {
                    "type": subject.get("type"),
                    "id": subject.get("id"),
                    "relation": subject.get("relation"),
                },
            }
        ],
    }
    response = requests.post(constants.PERMIFY_RELATIONAL_TUPLE_URL, json=body)
    if response.status_code == 200:
        data = response.json()
        print(data)
        return data["snap_token"]
    else:
        print(response.json())


def check_permission(
    entity: str,
    relation: str,
    subject: str,
    schema_version: str = constants.schema_version,
):
    body = {
        "metadata": {"schema_version": schema_version},
        "tuples": [
            {
                "entity": {"type": entity.get("type"), "id": entity.get("id")},
                "relation": relation,
                "subject": {
                    "type": subject.get("type"),
                    "id": subject.get("id"),
                    "relation": subject.get("relation"),
                },
            }
        ],
    }
    response = requests.post(constants.PERMIFY_RELATIONAL_TUPLE_URL, json=body)
    if response.status_code == 200:
        data = response.json()
        print(data)
        return data["snap_token"]
    else:
        print(response.json())

Import the newly created permify module, and call it just after saving the organization data to the DB as shown below.

@auth.post("/organizations")
@login_required
def create_organization():
    data = request.json
    name = data.get("name")
    print(data)
    organization = Organization(name=name, creator_id=current_user.id)
    try:
        db.session.add(organization)
        db.session.commit()
    except IntegrityError as e:
        print(f"An Error occurred {e}")
    else:
        # create authorization data for this in permify service so it can be used to
        # check for permission later on
        snap_token = permify.create_relational_tuple(
            {
                "type": "organization",
                "id": str(
                    organization.id
                ),  # id of the newly created organization in our db
            },
            "admin",
            {"type": "user", "id": str(organization.id), "relation": ""},
        )
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": "Organization created successfully",
                    "data": {
                        "id": organization.id,
                        "name": organization.name,
                        "admin": organization.creator_id,
                        "snap_token": snap_token
                    },
                }
            ),
            200,
        )

Let’s create an endpoint to allow users to join the created organization as members.

Note: We will also add permify endpoint to this to create this relational tuple: organization:<id>#member@user:<id>.

@auth.post("/organizations/<int:organ_id>/join")
@login_required
def join_organization(organ_id):
    organization = Organization.query.get(organ_id)
    if organization:
        user_id = current_user.id
        user = User.query.get(user_id)
        user.organization_id = organ_id
        db.session.commit()
        # create authorization data for this in permify service so it can be used to
        # check for permission later on
        snap_token = permify.create_relational_tuple(
            {
                "type": "organization",
                "id": str(organ_id),  # id organization in our db
            },
            "member",
            {"type": "user", "id": str(current_user.id), "relation": ""},
        )
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": f"{current_user.username} now member of {organization.name}",
                    "data": {
                        "id": organization.id,
                        "name": organization.name,
                        "admin": organization.creator_id,
                        "snap_token": snap_token,
                    },
                }
            ),
            200,
        )
    else:

        return (
            jsonify(
                {
                    "status": "ERROR",
                    "message": f"The organization does not exist",
                    "data": None,
                }
            ),
            404,
        )

We are done with the authentication and authorization part of the Blog Application. What's left now is the resources (i.e Blog posts) we want to restrict access to. Proceed to the next section to start building the blog post functionalities.

Step 5: Build Blog Endpoints

We'll create the following endpoints:

A POST /posts API route to create posts

A GET /posts/:id API route to view a post

A PUT /posts/:id API route to edit a post

A DELETE /posts/:id API route to delete a post

Note: We will apply Permify check requests on each endpoint listed above to control the authorization. And the resource-based authorization checks (in the form of Can user U perform action Y in resource Z?) will be used extensively.

Remember, we have created User and Organization models, but we haven’t created Post model. Let’s do that now.

Add the following code to the models.py file.

... #some omitted code
class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50))
    date_created = db.Column(db.DateTime(timezone=True), default=func.now())
    content = db.Column(db.Text, nullable=False)
    author_id = db.Column(
        db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"), nullable=False
    )
    organization_id = db.Column(
        db.Integer, db.ForeignKey("organization.id", ondelete="CASCADE"), nullable=False
    )

Note: In order for the post table to be created, we’ll need to delete the previously created db.sqlite file in the instance folder.

As done in the creating relational tuples section, we’ll also create relational tuples here in the create a post endpoint so our authorization data on Permify writeDB can be accurate and up-to-date.

Add the following codes to the blog.py file.

from flask import Blueprint, jsonify, request
from flask_login import login_required, current_user
from .models import Post, Organization
from . import db
from .decorators import create_post_check, edit_post_check, delete_post_check
from . import permify
...

@blog.post("/organizations/<int:organ_id>/posts")
@login_required
@create_post_check
def create_post(organ_id):
    print("type of organ_id ", organ_id)
    organization = Organization.query.get(organ_id)
    print("type of organ_id ", organ_id)
    if organization is None:
        return (
            jsonify(
                {"status": "ERROR", "message": "Organization not found", "data": None}
            ),
            404,
        )
    data = request.json
    title = data.get("title")
    content = data.get("content")
    author_id = current_user.id
    post = Post(
        title=title,
        content=content,
        author_id=author_id,
        organization_id=organ_id,
    )
    db.session.add(post)
    db.session.commit()
    # create authorization data in permify
    snap_token_1 = permify.create_relational_tuple(
        {
            "type": "post",
            "id": str(post.id),  # id of the newly created post in our db
        },
        "author",
        {"type": "user", "id": str(post.author_id), "relation": ""},
    )
    snap_token_2 = permify.create_relational_tuple(
        {
            "type": "post",
            "id": str(post.id),  # id of the newly created post in our db
        },
        "organization",
        {"type": "organization", "id": str(post.organization_id), "relation": "..."},
    )
    return (
        jsonify(
            {
                "status": "SUCCESS",
                "message": f"Post created successfully",
                "data": {
                    "id": post.id,
                    "title": post.title,
                    "date_created": post.date_created,
                    "content": post.content,
                    "author": post.author_id,
                    "organization": post.organization_id,
                },
            }
        ),
        201,
    )


@blog.route("/posts/<int:post_id>", methods=["GET", "PUT", "DELETE"])
@edit_post_check
@delete_post_check
def post(post_id):
    post = Post.query.get(post_id)
    if post is None:
        return (
            jsonify({"status": "ERROR", "message": "Post not found", "data": None}),
            404,
        )
    if request.method == "GET":
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": f"Post with id: {post_id} retrieved successfully",
                    "data": {
                        "id": post.id,
                        "title": post.title,
                        "date_created": post.date_created,
                        "content": post.content,
                        "author": post.author_id,
                        "organization": post.organization_id,
                    },
                }
            ),
            200,
        )
    elif request.method == "PUT":
        data = request.json
        post.title = data.get("title")
        post.date_created = data.get("date_created")
        post.content = data.get("content")
        post.author_id = data.get("author_id")
        post.organization_id = data.get("organization_id")
        db.session.commit()
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": f"Post with id: {post_id} updated successfully",
                    "data": {
                        "id": post.id,
                        "title": post.title,
                        "date_created": post.date_created,
                        "content": post.content,
                        "author": post.author_id,
                        "organization": post.organization_id,
                    },
                }
            ),
            200,
        )
    elif request.method == "DELETE":
        db.session.delete(post)
        db.session.commit()
        return (
            jsonify(
                {
                    "status": "SUCCESS",
                    "message": f"Post with id: {post_id} deleted successfully",
                    "data": None,
                }
            ),
            204,
        )
    return jsonify({"status": "Up and running"}), 200

Create decorator.py file inside the src/ folder and copy the following codes into it.

from flask_login import current_user
from flask import request
from functools import wraps
from . import permify
from flask import abort

def create_post_check(view_func):
    @wraps(view_func)
    def wrapper(*args, **kwargs):
        # call permify func to check for permission
        can = permify.check_permission(
            {"type": "organization", "id": str(kwargs.get("organ_id"))},
            "create_post",
            {"type": "user", "id": str(current_user.id), "relation": ""},
        )
        if can == "RESULT_ALLOWED":
            return view_func(*args, **kwargs)
        return abort(403)

    return wrapper


def delete_post_check(view_func):
    @wraps(view_func)
    def wrapper(*args, **kwargs):
        # due to the route chaining used in the views
        # we need to distinguish between each request
        # so a delete action access check won't impair the edit access
        # check
        if request.method == "DELETE":
            # call permify func to check for permission
            can = permify.check_permission(
                {"type": "post", "id": str(kwargs.get("post_id"))},
                "delete",
                {"type": "user", "id": str(current_user.id), "relation": ""},
            )
            if can == "RESULT_ALLOWED":
                return view_func(*args, **kwargs)
            return abort(403)
        return view_func(*args, **kwargs)

    return wrapper


def edit_post_check(view_func):
    @wraps(view_func)
    def wrapper(*args, **kwargs):
        if request.method == "PUT":
            # call permify func to check for permission
            can = permify.check_permission(
                {"type": "post", "id": str(kwargs.get("post_id"))},
                "edit",
                {"type": "user", "id": str(current_user.id), "relation": ""},
            )
            if can == "RESULT_ALLOWED":
                return view_func(*args, **kwargs)
            return abort(403)
        return view_func(*args, **kwargs)
    return wrapper

Note: We added the authorization check in the decorators.py file so as to avoid repetition all over our blog endpoints (i.e adhere to the DRY principle).

What’s happening in the above code snippet:

We created three decorators to ensure that a User has permission to perform a certain action on the blog resource.
Within each decorator, we make API requests to the appropriate Permify REST API endpoints to determine whether a user’s request should be granted or denied.
If the request should be denied, we return abort(403), which will raise an HTTP exception with the status code 403 (forbidden).
If the request should be allowed, we grant the user access.

Step 6: Add Dummy Organizations & Resources and Store them in Flask App

Now that we have the following ready and completed:

Flask app
Permify service
Access control checks added to the appropriate endpoints

It’s time to access the write endpoints and create users, organizations, and posts to use as dummy data for testing in the next step.

I have created a postman collection to make things easier for you.

Create an organization using the newly registered user.

Register another user.
Make the second registered user a member of the organization created in no. 2 above.

Register a third user.
Create an organization using the registered user in no. 5 above.

Step 7: Test the Endpoints with the Dummy Resources out

Log in as the first user created in Step 6. I.e the admin of the first organization created.
Note: You can send a GET request to the me/ endpoint to know the currently logged-in user.
Try creating a post for a second organization which obviously should fail as the user isn’t an admin nor a member of the organization.

Change the organization in no. 2 above to the first organization in which the logged-in user is an admin.

Log in as the second user and try deleting the post created in no. 2 above. This should fail because even though the second user is a member of the organization, he isn’t the author nor an admin of the organization.

Conclusion

You have just learned about how to use flask-login to set up session authentication and Permify to decouple authorization modeling, data, and logic away from your core application. If you have questions feel free to reach out.

DEV Community: FAYEMI BOLUWATIFE