JWT Bearer Guard¶

jwt_bearer(...) — validates JSON Web Tokens with support for HS256/HS384/HS512, RS256/RS384/RS512, and ES256/ES384/ES512, including JWKS endpoint fetching with kid rotation.

Requires: pip install "lauren-guards[jwt]" (plus cryptography for RS/ES; httpx for jwks_url).

Signature¶

def jwt_bearer(
    *,
    secret: str | bytes | None = None,
    public_key: str | bytes | None = None,
    jwks_url: str | None = None,
    algorithms: Iterable[str] = ("HS256",),
    issuer: str | None = None,
    audience: str | Iterable[str] | None = None,
    leeway: int = 0,
    sub_claim: str = "sub",
    role_claim: str | Callable[[dict], Iterable[str]] | None = None,
    scope_claim: str | Callable[[dict], Iterable[str]] | None = "scope",
    jwks_cache_seconds: int = 300,
    realm: str = "lauren",
) -> type:
    ...

Provide exactly one of secret / public_key / jwks_url — otherwise ValueError is raised.

Parameters:

Parameter	Type	Default	Description
`secret`	`str \\| bytes \\| None`	`None`	Symmetric secret for HS256/HS384/HS512. Only HMAC algorithms are allowed with `secret`.
`public_key`	`str \\| bytes \\| None`	`None`	Public key (PEM string or bytes) for RS/ES verification.
`jwks_url`	`str \\| None`	`None`	Fetch the issuer's JWKS document, pick the key matching the token's `kid`, cache for `jwks_cache_seconds`.
`algorithms`	iterable	`("HS256",)`	Allowed algorithms. The JWT's `alg` header must be in this list — it is never trusted on its own.
`issuer`	`str \\| None`	`None`	Required `iss` claim when set.
`audience`	`str \\| Iterable[str] \\| None`	`None`	Required `aud` claim value(s).
`leeway`	`int`	`0`	Clock-skew tolerance (seconds) for `exp`/`nbf`/`iat`.
`sub_claim`	`str`	`"sub"`	Claim used for `AuthUser.id`.
`role_claim`	`str \\| callable \\| None`	`None`	Where to read roles; a string claim name or a callable `(claims) -> Iterable[str]` for nested paths (e.g. Keycloak's `realm_access.roles`).
`scope_claim`	`str \\| callable \\| None`	`"scope"`	Where to read scopes; OAuth standard space-separated string or list.
`jwks_cache_seconds`	`int`	`300`	TTL (seconds) for the cached JWKS document.
`realm`	`str`	`"lauren"`	Surfaced in error responses.

Symmetric (HS256)¶

import os
from lauren_guards import jwt_bearer

JwtGuard = jwt_bearer(
    secret=os.environ["JWT_SECRET"],
    algorithms=["HS256"],
    audience="my-api",
    issuer="https://auth.example.com",
)

Issue tokens with:

import jwt

token = jwt.encode(
    {"sub": "user-123", "aud": "my-api", "iss": "https://auth.example.com"},
    os.environ["JWT_SECRET"],
    algorithm="HS256",
)

Asymmetric (RS256 with JWKS)¶

JwtGuard = jwt_bearer(
    jwks_url="https://auth.example.com/.well-known/jwks.json",
    algorithms=["RS256"],
    audience="my-api",
    issuer="https://auth.example.com",
)

The JWKS document is fetched lazily on first use and cached for jwks_cache_seconds (default 300 s). If a token arrives with an unknown kid, the guard drops the cache and refetches once to handle in-flight key rotation.

Custom claims mapping¶

By default jwt_bearer maps sub → AuthUser.id, the scope claim → AuthUser.scopes, and (if role_claim is set) roles → AuthUser.roles. The full decoded payload is always available on AuthUser.claims, and credential_type is "jwt".

For non-standard layouts use the claim options — e.g. Auth0-style roles:

JwtGuard = jwt_bearer(
    secret="secret",
    algorithms=["HS256"],
    scope_claim=lambda c: c.get("https://auth.example.com/scopes", []),
    role_claim=lambda c: c.get("https://auth.example.com/roles", []),
)

or Keycloak nested roles:

role_claim=lambda c: c.get("realm_access", {}).get("roles", [])

Full example¶

import os

from lauren import LaurenFactory, controller, get, module, use_guards
from lauren_guards import jwt_bearer, require_scopes

JwtGuard = jwt_bearer(
    jwks_url=os.environ["JWKS_URL"],
    algorithms=["RS256"],
    audience=os.environ.get("JWT_AUDIENCE", "api"),
    issuer=os.environ.get("JWT_ISSUER"),
)

@use_guards(JwtGuard)
@controller("/api")
class ApiController:
    @get("/me")
    async def me(self, request) -> dict:
        user = request.state.get("user")
        return {"id": user.id, "scopes": list(user.scopes)}

    @get("/admin")
    @use_guards(require_scopes("admin"))
    async def admin(self) -> dict:
        return {"admin": True}

@module(controllers=[ApiController])
class AppModule:
    pass

app = LaurenFactory.create(AppModule)

Algorithm security¶

Always pin algorithms explicitly — never accept an algorithm from the token header alone. Choose the narrowest list your issuers actually use.