Middleware: deferred (async) authentication with replace_event + redirect_event

The real reason MiddlewareControl.replace_event(...) exists: gate a request behind an asynchronous authentication check the middleware cannot perform inline. The token has to be validated against another component (an external auth service, a DB) — you can't block the Broker on that round-trip, and for a WebSocket message there is no synchronous 401 to return anyway.

The trick: when a token isn't known-good, the middleware replaces the incoming event with an Authenticate event that carries the original inside it. A dedicated auth component validates the token and answers. On its response the middleware either redirects the wrapped original onward (authenticated) or replaces it with a 401 (rejected). One pattern, identical for REST and WS.

client ──(AuthedEvent + token)──▶ Broker / request middleware
                                        │  token unknown?
                                        ▼  replace_event
                                  Authenticate(event=original, token=…)
                                        ▼
                                  AuthManager → AuthWorker (calls auth service)
                                        ▼  response: authenticated=True/False
                              Broker / response middleware
                   authenticated ──redirect_event(original)──▶ Router ─▶ real handler
                   rejected ──────replace_event(original+401)─▶ client

---

The three mechanics it relies on

• register_events() — subscribe the Broker to the Authenticate response so it routes back through the middleware and process_response can act on the auth result. Without it that response has no subscriber at all (see Things to watch below).

• replace_event(Authenticate(event=original, …)) — swap the user's event for the auth event and pause the real work.

• redirect_event(original) — once authenticated, hand the original event to the Router to be handled normally. Because it is the same object the client sent, its correlation (_api) is intact — the eventual response goes back to the right caller automatically.

redirect_event on a response turns it back into a request (puts it on the Router queue); replace_event on a response sends it straight back to the client. That asymmetry is exactly what "allow" vs "reject" need.

---

Events

# events/auth.py
from dataclasses import field

from edri.api.dataclass import Header
from edri.dataclass.event import Event, event
from edri.dataclass.response import Response, response


# Base for anything that needs auth. Over REST the token comes from a header
# (a list[str]); over WebSocket the client puts it in the message (a str).
@event
class AuthenticatedEvent(Event):
    token: list[str] | str = field(default=Header("authorization"), kw_only=True)


@response
class AuthenticateResponse(Response):
    authenticated: bool
    message: str | None = None


# Wraps the original event while its token is being verified.
@event
class Authenticate(Event):
    event: Event
    token: str
    response: AuthenticateResponse

A concrete protected endpoint just inherits the base:

# events/ping.py
from http import HTTPMethod

from edri.api.dataclass.api_event import api, ApiType
from edri.dataclass.response import Response, response

from events.auth import AuthenticatedEvent


@response
class PingResponse(Response):
    message: str


@api(url="/ping", method=HTTPMethod.GET, resource="ping")
class AuthedPing(AuthenticatedEvent):
    response: PingResponse

---

The middleware

# middlewares/auth.py
from edri.api.middleware import Middleware, MiddlewareControl
from edri.dataclass.directive.base import UnauthorizedResponseDirective
from edri.dataclass.event import Event
from edri.dataclass.response import ResponseStatus

from events.auth import AuthenticatedEvent, Authenticate


class AuthMiddleware(Middleware):
    def __init__(self) -> None:
        super().__init__()
        self._ok: set[str] = set()      # tokens already verified (cache; add TTL in real use)

    @staticmethod
    def register_events():
        # Route the Authenticate *response* back through process_response.
        return [], [Authenticate]

    def process_request(self, event: Event, info) -> MiddlewareControl:
        if not isinstance(event, AuthenticatedEvent):
            return MiddlewareControl.continue_processing()

        token = event.token[0] if isinstance(event.token, list) else event.token
        if token in self._ok:
            return MiddlewareControl.continue_processing()        # fast path

        # Unknown token -> defer: hand off to the auth component, pausing the real work.
        return MiddlewareControl.replace_event(Authenticate(event=event, token=token))

    def process_response(self, event: Event, info) -> MiddlewareControl:
        if not isinstance(event, Authenticate):
            return MiddlewareControl.continue_processing()

        if event.response.authenticated:
            self._ok.add(event.token)
            return MiddlewareControl.redirect_event(event.event)  # release the original onward

        event.event.response.set_status(ResponseStatus.FAILED)
        event.event.response.add_directive(UnauthorizedResponseDirective("Invalid token."))
        return MiddlewareControl.replace_event(event.event)       # 401 back to the client

---

The auth component

A normal manager + worker. The worker does the slow part (here a call to an external auth service) off the manager thread — which is the whole reason the check had to be deferred.

# managers/auth.py
from edri.abstract import ManagerBase, request

from events.auth import Authenticate
from workers.auth import AuthWorker


class AuthManager(ManagerBase):
    @request
    def authenticate(self, event: Authenticate) -> None:
        self.workers.start(event, AuthWorker)

# workers/auth.py
import requests

from edri.abstract.worker import WorkerThread

from events.auth import Authenticate


class AuthWorker(WorkerThread[Authenticate]):
    def do(self) -> None:
        try:
            resp = requests.get(
                "https://auth.example/valid_token",
                headers={"Authorization": f"Bearer {self.event.token}"},
                timeout=5,
            )
            resp.raise_for_status()
        except Exception:
            self.event.response.authenticated = False
            self.event.response.message = "authentication failed"
        else:
            self.event.response.authenticated = True
        self.event_send(self.event)

The real handler is plain — it never sees a token and assumes it is reached only when authenticated:

# managers/ping.py
from edri.abstract import ManagerBase, request

from events.ping import AuthedPing


class PingManager(ManagerBase):
    @request
    def ping(self, event: AuthedPing) -> None:
        event.response.message = "pong"

---

Entry point & calling it

# main.py
from edri import EDRI

from managers.auth import AuthManager
from managers.ping import PingManager
from middlewares.auth import AuthMiddleware

if __name__ == "__main__":
    app = EDRI()
    app.add_component(AuthManager())
    app.add_component(PingManager())
    app.start_api(middlewares=[AuthMiddleware()])
    app.run()

curl http://localhost:8080/ping -H "Accept: application/json" -H "Authorization: secret"
# {"data": {"message": "pong"}, "status": "OK"}     (verified, then cached)

curl -o /dev/null -w '%{http_code}\n' http://localhost:8080/ping \
     -H "Accept: application/json" -H "Authorization: nope"
# 401

Over WebSocket nothing changes in the middleware: the client sends the token as a field on the message instead of a header (token is a str, not a list[str]), the same replace/redirect dance gates the message, and the real handler runs only after the token checks out.

---

Things to watch

• register_events is mandatory here. It is what subscribes the Broker to the Authenticate response. Without it that response has no subscriber: the Router drops it as unrouted, process_response never runs, and the wrapped original is never released. The auth answer never reaches the client — every client-send path is gated on the event being registered — so the caller simply gets no response: 500 after API_RESPONSE_TIMEOUT over REST/HTML, an open connection that never replies over WebSocket.

• Correlation is free on redirect. redirect_event(event.event) reuses the original object, so _api (the client correlation) is preserved. If you build a fresh replacement instead, copy _api across yourself.

• Cache with a TTL. The self._ok set here never expires; in production key it by token with an expiry (the field can hold a datetime and you re-verify once stale), so a revoked token stops working.

• Branch REST vs WS if needed. event._api.type tells you the transport (ApiType.REST / HTML / WS) when the two paths must differ (e.g. multiple Authorization headers are a REST-only concern).