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:
    # ahash runs argon2 on a dedicated thread pool. Hashing a password
    # costs about 40ms of pure CPU, and every one of those milliseconds
    # would otherwise be a millisecond your event loop serves nobody.
    psw_hash = await security.hash_service.ahash(data.password)

    async with db.async_executor(model=User) as e:
        user = User(data.username, data.email, psw_hash)
        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 await security.hash_service.averify(user.psw_hash, password):
        return None
    return user

hash and verify still exist for synchronous code; inside an async handler, prefer the awaitable pair. The pool is sized by SECURITY_HASHER_THREADS, which also caps how many logins can hash at the same time — that is a feature, not an oversight, since argon2 is deliberately expensive.

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.

The guards build on each other, in one chain. Reading it from the bottom up is the fastest way to know what any single one of them actually enforces:

current_user            no enforcement; User or None
  |
require_user           401 NOT_AUTHENTICATED  (+ CSRF on unsafe methods)
  |
[email verified]       401 EMAIL_NOT_VERIFIED
  |
require_2fa            401 TWO_FA_REQUIRED    (only if the user HAS a second factor)
  |
require_admin          403 NOT_AUTHORIZED
require_roles / require_any_role
require_permissions / require_any_permission
requirement_or_grant / requirement_and_grant

Three things follow from that shape. A permission check also enforces authentication, CSRF, a verified address and — for a user who has one configured — a second factor verified in this session. require_2fa is a no-op for a user with no second factor at all; it gates the ones who have one. And bare require_user is the only guard that lets an unverified address through, which is exactly what makes a verification flow reachable: put your "please confirm your email" page behind require_user, and everything else behind anything above it.

Every guard also has a plain-function twin ending in _fn — require_admin_fn, require_roles_fn(roles) and so on. Those are the same resolvers without the FastAPI Depends wrapper, for the times you need to run a check somewhere that is not a route signature. And for a check on a user you already hold, the predicates are exported directly: has_2fa(user), await is_admin(user), await has_roles(user, roles) and their any variants.

Relationships do not load themselves

Here is the one thing to internalise about these models. Every relationship on User, Role and Permission is declared lazy="raise_on_sql", with exactly two exceptions: user.totp_secret and user.webauthn_credentials, which the 2FA gate reads on every gated request and which therefore have to arrive eagerly.

Everything else — user.roles, role.permissions, user.identities, user.backup_codes and every reverse side — raises the moment you touch it, instead of quietly firing a query. That is deliberate: the gates query the association tables directly and never navigate a relationship, so loading them for every authenticated request was pure waste — and role.users in particular meant hydrating every user holding a role on every admin request.

So when you do want them, say so:

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


async def roles_of(user_id: int) -> list[str]:
    async with db.async_executor(model=User) as e:
        result = await e.exec(
            select(User)
            .where(User.id == user_id)
            .options(selectinload(User.roles))
        )
        user = result.first()

    # Loaded explicitly, so it survives the closed session:
    return [role.name for role in user.roles]

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):
    ...

Single-use links

The same token service also mints signed one-time tokens for the flows every app eventually needs — confirm your email, reset your password. generate_token(data, salt) packs a payload into a signed string; validate_token(token, salt) unpacks it and, for any salt other than "csrf", burns it: the token is recorded in an ExpiredToken table so a second attempt is rejected even though the signature is still perfectly valid.

from webfluid.core.ext import security

svc = security.token_service

# Hand this out in a mail link:
token = svc.generate_token({"user_id": user.id}, salt="verify-email")

# On the receiving route. Raises 403 TOKEN_EXPIRED the second time
# around, and 403 INVALID_TOKEN if it was tampered with:
data = await svc.validate_token(token, salt="verify-email")

Spent tokens are swept from the table on a schedule once they are older than SECURITY_TOKEN_MAX_AGE, so the table stays small — provided EXT_SCHEDULING is on to run that job.

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. 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 User model carries email_verified and pending_email for the address-change dance.

Role carries one field the battery never reads itself: requires_2fa. It is there for the app on top — an admin panel or an accounts Additive can look up whether any of a user's roles demand a second factor and compose that with require_user and has_2fa into a gate of its own, the same way require_2fa composes require_user internally. The framework hands you the flag and the primitives to act on it; deciding what "requires 2FA" means for your roles is policy, and policy belongs to your app.

Letting machines in

One last pair, for when a route has to answer both a browser and a script. With EXT_JWT enabled, requirement_or_grant accepts either a session that satisfies the requirement or a bearer token whose permissions claim carries the named grant. requirement_and_grant demands both. The requirement is a small dictionary, so the same shape works for roles and permissions:

from webfluid.core.ext import security

svc = security.user_service


# A logged-in editor, or a token carrying the "posts:write" grant:
async def publish(user = svc.requirement_or_grant(
    {"requirement": "has_any_role", "roles": ["editor", "admin"]},
    "posts:write"
)): ...


# Simply "somebody is authenticated", or a valid grant:
async def ingest(user = svc.requirement_or_grant(
    {"requirement": "is_authenticated"}, "data:ingest"
)): ...

The token is minted by the JWTManager, so its sub claim resolves back to a real User and your handler receives the same object either way. An admin always passes the requirement side, whatever it says. If you only ever want the token identity, resolve_bearer and bearer_principal give it to you without any session logic at all.

The config values

• SECURITY_SECRET — signs CSRF and one-time tokens. Required in production; in debug mode a fixed development secret is used and warned about.
• 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. Both sides of the double-submit check read the name from here, so renaming the cookie is a one-line change.
• SECURITY_HASHER_TIME_COST / _MEMORY_COST / _PARALLELISM — argon2 tuning. Defaults 3 / 65536 / 4.
• SECURITY_HASHER_THREADS — workers in the pool ahash / averify run on. Default 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.