fastapi-otp-authentication 0.1.1

A flexible, type-safe library for adding OTP-based authentication to FastAPI applications

FastAPI OTP Authentication

A flexible, type-safe library for adding OTP (One-Time Password) based authentication to FastAPI applications with minimal configuration.

Features

• 🔐 Cryptographically Secure: Uses secrets module for OTP generation
• 🍪 HTTP-Only Cookies: Refresh tokens stored securely
• 🚫 Token Blacklisting: Revoke tokens on logout
• 💯 100% Type Safe: Full mypy strict mode compliance
• 🎯 Flexible: Extend abstract classes to customize behavior
• 🧑‍💻 Developer Mode: Testing mode with predictable OTP codes
• 🔧 Multiple Database Backends: SQLAlchemy (SQL) or MongoDB (NoSQL)
• 📝 Custom Claims: Add your own JWT claims
• ⚡ Production Ready: Secret validation and security best practices

Database Support

This library supports both SQL and NoSQL databases:

Backend	Adapter	Models	Use Case
SQLAlchemy	SQLAlchemyAdapter	BaseOTPUserTable (ORM)	PostgreSQL, MySQL, SQLite, etc.
MongoDB	MongoDBAdapter	BaseOTPUserDocument (Pydantic)	MongoDB Atlas, local MongoDB

Choose the backend that fits your stack. Both implementations provide the same OTP authentication features.

Installation

# Install base package
uv add git+https://github.com/gronnmann/fastapi-otp-authentication.git

# Install with SQLAlchemy support
uv add "fastapi-otp-authentication[sqlalchemy] @ git+https://github.com/gronnmann/fastapi-otp-authentication.git"

# Install with MongoDB support
uv add "fastapi-otp-authentication[mongodb] @ git+https://github.com/gronnmann/fastapi-otp-authentication.git"

# Install both (if needed)
uv add "fastapi-otp-authentication[all] @ git+https://github.com/gronnmann/fastapi-otp-authentication.git"

# Then install your database driver
uv add asyncpg      # PostgreSQL with SQLAlchemy
uv add aiomysql     # MySQL with SQLAlchemy
uv add aiosqlite    # SQLite with SQLAlchemy
# MongoDB driver (motor) is included with [mongodb] extra

Quick Start (SQLAlchemy)

1. Create Your User Model and Register Blacklist table

from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from fastapi_otp_authentication import BaseOTPUserTable
from fastapi_otp_authentication import TokenBlacklist


class Base(DeclarativeBase):
    pass

class User(BaseOTPUserTable[int], Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

class Blacklist(TokenBlacklist, Base):
    pass

2. Configure OTP Authentication

You configure the library by extending the OTPAuthConfig abstract class:

from datetime import timedelta
from fastapi_otp_authentication import OTPAuthConfig

class MyOTPConfig(OTPAuthConfig):
    # Security - REQUIRED
    secret_key = "your-secret-key-generate-with-openssl-rand-hex-32"
    
    # Optional: Developer mode for testing (default: False)
    developer_mode = False
    
    cookie_secure = True  # Use secure cookies (https) in production
    
    # Token lifetimes
    access_token_lifetime = timedelta(hours=1)
    refresh_token_lifetime = timedelta(days=7)
    
    # OTP settings
    otp_length = 6
    otp_expiry = timedelta(minutes=10)
    max_otp_attempts = 5
    
    async def send_otp(self, email: str, code: str) -> None:
        """Implement your OTP delivery method."""
        # Send via email, SMS, etc.
        print(f"OTP for {email}: {code}")
    
    def get_additional_claims(self, user: User) -> dict[str, Any]:
        """Add custom claims to JWT tokens."""
        return {"username": user.username}

3. Set Up Database and Dependencies

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from fastapi_otp_authentication import SQLAlchemyAdapter

# Database setup
engine = create_async_engine("sqlite+aiosqlite:///./app.db")
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)

async def get_async_session():
    async with async_session_maker() as session:
        yield session

async def get_otp_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyAdapter(session, User, Blacklist)

4. Register Authentication Router

from fastapi import FastAPI
from fastapi_otp_authentication import get_auth_router

app = FastAPI()
config = MyOTPConfig()

auth_router = get_auth_router(get_otp_db, config)
app.include_router(auth_router, prefix="/auth", tags=["auth"])

5. Protect Your Routes

from fastapi import Depends
from fastapi_otp_authentication import (
    get_current_user_dependency,
    get_verified_user_dependency,
    get_custom_claims_dependency,
)

# Create dependencies
current_user = Depends(get_current_user_dependency(get_otp_db, config))
verified_user = Depends(get_verified_user_dependency(get_otp_db, config))
custom_claims = Depends(get_custom_claims_dependency(config))

@app.get("/protected")
async def protected_route(user: User = current_user):
    return {"user_id": user.id, "email": user.email}

@app.get("/verified-only")
async def verified_only(user: User = verified_user):
    return {"message": "Access granted to verified user"}

@app.get("/check-claims")
async def check_claims(claims: dict = custom_claims):
    return {"custom_claims": claims}

Quick Start (MongoDB)

1. Create Your User Model

from pydantic import Field
from fastapi_otp_authentication import BaseOTPUserDocument, TokenBlacklistDocument

class User(BaseOTPUserDocument):
    """User model with custom fields."""
    username: str = Field(..., max_length=50)
    full_name: str | None = Field(None, max_length=100)

# Blacklist uses default TokenBlacklistDocument
class Blacklist(TokenBlacklistDocument):
    pass

2. Configure OTP Authentication

Same as SQLAlchemy - extend OTPAuthConfig:

from datetime import timedelta
from fastapi_otp_authentication import OTPAuthConfig

class MyOTPConfig(OTPAuthConfig):
    secret_key = "your-secret-key-generate-with-openssl-rand-hex-32"
    developer_mode = False
    
    access_token_lifetime = timedelta(hours=1)
    refresh_token_lifetime = timedelta(days=7)
    
    otp_length = 6
    otp_expiry = timedelta(minutes=10)
    max_otp_attempts = 5
    
    async def send_otp(self, email: str, code: str) -> None:
        print(f"OTP for {email}: {code}")
    
    def get_additional_claims(self, user: User) -> dict[str, Any]:
        return {"username": user.username}

3. Set Up MongoDB and Dependencies

from motor.motor_asyncio import AsyncIOMotorClient
from fastapi_otp_authentication import MongoDBAdapter

# MongoDB setup
client = AsyncIOMotorClient("mongodb://localhost:27017")
database = client["myapp"]

async def get_otp_db():
    return MongoDBAdapter(
        database=database,
        user_collection_name="users",
        blacklist_collection_name="token_blacklist",
        user_model_class=User,
    )

4. Create Indexes (Important!)

MongoDB requires indexes for optimal performance:

@app.on_event("startup")
async def startup():
    # Unique index on email
    await database["users"].create_index("email", unique=True)
    
    # Index on jti for blacklist lookups
    await database["token_blacklist"].create_index("jti", unique=True)
    
    # TTL index to auto-delete expired tokens
    await database["token_blacklist"].create_index(
        "expires_at", expireAfterSeconds=0
    )

5. Register Router & Protect Routes

Same as SQLAlchemy:

from fastapi_otp_authentication import get_auth_router, get_current_user_dependency

config = MyOTPConfig()
auth_router = get_auth_router(get_otp_db, config)
app.include_router(auth_router, prefix="/auth", tags=["auth"])

current_user = Depends(get_current_user_dependency(get_otp_db, config))

@app.get("/protected")
async def protected_route(user: User = current_user):
    return {"user_id": str(user.id), "email": user.email}

API Endpoints

The auth router provides these endpoints:

POST /auth/request-otp

Request OTP code to be sent to user's email.

{
  "email": "[e-mail]"
}

POST /auth/verify-otp

Verify OTP code and receive authentication tokens.

{
  "email": "[e-mail]",
  "code": "123456"
}

Response:

{
  "access_token": "eyJ...",
  "token_type": "bearer"
}

The refresh token is set in an HTTP-only cookie.

POST /auth/refresh

Refresh access token using refresh token from cookie.

POST /auth/logout

Blacklist tokens and clear refresh cookie.

Security Best Practices

Generate Secure Secret Key

openssl rand -hex 32

The library validates that your secret key is at least 32 characters long (unless in developer mode).

Developer Mode

For testing, enable developer mode:

class MyOTPConfig(OTPAuthConfig):
    developer_mode = True
    secret_key = "any-key-allowed-in-dev-mode"

In developer mode:

• OTP codes are always 000000 (or length of zeros)
• Secret key validation is relaxed

⚠️ Never use developer mode in production!

Advanced Usage

Custom Claims

Add custom data to JWT tokens:

class MyOTPConfig(OTPAuthConfig):
    def get_additional_claims(self, user: User) -> dict[str, Any]:
        return {
            "role": user.role,
            "permissions": user.permissions,
            "organization_id": user.organization_id,
        }

Access custom claims in your routes:

from fastapi_otp_authentication import get_custom_claims_dependency

custom_claims = Depends(get_custom_claims_dependency(config))

@app.get("/admin-check")
async def admin_check(claims: dict = custom_claims):
    if claims.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return {"message": "Admin access granted"}

Rate Limiting

OTP verification attempts are tracked automatically. Configure max attempts:

class MyOTPConfig(OTPAuthConfig):
    max_otp_attempts = 3  # Stricter rate limiting

Custom OTP Length

class MyOTPConfig(OTPAuthConfig):
    otp_length = 8  # 8-digit OTP codes

Token Blacklist Cleanup

Periodically clean up expired blacklisted tokens:

SQLAlchemy:

async def cleanup_expired_tokens():
    async with async_session_maker() as session:
        db = SQLAlchemyAdapter(session, User, Blacklist)
        removed = await db.cleanup_blacklist()
        print(f"Removed {removed} expired tokens")

MongoDB:

async def cleanup_expired_tokens():
    db = MongoDBAdapter(database, "users", "token_blacklist", User)
    removed = await db.cleanup_blacklist()
    print(f"Removed {removed} expired tokens")

Note: MongoDB can automatically clean up expired tokens using TTL indexes (see MongoDB Quick Start).

Example Applications

SQLAlchemy Example

See examples/basic_usage.py for a complete SQLite example.

uv run python examples/basic_usage.py

MongoDB Example

See examples/mongodb_usage.py for a complete MongoDB example.

Prerequisites: MongoDB must be running locally or update the connection string.

# Start MongoDB (if using Docker)
docker run -d -p 27017:27017 mongo

# Run the example
uv run python examples/mongodb_usage.py

Then visit http://localhost:8000/docs to try the API.

Testing

Run the test suite:

uv run pytest tests/ -v

Contributing

Contributions are welcome! Please ensure:

1. All tests pass: pytest tests/
2. Code is type-safe: mypy . --strict
3. Code is linted: ruff check .
4. Code is formatted: black .