Custom Guards

A guard is a small class with a single async method, can_activate(ctx), that returns True to allow a request to proceed and False (or raises an HTTPError) to reject it. Guards are how Lauren models authorization — and they're decoupled from authentication, so you can compose them freely.

The minimum viable guard

python

from lauren import ExecutionContext, injectable, Scope

@injectable(scope=Scope.SINGLETON)
class AdminGuard:
    async def can_activate(self, ctx: ExecutionContext) -> bool:
        return ctx.request.headers.get("x-role") == "admin"

That's the entire surface. Attach it to a route or controller and Lauren runs can_activate on every matching request.

Attaching guards

python

from lauren import use_guards, controller, get

# Per route — runs only on this handler:
@controller("/x")
class X:
    @get("/admin")
    @injectable(scope=Scope.SINGLETON)
    @use_guards(AdminGuard)
    async def admin_only(self): ...

# Per controller — runs on every handler in the class:
@use_guards(AuthenticatedGuard)
@controller("/private")
class P: ...

# Both — class-level guards run FIRST, then route-level:
@use_guards(AuthenticatedGuard)
@controller("/mixed")
class M:
    @get("/admin")
    @use_guards(AdminGuard)        # runs after AuthenticatedGuard
    async def admin(self): ...

Both decoration orders work for class-level guards:

python

@use_guards(AuthGuard)        # outer
@controller("/x")
class A: ...

@controller("/x")
@use_guards(AuthGuard)        # outer
class B: ...

What `can_activate` receives

ExecutionContext exposes:

Attribute	Type	Purpose
`ctx.request`	`Request`	Full request — headers, cookies, state, body.
`ctx.handler`	callable	The handler method about to be invoked.
`ctx.controller_class`	`type \\| None`	The controller class, if any.
`ctx.get_metadata(key, default)`	`Any`	Read per-route metadata set with `@set_metadata`.

The get_metadata hook is what lets a single guard implement parametric policies — read on.

Guards and DI

Guards are classes, and Lauren auto-marks them as injectables. They can take constructor dependencies just like any other service:

python

from lauren import injectable

@openapi_security({"BearerAuth": []})
@injectable(scope=Scope.SINGLETON)
class TokenGuard:
    def __init__(self, jwt: JwtService, log: Logger) -> None:
        self.jwt = jwt
        self.log = log

    async def can_activate(self, ctx: ExecutionContext) -> bool:
        token = ctx.request.headers.get("authorization", "")
        if not token.startswith("Bearer "):
            return False
        try:
            claims = self.jwt.decode(token[7:])
        except InvalidToken:
            self.log.warn(f"invalid token from {ctx.request.client.host}")
            return False
        ctx.request.state.set("user_id", claims["sub"])
        return True

Two important properties:

The guard is constructed once per scope like any other injectable. By default that means once per request (since guards are auto-marked as request-scoped if they have request-scoped deps).
ctx.request.state is fair game — guards routinely enrich the request (setting user_id, tenant, roles) so that downstream extractors and handlers can use the parsed values without reparsing.

Allowing vs denying

There are three ways for a guard to block a request:

python

@injectable(scope=Scope.SINGLETON)
class G:
    async def can_activate(self, ctx) -> bool:
        # 1. Return False — Lauren raises ForbiddenError(403), or the
        #    guard's deny_error (see below).
        if not ok: return False

        # 2. Raise UnauthorizedError(401) — typical for missing/invalid auth tokens.
        if not authn: raise UnauthorizedError("invalid token", challenge='Bearer realm="api"')

        # 3. Raise any HTTPError — full control over status + body.
        if banned: raise HTTPError("account banned", status_code=403,
                                   code="account_banned",
                                   detail={"reason": "TOS violation"})
        return True

Choose by intent:

return False — generic 403. Good for "the user just isn't allowed here".
raise UnauthorizedError(...) — 401. Good for missing/invalid credentials; challenge= sets the WWW-Authenticate header RFC 9110 requires on a 401.
raise <custom HTTPError> — when the response body needs domain-specific shape.

Choosing the denial error

Authentication guards ("who are you?") should answer 401, authorization guards ("you may not") 403. Instead of raising, a guard can declare which error its False produces with a deny_error class attribute:

python

from typing import Any, ClassVar

from lauren import HTTPError, LaurenFactory, UnauthorizedError, controller, get, module, use_guards
from lauren.testing import TestClient
from lauren.types import ExecutionContext


class ApiKeyGuard:
    deny_error: ClassVar[type[HTTPError]] = UnauthorizedError  # default: ForbiddenError

    async def can_activate(self, ctx: ExecutionContext) -> bool:
        return ctx.request.headers.get("x-api-key") == "secret"


class BasicAuthGuard:
    @staticmethod
    def deny_error() -> HTTPError:  # a zero-argument factory: full control
        return UnauthorizedError("login required", challenge='Basic realm="admin"')

    async def can_activate(self, ctx: ExecutionContext) -> bool:
        return ctx.request.headers.get("authorization") == "Basic YWRtaW46cHc="


@controller("/reports")
class Reports:
    @get("/")
    @use_guards(ApiKeyGuard)
    async def index(self) -> dict[str, Any]:
        return {"ok": True}

    @get("/admin")
    @use_guards(BasicAuthGuard)
    async def admin(self) -> dict[str, Any]:
        return {"ok": True}


@module(controllers=[Reports])
class ReportsModule:
    pass


client = TestClient(LaurenFactory.create(ReportsModule))
denied = client.get("/reports/")
assert denied.status_code == 401
assert denied.json()["error"] == {
    "code": "unauthorized",
    "message": "guard ApiKeyGuard denied the request",
    "detail": {"guard": "ApiKeyGuard"},
}
assert client.get("/reports/admin").header("www-authenticate") == 'Basic realm="admin"'

A class (HTTPError subclass) is called as deny_error("guard <Name> denied the request", detail={"guard": "<Name>"}).
A zero-argument factory (a @staticmethod, a lambda, any callable) is called on every denial and must return an HTTPError instance — use it for challenge headers or a custom message.
deny_error is read once by LaurenFactory.create. Anything else — a class that is not an HTTPError, an exception instance, a callable that needs arguments — raises GuardConfigError (detail["reason"] == "invalid_deny_error") at startup.
It is plain class configuration, so a guard subclass inherits it (the guard role still is not inherited — re-decorate subclasses as usual).
WebSocket guards honour it too: a per-message denial answers with its error frame, and a connect-time denial closes with 1008 and the error's code (e.g. "unauthorized") as the close reason.

Parametric guards with `@set_metadata`

Instead of writing one guard per role / scope / permission, write one guard and parametrize it via @set_metadata:

python

from lauren import set_metadata

@injectable(scope=Scope.SINGLETON)
class RoleGuard:
    async def can_activate(self, ctx: ExecutionContext) -> bool:
        required = ctx.get_metadata("required_role", "user")
        actual = ctx.request.state.get("user", {}).get("role")
        return actual == required

@controller("/admin")
class AdminController:
    @get("/purge")
    @use_guards(RoleGuard)
    @set_metadata("required_role", "admin")
    async def purge(self): ...

    @get("/super-purge")
    @use_guards(RoleGuard)
    @set_metadata("required_role", "superadmin")
    async def super_purge(self): ...

This is the standard pattern in NestJS-influenced codebases and works just as well in Lauren.

A scopes-and-roles policy guard might look like:

python

@injectable(scope=Scope.SINGLETON)
class PolicyGuard:
    async def can_activate(self, ctx) -> bool:
        scopes = set(ctx.get_metadata("required_scopes", []))
        user_scopes = set(ctx.request.state.get("scopes", []))
        if not scopes.issubset(user_scopes):
            return False
        return True

@get("/billing/export")
@use_guards(PolicyGuard)
@set_metadata("required_scopes", ["billing.read", "exports.create"])
async def export(self): ...

Composing guards

Class-level and route-level guards run in order: class first, then route. All guards must pass; if any returns False or raises, the chain stops.

python

@use_guards(AuthenticatedGuard)
@controller("/orders")
class OrdersController:
    @get("/")
    async def list(self): ...                              # AuthenticatedGuard

    @post("/")
    @use_guards(WriteScopeGuard)
    async def create(self): ...                             # AuthenticatedGuard, WriteScopeGuard

    @delete("/{id}")
    @use_guards(WriteScopeGuard, AdminGuard)
    async def delete(self): ...                             # AuthenticatedGuard, WriteScopeGuard, AdminGuard

You can also pass multiple guards to a single decorator:

python

@use_guards(WriteScopeGuard, AdminGuard)
async def h(self): ...

Guards, extractors, and middleware — who does what?

A common confusion: when do I write a guard, when an extractor, when middleware?

Concern	Use a...
"Should this request even run?" — yes/no decision	Guard
"Decode this domain value from the request" — typed parsing	Extractor
"Wrap every request with cross-cutting behavior" — tracing, request IDs, response headers	Middleware
"Translate this error class into a response"	Exception handler

Guards are a predicate over the request. Extractors are a function on the request. Middleware is an interceptor around the entire dispatch. Keep these clean and your codebase reads beautifully.

Inheritance

Like every other class-level decorator, @use_guards attaches to the exact target only. A subclass that wants the parent's class-level guards must re-declare them. See Class Inheritance Rules.

python

@use_guards(AuthGuard)
@controller("/private")
class Parent: ...

@controller("/v2")
class Child(Parent):
    pass
# → Child has parent's handlers but NOT AuthGuard. This is intentional — security
# decisions should never silently inherit. Re-declare:

@use_guards(AuthGuard)
@controller("/v2")
class ChildOK(Parent): ...

Testing guards

Easiest path: drive guards through the TestClient.

python

from lauren.testing import TestClient

def test_admin_required():
    c = TestClient(app)
    r = c.get("/admin/purge")
    assert r.status_code == 403

    r = c.get("/admin/purge", headers={"x-role": "admin"})
    assert r.status_code == 200

For unit tests of complex guards, build an ExecutionContext manually or mock the request.

Guards on WebSocket gateways

@use_guards on a @ws_controller class runs the guard chain before the @on_connect hook and before the connection is accepted — a rejected client receives WebSocket close code 1008 (policy violation), with the guard's deny_error code ("forbidden" by default) as the close reason, without the handshake ever completing. On an @on_message method, the guard runs for every frame routed to that handler instead, receives a WsMessageContext (the connection context plus event and the validated payload), and a denial answers with a forbidden error frame (or its deny_error) while the connection stays open. See WebSockets → Per-message guards and interceptors.

python

from lauren import (
    injectable, Scope, use_guards,
    ws_controller, on_connect, WebSocket,
    WsConnectionContext,
)

@injectable(scope=Scope.SINGLETON)
class WsApiKeyGuard:
    async def can_activate(self, ctx: WsConnectionContext) -> bool:
        # ctx.request mirrors the HTTP upgrade request (headers, path, …)
        return ctx.request.headers.get("x-api-key") == "secret"

@use_guards(WsApiKeyGuard)
@ws_controller("/live")
class LiveGateway:
    @on_connect
    async def connected(self, ws: WebSocket) -> None:
        # only reached when WsApiKeyGuard.can_activate returned True
        await ws.send_json({"event": "welcome"})

`WsConnectionContext` vs `ExecutionContext`

Guards receive a WsConnectionContext instead of an ExecutionContext. The two types are duck-compatible on the fields guards most commonly read:

Field	HTTP (`ExecutionContext`)	WebSocket (`WsConnectionContext`)
`ctx.request.headers`	✓	✓
`ctx.request.path`	✓	✓
`ctx.request.path_params`	✓	✓
`ctx.request.method`	✓	always `"GET"`
`ctx.handler_class`	✓	✓
`ctx.route_template`	✓	✓
`ctx.get_metadata(key)`	✓	✓
`ctx.connection`	—	✓ (the live `WebSocket`)

A guard that only reads ctx.request.headers works unchanged on both HTTP controllers and WebSocket gateways:

python

@injectable(scope=Scope.SINGLETON)
class ApiKeyGuard:
    async def can_activate(self, ctx) -> bool:
        # Same guard, works for @controller AND @ws_controller
        return ctx.request.headers.get("x-api-key") == "secret"

Global WebSocket guards

Pass global_ws_guards to LaurenFactory.create() to apply guards to every WebSocket gateway in the application:

python

app = LaurenFactory.create(
    AppModule,
    global_ws_guards=[WsApiKeyGuard],
)

Global guards run before class-level guards, matching the ordering for HTTP global guards.

Reading WS guard metadata at runtime

lauren.reflect exposes helpers for extension packages that need to inspect a gateway's guard/interceptor metadata programmatically:

python

from lauren.reflect import reflect_guards, reflect_interceptors

guards = reflect_guards(MyGateway)       # tuple[type, ...] — own __dict__ only
interceptors = reflect_interceptors(MyGateway)

See Reference → Reflect for the full API.

Best practices

One guard, one concern. A guard that does authentication AND authorization AND rate-limiting is hard to compose. Split it.
Enrich request.state. Guards routinely parse a token and store decoded claims in request.state. Extractors and handlers downstream then consume the parsed values cheaply.
Use @set_metadata to parametrize. One RoleGuard is better than ten role-specific guards.
Order matters. Authentication guards always run before authorization guards. Place authentication at the controller level; authorization at the route level.
Don't build responses inside guards. Raise an HTTPError and let the exception handler / middleware shape the response.

Errors raised at startup

Error	Meaning
`GuardConfigError`	A guard class is missing `can_activate(ctx)`, or declares an invalid `deny_error`.
`MetadataInheritanceError`	A subclass of a guard was registered without re-decorating.
`MissingProviderError`	A guard's `__init__` requires a provider that isn't visible from the controller's module.

Custom Guards

The minimum viable guard

Attaching guards

What can_activate receives

Guards and DI

Allowing vs denying

Choosing the denial error

Parametric guards with @set_metadata

Composing guards

Guards, extractors, and middleware — who does what?

Inheritance

Testing guards

Guards on WebSocket gateways

WsConnectionContext vs ExecutionContext

Global WebSocket guards

Reading WS guard metadata at runtime

Best practices

Errors raised at startup

See also

What `can_activate` receives

Parametric guards with `@set_metadata`

`WsConnectionContext` vs `ExecutionContext`