Extensions

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:

app_configs/app.ini ini 
[general]
SECRET_KEY = supersecret

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

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

On startup the manager mints its first secret and registers a job that rotates it every JWT_ROTARY_INTERVAL days. Each secret lives in the cache under its own key id, so a token signed with an older secret still verifies until that secret expires.

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:

fluid/services/auth.py python 
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 a route, issuing a token on login and reading it back on a protected endpoint looks like this:

fluid/app/routes.py python 
from fastapi import Request
from fastapi.exceptions import HTTPException

from fluid.services.auth import issue_token, read_token


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


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"]}

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:

fluid/config.py python 
from webfluid.core.config import register_config


@register_config(10)
class Config:
    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.