How to Implement Custom User Authentication Flows in FastAPI Boilerplate

The benavlabs/fastapi-boilerplate repository provides a complete JWT-based authentication system built around OAuth2PasswordBearer that you can extend by modifying 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

The file 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 and 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 – SQLAlchemy model defining fields like hashed_password, is_superuser, and timestamps
• 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:

# 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 to pass role data:

# 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 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:

# 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 to check verification status:

# 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 that validates a temporary token and OTP before issuing the final access token:

# 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:

# 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	JWT creation, verification, password hashing, blacklist handling	src/app/core/security.py
login.py	/login and /refresh endpoints	src/app/api/v1/login.py
logout.py	/logout endpoint and token revocation	src/app/api/v1/logout.py
crud_users.py	FastCRUD operations for User model	src/app/crud/crud_users.py
user.py	SQLAlchemy User model definition	src/app/models/user.py
schemas.py	Pydantic models for tokens and validation	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, 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, 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 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 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. 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, 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 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 and migrate via Alembic). If is_verified is False, raise an UnauthorizedException with a message indicating the account requires verification.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}