JWTManager

The last battery in the box is the JWTManager. It signs and verifies JSON Web Tokens, but with a small luxury on top: it rotates its signing secret on a schedule and keeps the recent secrets in the cache, so old tokens stay valid until they age out on their own.

Enabling JWT

This is the first battery that builds on two others: it needs both EXT_SCHEDULING (to run the rotation) and EXT_CACHE (to store the secrets). The manager will refuse to start otherwise:

[general]
SECRET_KEY = supersecret

[data]
REDIS_URI = redis://localhost:6379

[extensions]
EXT_SCHEDULING = 1
EXT_CACHE = 1
EXT_JWT = 1

Encoding and decoding

The API is four methods: a sync and async pair for each direction. You hand encode a payload and get a signed token back; decode verifies it and returns the claims. The manager fills in the standard exp, iat, nbf, iss and aud claims for you:

from webfluid.core.ext import jwt


async def issue_token(user_id: int) -> str:
    return await jwt.aencode({"sub": str(user_id)})


async def read_token(token: str) -> dict:
    # Raises if the token is invalid, expired or not yet valid.
    return await jwt.adecode(token)

Plugged into routes, this becomes a bearer-token flow for API clients — separate from the session login we built in Security. It gets its own module in fluid/api, minting a token and reading it back on a protected endpoint:

from fastapi import Request
from fastapi.exceptions import HTTPException

from fluid.services.auth import issue_token, read_token


async def token(request: Request):
    data = await request.json()
    # ... verify credentials here ...
    signed = await issue_token(user_id=1)
    return {"token": signed}


async def whoami(request: Request):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing token")

    try:
        claims = await read_token(auth.removeprefix("Bearer "))
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid token")

    return {"user_id": claims["sub"]}

That is the manual version, and it is worth writing once to see the shape. In practice you rarely need it: if the security battery is enabled too, put a permissions list into the payload and its requirement_or_grant guard from the Security chapter does the header parsing, the decoding and the user lookup for you — on the same route that also serves a browser session.

Short-lived and revoked tokens

Two knobs turn that into something you can actually operate. encode takes an expire in days, overriding JWT_EXPIRY_DAYS for a single token — a thirty day session token and a ten minute one-off can come out of the same manager. And if you put a jti claim into your payload, decode checks the cache for a revocation entry under that id before it hands the claims back:

from webfluid.core.ext import cache, jwt
from uuid import uuid4


async def issue_short_lived(user_id: int) -> tuple[str, str]:
    jti = uuid4().hex
    token = await jwt.aencode({"sub": str(user_id), "jti": jti}, expire=1)
    return token, jti


async def revoke(jti: str):
    # Any key is fine; decode only checks whether it exists.
    await cache.aset(f"jwt:revoked:{jti}", "1", timeout=60 * 60 * 24)

Revocation is opt-in on purpose: a token without a jti costs no cache lookup at all, which is the whole point of a stateless token. Add one where you need a kill switch, and leave it off everywhere else.

Audiences and tuning

Both encode and decode take an audience argument (default "default"). Audiences let one app mint tokens for different consumers — an internal API versus a public one, say — and you map the short names to their full audience strings in the config:

from webfluid.core.config import register_config


@register_config(10)
class MyConfig:
    SESSION_COOKIE_SECURE = True

    JWT_ROTARY_INTERVAL = 15            # days between secret rotations
    JWT_SECRET_LENGTH = 128             # bytes of randomness per secret
    JWT_EXPIRY_DAYS = 30                # token lifetime
    JWT_ALGORITHM = "HS256"
    JWT_ISSUER = "MyApp"
    JWT_AUDIENCES = {
        "default": "Application",
        "api": "PublicAPI"
    }

That rounds out the battery set. With the database, migrations, mail, i18n, events, cache and tokens all in place, our app has grown into something real. Time to give it a face.

Continue reading

From here you can continue straight with the Frontend introduction.