Security

Our app can talk to its users now — but before it can have users at all, it needs to know who they are. The Security battery is WebFluid's answer to that: a user store with roles and permissions, argon2 password hashing, CSRF protection, two-factor auth and OAuth logins, all wrapped in ready-made FastAPI dependencies.

Enabling security

The battery keeps its users, roles and tokens in the database, so it builds on the SQLAlchemy extension — both switches have to be on. It also signs its CSRF and one-time tokens with a dedicated secret. In production a missing SECURITY_SECRET is a hard error; in debug mode it falls back to a fixed development secret and warns you about it.

[general]
SECRET_KEY = supersecret

[data]
DATABASE_URI = sqlite:///app.db

[extensions]
EXT_SQLALCHEMY = 1
EXT_SECURITY = 1

[security]
SECURITY_SECRET = another-long-secret

The services

Everything is reached through the shared registry in webfluid.core.ext. The security instance exposes four services:

• security.user_service — the current user and every route guard.
• security.hash_service — argon2 password hashing.
• security.token_service — CSRF and signed one-time tokens.
• security.oauth_service — social logins through authlib.

Registering and authenticating users

There is no one-size-fits-all signup endpoint — you own that flow. The battery gives you the pieces: the User model, the hasher and a pair of validators. The validators are built to plug into Pydantic, so the natural home for them is a schema — and that earns our project its fluid/schemas package, sitting right next to the models and services we've already grown:

from pydantic import BaseModel, EmailStr, field_validator

from webfluid.extensions.security.utils import (
    validate_username, validate_password
)


class CreateUser(BaseModel):
    username: str
    email: EmailStr
    password: str

    # validate_username / validate_password raise on bad input, so they
    # drop straight in as field validators (validate_password even
    # attaches an .errors list with the individual reasons).
    _validate_username = field_validator("username")(validate_username)
    _validate_password = field_validator("password")(validate_password)

With the input validated, a small service turns the schema into a stored user, hashing the password on the way in:

from webfluid.core.ext import db, security
from webfluid.extensions.security.models import User
from sqlalchemy import select

from fluid.schemas.accounts import CreateUser


async def register(data: CreateUser) -> User:
    async with db.async_executor(model=User) as e:
        user = User(
            data.username, data.email,
            security.hash_service.hash(data.password)
        )
        await e.insert(user, flush=True)
        return user


async def authenticate(username: str, password: str) -> User | None:
    async with db.async_executor(model=User) as e:
        result = await e.exec(select(User).where(User.username == username))
        user = result.first()

    if not user or not user.psw_hash:
        return None
    if not security.hash_service.verify(user.psw_hash, password):
        return None
    return user

Logging in and out

The current user is derived from request.session["user_id"], so logging in is simply storing that id and logging out is dropping it. The session cookie is already signed by the framework, so there is nothing else to wire up. These are JSON endpoints, so they join the api_router in fluid/api next to the routes from the Mail chapter:

from fastapi import Request
from fastapi.exceptions import HTTPException

from fluid.services.accounts import authenticate


async def login(request: Request):
    data = await request.json()
    user = await authenticate(data["username"], data["password"])
    if not user:
        raise HTTPException(status_code=401, detail="INVALID_CREDENTIALS")

    request.session["user_id"] = user.id
    return {"id": user.id, "username": user.username}


async def logout(request: Request):
    request.session.pop("user_id", None)
    return {"status": "ok"}

Knowing the current user

The user service hands you FastAPI dependencies. The plain current_user yields the logged-in User or None; require_user enforces a session (and validates CSRF on unsafe methods) and raises 401 otherwise:

from webfluid.core.ext import security
from webfluid.extensions.security.models import User


async def me(user: User = security.user_service.require_user):
    return {"id": user.id, "username": user.username}


async def maybe(user: User | None = security.user_service.current_user):
    return {"authenticated": user is not None}

Roles and permissions

Users carry Roles, roles carry Permissions and an is_admin flag. The user service turns all of that into dependencies you drop straight into your routes:

from webfluid.core.ext import security

svc = security.user_service

# Attributes (no call): a fixed guard.
async def dashboard(user = svc.require_admin): ...

# Methods (called): parametrised guards.
async def write_posts(user = svc.require_permissions(["posts:write"])): ...
async def staff_area(user = svc.require_any_role(["editor", "moderator"])): ...

The full set is require_user, require_2fa, require_admin, require_roles / require_any_role and require_permissions / require_any_permission. The _any_ variants pass if the user matches at least one entry; the others require all of them.

CSRF protection

The token service ships a double-submit CSRF guard. require_user already runs it on unsafe methods, but you can also apply it on its own. Hand the browser a token with csrf_response (it sets the cookie and the session token) and echo it back in the X-CSRF-Token header on your next request:

from fastapi import Request

from webfluid.core.ext import security


# GET this once to receive the csrf cookie + session token.
async def csrf(request: Request):
    return security.token_service.csrf_response(request)


# Protect a standalone route (require_user does this for you already).
async def submit(request: Request, _=security.token_service.csrf_protect):
    ...

Two-factor, backup codes and OAuth

The models also cover the harder parts of auth: TOTPSecret for authenticator apps, WebAuthnCredential for passkeys and BackupCode for one-time recovery codes. UserService.has_2fa(user) tells you whether a user has any second factor at all. Social logins run through security.oauth_service, which wraps authlib, is configured with SECURITY_OAUTH_CLIENTS and links each external login to a user through the Identity model.

The config values

• SECURITY_SECRET — signs CSRF and one-time tokens. Required in production.
• SECURITY_TOKEN_MAX_AGE — lifetime of signed tokens in seconds. Default 3600.
• SECURITY_CSRF_COOKIE_NAME / SECURITY_CSRF_COOKIE_SECURE — the CSRF cookie. Defaults csrf_token / True.
• SECURITY_HASHER_TIME_COST / _MEMORY_COST / _PARALLELISM — argon2 tuning. Defaults 3 / 65536 / 4.
• SECURITY_PASSWORD_MIN_LENGTH / SECURITY_PASSWORD_REQUIREMENTS — the policy the validators enforce. Defaults 8 and one each of lower, upper, digits and special.
• SECURITY_OAUTH_CLIENTS — a mapping of provider name to authlib client settings.
• SECURITY_MODELS_DB_BIND — put the security tables on a non-default bind.

Continue reading

From here you can continue straight with Events.