# How to Implement Custom User Authentication Flows in FastAPI Boilerplate

> Implement custom user authentication flows in FastAPI Boilerplate. Extend the JWT system with custom claims or MFA without breaking endpoints. Learn how to modify security.py for your needs.

- Repository: [Benav Labs/fastapi-boilerplate](https://github.com/benavlabs/fastapi-boilerplate)
- Tags: how-to-guide
- Published: 2026-02-26

---

**The benavlabs/fastapi-boilerplate repository provides a complete JWT-based authentication system built around OAuth2PasswordBearer that you can extend by modifying [`security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/security.py), adding custom claims, or implementing MFA flows without breaking existing endpoints.**

This guide walks through the architecture of the authentication layer in the **benavlabs/fastapi-boilerplate** repository and demonstrates how to implement custom user authentication flows—from simple role-based claims to multi-factor authentication—while maintaining compatibility with the existing FastAPI structure.

## Understanding the Default Authentication Architecture

The boilerplate ships with a stateless JWT implementation that handles password hashing, token creation, and revocation through a centralized security module.

### Core Security Components in [`security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/security.py)

The file [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py) serves as the central security utility. It contains:

- **`verify_password`** – Uses **bcrypt** to compare plain passwords against stored hashes
- **`authenticate_user`** – Queries the database via `crud_users.get` by username or email
- **`create_access_token`** and **`create_refresh_token`** – Encode JWT payloads with expiration claims (`exp`) and `token_type` distinctions
- **`verify_token`** – Decodes tokens, validates `token_type`, and checks the **token_blacklist** table
- **`blacklist_token`** – Revokes tokens by inserting them into the blacklist

### Login and Logout Endpoints

The authentication endpoints reside in [`src/app/api/v1/login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/login.py) and [`src/app/api/v1/logout.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/logout.py):

- **`/login`** – Accepts username/password, calls `authenticate_user`, returns the access token in the JSON body, and sets the refresh token as an **HttpOnly** cookie
- **`/logout`** – Receives the access token via the `Authorization` header (handled by `oauth2_scheme`) and the refresh token from the cookie, then calls `blacklist_token` for both

### Data Layer Components

The user data flow relies on two key files:

- **[`src/app/models/user.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/models/user.py)** – SQLAlchemy model defining fields like `hashed_password`, `is_superuser`, and timestamps
- **[`src/app/crud/crud_users.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/crud/crud_users.py)** – FastCRUD wrapper providing database operations used by `authenticate_user`

## Extending the Default Flow for Custom Requirements

Because the architecture isolates security logic in pure functions, you can inject custom behavior without modifying FastAPI core dependencies.

### Adding Custom JWT Claims for Role-Based Access

To include `roles` or `permissions` in the token payload, modify `create_access_token` in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py):

```python

# src/app/core/security.py

from typing import List, Any
from datetime import timedelta

async def create_access_token(
    data: dict[str, Any],
    expires_delta: timedelta | None = None,
    roles: List[str] | None = None,
) -> str:
    to_encode = data.copy()
    if roles:
        to_encode["roles"] = roles  # Inject custom claim

    
    # Existing expiration logic

    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire, "token_type": "access"})
    
    return jwt.encode(to_encode, SECRET_KEY.get_secret_value(), algorithm=ALGORITHM)

```

Then update the login endpoint in [`src/app/api/v1/login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/login.py) to pass role data:

```python

# src/app/api/v1/login.py

user_roles = ["admin"] if user["is_superuser"] else ["user"]
access_token = await create_access_token(
    data={"sub": user["username"]},
    expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    roles=user_roles,
)

```

Finally, extend `TokenData` in [`src/app/core/schemas.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/schemas.py) to include the new field for validation.

### Enforcing Email Verification Before Login

To require verified emails, first add the column to the user model in [`src/app/models/user.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/models/user.py):

```python

# src/app/models/user.py

is_verified: Mapped[bool] = mapped_column(default=False, index=True)

```

Run the Alembic migration to update the database schema.

Then modify `authenticate_user` in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py) to check verification status:

```python

# src/app/core/security.py

async def authenticate_user(username_or_email: str, password: str, db: AsyncSession):
    # Existing lookup logic via crud_users.get

    db_user = await crud_users.get(db=db, username=username_or_email) or \
              await crud_users.get(db=db, email=username_or_email)
    
    if not db_user:
        return False
    
    # New verification guard

    if getattr(db_user, "is_verified", False) is False:
        raise UnauthorizedException("Account not verified. Please check your email.")
    
    # Existing password verification

    if not verify_password(password, db_user.hashed_password):
        return False
    
    return db_user

```

### Implementing Multi-Factor Authentication (MFA)

For MFA, create a new endpoint at [`src/app/api/v1/mfa.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/mfa.py) that validates a temporary token and OTP before issuing the final access token:

```python

# src/app/api/v1/mfa.py

from fastapi import APIRouter, Body, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(tags=["mfa"])

@router.post("/mfa")
async def verify_mfa(
    token: str = Body(..., embed=True),  # Temporary MFA token

    otp: str = Body(..., embed=True),
    db: AsyncSession = Depends(async_get_db),
):
    # 1. Validate temporary token

    mfa_data = await verify_token(token, TokenType.ACCESS, db)
    if not mfa_data:
        raise UnauthorizedException("Invalid MFA token.")
    
    # 2. Verify OTP (implementation-specific, e.g., TOTP)

    if not await check_otp(mfa_data.username_or_email, otp):
        raise UnauthorizedException("Invalid OTP.")
    
    # 3. Issue final access token

    final_token = await create_access_token(
        data={"sub": mfa_data.username_or_email},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": final_token, "token_type": "bearer"}

```

Modify the standard `/login` endpoint to issue a temporary MFA token when `user.mfa_enabled` is `True`, instead of returning the final access token immediately.

### Implementing Refresh Token Rotation

To enhance security by rotating refresh tokens on every use, extend the `/refresh` endpoint in [`src/app/api/v1/login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/login.py):

```python

# src/app/api/v1/login.py

@router.post("/refresh")
async def refresh_access_token(
    request: Request,
    response: Response,
    db: AsyncSession = Depends(async_get_db),
):
    refresh_token = request.cookies.get("refresh_token")
    if not refresh_token:
        raise UnauthorizedException("Refresh token missing.")
    
    user_data = await verify_token(refresh_token, TokenType.REFRESH, db)
    if not user_data:
        raise UnauthorizedException("Invalid refresh token.")
    
    # Rotate: blacklist the old refresh token

    await blacklist_token(refresh_token, db)
    
    # Issue new token pair

    new_access = await create_access_token(data={"sub": user_data.username_or_email})
    new_refresh = await create_refresh_token(data={"sub": user_data.username_or_email})
    
    # Set new refresh token cookie

    max_age = settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60
    response.set_cookie(
        key="refresh_token",
        value=new_refresh,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=max_age,
    )
    
    return {"access_token": new_access, "token_type": "bearer"}

```

## Key Files Reference

| File | Purpose | Location |
|------|---------|----------|
| [`security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/security.py) | JWT creation, verification, password hashing, blacklist handling | [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py) |
| [`login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/login.py) | `/login` and `/refresh` endpoints | [`src/app/api/v1/login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/login.py) |
| [`logout.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/logout.py) | `/logout` endpoint and token revocation | [`src/app/api/v1/logout.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/logout.py) |
| [`crud_users.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/crud_users.py) | FastCRUD operations for User model | [`src/app/crud/crud_users.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/crud/crud_users.py) |
| [`user.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/user.py) | SQLAlchemy User model definition | [`src/app/models/user.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/models/user.py) |
| [`schemas.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/schemas.py) | Pydantic models for tokens and validation | [`src/app/core/schemas.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/schemas.py) |

## Summary

- **The benavlabs/fastapi-boilerplate** provides a complete JWT authentication system using **OAuth2PasswordBearer** with access tokens, refresh tokens, and a database-backed blacklist for logout functionality.
- **All security logic is centralized** in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py), making it straightforward to extend claims, add verification checks, or swap hashing algorithms without modifying endpoint logic.
- **Custom flows** such as email verification, MFA, and refresh token rotation are implemented by extending the `authenticate_user` function, creating new endpoints like [`mfa.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/mfa.py), and modifying the token creation logic while reusing existing `verify_token` and `blacklist_token` utilities.

## Frequently Asked Questions

### How do I add custom claims to the JWT payload?

Extend the `create_access_token` function in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py) to accept additional parameters (like `roles: List[str]`) and inject them into the `to_encode` dictionary before calling `jwt.encode`. Update the `TokenData` schema in [`src/app/core/schemas.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/schemas.py) to include the new fields for proper validation when verifying tokens.

### Can I replace bcrypt with Argon2 for password hashing?

Yes. Modify the `verify_password` and `get_password_hash` functions in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py). Replace the existing bcrypt calls with Argon2 implementations (using the `argon2-cffi` library) while maintaining the same function signatures that accept `(plain_password, hashed_password)` and `(password)` respectively. The rest of the authentication flow remains unchanged.

### How do I implement refresh token rotation securely?

In the `/refresh` endpoint located in [`src/app/api/v1/login.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/login.py), call `blacklist_token` on the incoming refresh token immediately after verifying it with `verify_token`. Then generate a new access token and a new refresh token using `create_access_token` and `create_refresh_token`. Set the new refresh token as an HttpOnly cookie and return the new access token in the JSON response. This ensures each refresh token is used only once.

### Where should I add email verification checks?

Add the verification logic to the `authenticate_user` function in [`src/app/core/security.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/security.py) after retrieving the user from the database but before verifying the password. Check the `is_verified` attribute (which you must add to the `User` model in [`src/app/models/user.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/models/user.py) and migrate via Alembic). If `is_verified` is `False`, raise an `UnauthorizedException` with a message indicating the account requires verification.