I'm a Technical Content Engineer,
bridging the gap between complex technology
and clear, actionable documentation.

FastAPI: http authentication

Since FastAPI is designed to be an API system, it doesn’t come with much in the way of dealing with html, and there’s no Session or Authentication system based on session authentication via html pages.

I created such a system, and along the way learned a lot of important lessons for dealing with FastAPI. So this series is not just about adding authentication, but also about some overall ways of using FastAPI.

My code can be found here.

Libraries Used

Before we get started, I want to go through the python libraries I have used, on top of FastAPI.

python-dotenv

If this library, or a similar one, is not part of your toolbox for web applications, then it should be. There are always user details, connection details, secret keys etc, that should NEVER be put in the repository. DotEnv is the standard for deal with them.

Tortoise ORM

We need a database for user management. The standard is SQLAlchemy, but I’ve never gotten on well with it. It always feels like they’ve added functionality, after functionality. This is great and the library can do everything you want, but it also means there are lots of ways to do the same thing, which can be confusing. Tortoise ORM, is an async ORM, loosely based on the Django ORM. Those familiar with Django will feel more comfortable with Tortoise, but its far from being the same. It handles a lot of things differently, and I think, somewhat better, than Django does.

asyncpg

I’m using a Postgresql database so we need the async driver for connecting to it. There are other options, so change this to the python based driver of your need

Bulma.io

Not a Python library but a CSS lib. I’ve used this for the html pages so they are easy to layout. I’ve also not added a template library to the list for this project. There is a Jinja2 response object available with FastAPI, but its very easy to add any other template language you want, using the HTMLResponse class.

Creating a settings module

I found that I had issues with circular imports, if I read in the dotenv file within the root app file, so I found it best to import all the settings in a file and then just import that when and where needed.

from dotenv import find_dotenv, dotenv_values
import pathlib
import string


file_path = pathlib.Path().cwd()
static_dir = str(pathlib.Path(pathlib.Path().cwd(), "static"))
config = dotenv_values(find_dotenv(".test_fastapi_config.env"))

db_user = config.get("DB_USER")
db_password = config.get("DB_PASSWORD")
db_host = config.get("DB_HOST")
db_name = config.get("DB_NAME")
db_port = config.get("DB_PORT")
db_url = f"asyncpg://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"
db_modules = {"models": ["models"]}

max_age = 3600

session_choices = string.ascii_letters + string.digits + "=+%$#"

You will notice that I like using pathlib over the os library. Its best to put all the settings that will be needed by multiple files in here, but don’t do any of the FastAPI set-up in this file.

Database models and set-up

The first part of this is obvious. I created a models.py file and defined my database tables in that.

from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator


def default_scope():
    return ["authenticated"]


class Users(models.Model):
    """
    The User model
    """

    id = fields.IntField(primary_key=True)
    username = fields.CharField(max_length=20, unique=True)
    first_name = fields.CharField(max_length=50, null=True)
    last_name = fields.CharField(max_length=50, null=True)
    p_hash = fields.BinaryField(max_length=128, null=True)
    p_salt = fields.BinaryField(max_length=128, null=True)
    scope = fields.JSONField(default=default_scope)
    info = fields.JSONField(default=dict)

    class PydanticMeta:
        exclude = ["password_hash"]


User_Pydantic = pydantic_model_creator(Users, name="User")
UserIn_Pydantic = pydantic_model_creator(Users, name="UserIn", exclude_readonly=True)


class Session(models.Model):

    id = fields.IntField(primary_key=True)
    token = fields.CharField(max_length=128, unique=True, db_index=True)
    user = fields.IntField(default=0)
    created_at = fields.DatetimeField(auto_now_add=True)
    expires_at = fields.DatetimeField(auto_now_add=True)


Session_Pydantic = pydantic_model_creator(Session, name="Session")

There are a couple of items we should discuss in these models.

Scope

This was a feature I noticed in the Starlette middleware, where each user has a list of permission levels. The default is just “authenticated”, but you can add “admin”, “super”, etc for latter use for views where we want to grant access to only selected users.

p_hash, p_salt

I took a look at a few different ways of storing passwords and settled on this one. I’m using the built in hashlib library, with the pbkdf2_hmac method with the “sha256” encryption method. Passwords are one way encrypted, using a salt that is unique to the give user. To validate a password, the password need to be encrypted using the p_salt value, which should then match the p_hash value. I did look at using a common salt, saved on the server against this, but it appears this method is considered more secure. Feel free to change if you want to use a different method. The salt is a random 128 byte value which us saved in a Binary Field type

Session Token

For session management, I wanted to follow Django in that the session cookie holds no user information. The session Cookie is given a random 128 character string, which is used to match against the session database. Session records are only created by authenticated users.

Adding the ORM to FastAPI

Adding the ORM makes use of a fairly new addition to FastAPI. That is the lifespan object. You need to get to grips with this to understand whats going on here. Starlette and Fast API allows for actions to be taken on the “startup” and “shutdown” events for the FastAPI/app object. This used to work using.

@app.on_event("startup")
async def my_event():
  pass

This has been depreciated in FastAPI and is being replaced by “lifespan”.

With lifespan, we basically wrap the FastAPI object in a python context manager. This means we are says, run this set of code before the main code block and this second set, once the code block has completed. Thus a sort of middle ware over the main object.

This does need to be an asynccontextmanager, and the “yield” statement marks the before and after sections of the code.

from fastapi import FastAPI
import settings
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from tortoise.contrib.fastapi import RegisterTortoise


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    # app startup
    async with RegisterTortoise(
        app,
        db_url=settings.db_url,
        modules=settings.db_modules,
        generate_schemas=True,
        add_exception_handlers=True,
    ):
        # db connected
        yield
        # app teardown

main_app = FastAPI(lifespan=lifespan)

Do note that I placed the database connection url and the modules in the settings file. modules is used to define where your models are to be found.

A second point to note, which might come up. If you have used the “lifespan” argument, then any “on_event()” method calls will be ignored. Ai assistants etc, do still advise to use the on_event, even though it won’t work.

Middleware

Starlette has middleware the session and authentication, but they don’t work. The FastAPI documentation does have a link to the starlette middleware section, but I found that they only partially work and give unexpected results. Also items like the “require()” decorator don’t work at all.

However, creating the middleware using FastAPI really wasn’t that difficult. I think I spent more time trying to workout why the Starlette middleware didn’t work, than I did writing my own. FastAPI makes is really easy to create middleware.

Session Middleware

For my system, the only thing I needed was a cookie created and resent after a call, where I had a unique token that can be matched to the Session table on the backend.

@main_app.middleware("http")
async def session_middleware(request: Request, call_next):
    cookie_val = request.cookies.get("session")
    if cookie_val:
        request.scope['session'] = cookie_val
    else:
        request.scope['session'] = "".join(random.choices(settings.session_choices, k=128))
    response = await call_next(request)
    response.set_cookie("session", value=request.session,
                        max_age=settings.max_age, httponly=True)
    return response

A couple of point to note here. The session token is just a 128 character string. I limited the values in the string so I didn’t end up with any punctuation that would cause issues.

The second point, and one you might not have noticed is that the cookie is recreated after the view has been processed, I am creating a new cookie. This (from what I’ve read) should reset the expiry time on the cookie in the browser.

Notice that I’m updating using request.scope[“session”]. You cannot update request.session. You need to update the scope dictionary. In the request object. session and auth properties just send back the values in the obj.scope and don’t have a setting method.

One last point. The Request class is the starlette Request class. The “auth” and “session” properties have defaults of error messages, saying the middleware is not loaded, if they are empty. Just be aware your middleware is replacing the Starlette middleware in this case and don’t get confused and try to load the Starlette ones.

Authentication Middleware

Requests is set-up with properties for “user” and “auth” by starlette. This seemed sensible to me, so I kept with those properties.

“auth” should hold the list of values from the User model’s “scope” field. We will be using this later for dependencies that will decide if the user has access to an end point.

“user” in Starlette holds a BaseUser class type. This is useful as it allows us to say if a user is an authenticated user or anonymous user, and has an is_authenticated property. It did however have other items that were not of much use, so I created my own classes to replace these.

from models import Session, Users


class BaseUser:
    @property
    def is_authenticated(self) -> bool:
        raise NotImplementedError()


class UnauthenticatedUser(BaseUser):
    @property
    def is_authenticated(self) -> bool:
        return False


class AuthUser(BaseUser):
    def __init__(self, session: Session) -> None:
        self.session = session
        self.__user = None

    @property
    def is_authenticated(self) -> bool:
        return True

    async def user(self) -> Users:
        if not self.__user:
            self.__user = await Users.get_or_none(id=self.session.user)
        return self.__user

I placed these in auth.__init__.py

So here is the middleware

@main_app.middleware("http")
async def authentication_middleware(request: Request, call_next):
    token = request.cookies.get("session")
    if not token:
        request.scope["auth"] = ["anonymous"]
        request.scope["user"] = UnauthenticatedUser()
    else:
        session = await Session.get_or_none(token=token)
        if session is None:
            request.scope["auth"] = ["anonymous"]
            request.scope["user"] = UnauthenticatedUser()
        else:
            request.scope["user"] = AuthUser(session)
            user = await Users.get_or_none(id=session.user)
            request.scope["auth"] = user.scope
    response = await call_next(request)
    return response

Notice that this only sets the auth and user in the request and nothing else.

Authenticating a User

We need a function to authenticate a user and create a session record if they are valid, which basically sets them as logged in.

I placed this in auth.__init__.py

async def authenticate(request: Request, user: str, password: str) -> bool:
    user = await Users.get_or_none(username=user)
    if user:
        p_hash = hashlib.pbkdf2_hmac("sha256", password.encode(), user.p_salt, 100000)
        if p_hash == user.p_hash:
            expires = datetime.datetime.now() + datetime.timedelta(seconds=max_age)
            if await Session.filter(token=request.session).exists():
                await Session.filter(token=request.session).delete()
            session = await Session.create(user=user.id, expires_at=expires,
                                           scope=user.scope, token=request.session)
            request.scope["session"] = session.token
            request.scope["auth"] = user.scope
            auth_user = AuthUser(session)
            auth_user.__user = user
            request.scope["user"] = auth_user
            return True
        else:
            return False
    else:
        return False

This basically checks the users credentials and if valid, creates a Session record and sets the request object. The deleting of existing session records is just to make sure we don’t end up with duplicates in the table.

Checking User Access and if Logged in

ok, so the user has sent their username and password. This has been checked and a session record created. When they send a request to the server, the request object is updated with session, auth and user details. Now we just need to check those before they can get to the view.

In Django, we would set a logged_in decorator or similar, but with FastAPI, we are going to use Dependency Injection. I’ve added these to the path decorators. I am assuming you are familiar with Dependency Injection and so will realize that dependencies in the path decorator can only pass or raise an error. You can’t redirect from a dependency. FastAPI, of course deals with this using exception_handling.

A Quick explanation.

We create a custom Exception, like NotLoggedInException. Just needs to have a unique name, not much else.

class NotLoggedInException(Exception):
    def __init__(self):
        pass

Our dependency checks the request object and raises the exception if the user is not logged in.

async def logged_in(request: Request, background_tasks: BackgroundTasks) -> bool:
    if not request.user.is_authenticated:
        raise NotLoggedInException()
    background_tasks.add_task(update_session_expiry, request)
    return True

ok so our view has failed with a NotLoggedInException.

We can capture this exception with FastAPI using an exception handler

def not_logged_in_handler(request: Request, exc: NotLoggedInException):
    """Redirect to the login page if login is required"""
    return RedirectResponse("/auth")

app.add_exception_handler(NotLoggedInException, not_logged_in_handler)

So now our dependency raises an exception NotLoggedInException. FastAPI sees that and says, oh run the no_logged_in_handler(). This in turn redirects the user to the login page.

So we can now write a series of dependencies, custom exceptions and exception handlers, to deal with our authentication and permission checking.

from auth.exceptions import NotLoggedInException, PermissionFailedException, AlreadyLoggedInException
from fastapi import Request, BackgroundTasks
import datetime
from settings import max_age
from models import Session


async def update_session_expiry(request: Request):
    session = await Session.get(token=request.session)
    session.expires_at = datetime.datetime.now() + datetime.timedelta(seconds=max_age)
    await session.save()
    await Session.filter(expires_at__lt=datetime.datetime.now()).delete()


async def already_logged_in(request: Request):
    if request.user.is_authenticated:
        raise AlreadyLoggedInException()
    return True


async def logged_in(request: Request, background_tasks: BackgroundTasks) -> bool:
    if not request.user.is_authenticated:
        raise NotLoggedInException()
    background_tasks.add_task(update_session_expiry, request)
    return True


async def admin_user(request: Request, background_tasks: BackgroundTasks) -> bool:
    if not request.user.is_authenticated:
        raise NotLoggedInException()
    if "admin" not in request.auth or "super" not in request.auth:
        raise PermissionFailedException("You need to be marked as an admin user to access this endpoint")
    background_tasks.add_task(update_session_expiry, request)
    return True


async def super_user(request: Request, background_tasks: BackgroundTasks) -> bool:
    if not request.user.is_authenticated:
        raise NotLoggedInException()
    if "super" not in request.auth:
        raise PermissionFailedException("You need to be marked as an admin user to access this endpoint")
    background_tasks.add_task(update_session_expiry, request)
    return True

from fastapi.responses import RedirectResponse, HTMLResponse
from fastapi import Request, FastAPI

"""
Authentication custom exceptions and exception handlers
"""


class NotLoggedInException(Exception):
    def __init__(self):
        pass


class AlreadyLoggedInException(Exception):
    def __init__(self):
        pass


def already_logged_in_handler(request: Request, exc: AlreadyLoggedInException):
    return RedirectResponse("/")


def not_logged_in_handler(request: Request, exc: NotLoggedInException):
    """Redirect to the login page if login is required"""
    return RedirectResponse("/auth")


class PermissionFailedException(Exception):
    def __init__(self, permissions: list):
        self.permissions = permissions


def permission_failed_handler(request: Request, exc: PermissionFailedException):
    """shows an error page if the users authentication scope fails to meet the requirements"""
    return HTMLResponse(content="templates/permission_failed.html", status_code=401)


def auth_exceptions_add(app: FastAPI):
    """Loads the exception handlers into the app"""
    app.add_exception_handler(NotLoggedInException, not_logged_in_handler)
    app.add_exception_handler(PermissionFailedException, permission_failed_handler)
    app.add_exception_handler(AlreadyLoggedInException, already_logged_in_handler)

So two things not mentioned yet. Firstly I created an auth_exceptions_add() function so the exceptions are managed outside of main.py, allowing some separation.

The second is the use of the BackgroundTask object in the dependencies. This is used so the Session record expiry date can be updated without slowing the view down.

Conclusion

I started this project assuming I could use some of the code elements from Starlette, but instead, I hit all the items that showed why FastAPI is not Starlette, and just because they are in Starlette, it does not mean they will work in FastAPI.

However, FastAPI makes, what I thought would be complex items, easy to create. Adding my own middleware, dependencies and handlers was in the end, while not simple, was relatively easy and quite a bit of fun. While FastAPI might not have the batteries that other frameworks have, it gives really easy paths to create those for yourself.