tempestweb.server

O host do Modo B: create_app monta o app FastAPI com as rotas WebSocket e SSE, SecurityConfig liga autenticação, origem e limites de carga, e o RedisSessionRouter dispensa sticky session no SSE. É o que você importa no server.py de um deploy.

Guia com exemplos: Deploy em produção · Segurança (Modo B).

tempestweb.server

tempestweb.server — Mode B FastAPI host (WebSocket + SSE) and WebPush.

Re-exports the server factory/class and the WebPush service (P3). See docs/plan.md (Trilhos B e P) and docs/contract.md for the wire format carried over both transports.

TempestWebServer

Bases: Generic[S]

Holds the app definition and the live SSE session registry.

S is the application state type.

Attributes:

Name	Type	Description
api	FastAPI	The FastAPI application instance with the routes mounted.

Source code in tempestweb/server/app.py

class TempestWebServer(Generic[S]):
    """Holds the app definition and the live SSE session registry.

    ``S`` is the application state type.

    Attributes:
        api: The FastAPI application instance with the routes mounted.
    """

    def __init__(
        self,
        state_factory: Callable[[], S],
        view: Callable[[App[S]], Widget],
        *,
        title: str = "tempestweb",
        security: SecurityConfig | None = None,
        metrics: bool = False,
        observability: ServerObservability | None = None,
        sse_backend: SessionRouter | None = None,
        concurrent_dispatch: bool = False,
        theme: Theme | None = None,
    ) -> None:
        """Build the server and register the WebSocket and SSE routes.

        Args:
            state_factory: Builds a fresh state per connection (isolation).
            view: The shared ``view`` function rendered for each session.
            title: OpenAPI title for the FastAPI app.
            security: Opt-in auth + origin controls (Track S). ``None`` leaves
                the host open (dev) and logs a warning saying so — an open host
                accepts a WebSocket from any origin, which CORS does not guard.
            metrics: When ``True``, mount ``GET /metrics`` (Prometheus text) with
                connection counters (Track S — S8). Latency and throughput join
                them when ``observability`` carries a ``PatchMetrics``.
            observability: Patch latency, structured session logs and tracing
                (Track S — S8). ``None`` keeps the inert default: nothing is
                collected, no exporter is imported, and the hot path pays two
                attribute lookups. Pass
                ``ServerObservability(metrics=PatchMetrics(), logger=…,
                tracer=otel_tracer())`` for as much of it as you want.
            sse_backend: Router for SSE inbound events (Track S — S4). ``None``
                uses the in-process router (needs sticky sessions across
                instances); a :class:`RedisSessionRouter` drops that requirement.
            theme: The palette every component resolves its colors against,
                forwarded to each session's ``App``. ``None`` keeps the
                Material baseline.
            concurrent_dispatch: Run each event's handler as its own task instead
                of awaiting it before the next event is read. Events for the same
                widget key keep their arrival order; handlers for different keys
                overlap, so one slow handler no longer freezes the connection.
                Off by default — see :class:`~tempestweb.runtime.AppSession`.
        """
        if security is None:
            _LOGGER.warning(
                "tempestweb: serving without a SecurityConfig — no auth, no origin "
                "allowlist (so any site can open a WebSocket to this host), and no "
                "connection or message limits. Intended for local development; pass "
                "security=SecurityConfig(...) before exposing the host."
            )
        self._router: SessionRouter = sse_backend or InProcessRouter()
        self._state_factory: Callable[[], S] = state_factory
        self._view: Callable[[App[S]], Widget] = view
        self._sse_sessions: dict[str, _SSESession[S]] = {}
        self._concurrent_dispatch: bool = concurrent_dispatch
        self._theme: Theme | None = theme
        self._security: SecurityConfig = security or SecurityConfig()
        self._live: int = 0  # concurrent live sessions (S2 cap)
        self._metrics_enabled: bool = metrics
        self._observability: ServerObservability = (
            observability if observability is not None else ServerObservability()
        )
        self._opened: int = 0  # total sessions ever accepted
        self._rejected: int = 0  # total connections refused (auth/origin/cap)
        rpm = self._security.max_connections_per_minute
        self._rate: RateLimiter | None = RateLimiter(rpm) if rpm else None
        epm = self._security.max_events_per_minute
        self._event_rate: RateLimiter | None = RateLimiter(epm) if epm else None
        self.api: FastAPI = FastAPI(title=title)
        self._install_cors()
        self._install_security_headers()
        self._register_routes()

    def _prometheus(self) -> str:
        """Render the connection counters as Prometheus text (S8)."""
        cap = self._security.max_connections
        lines = [
            "# HELP tempestweb_sessions_live Currently connected sessions.",
            "# TYPE tempestweb_sessions_live gauge",
            f"tempestweb_sessions_live {self._live}",
            "# HELP tempestweb_sessions_opened_total Sessions accepted since start.",
            "# TYPE tempestweb_sessions_opened_total counter",
            f"tempestweb_sessions_opened_total {self._opened}",
            "# HELP tempestweb_connections_rejected_total Connections refused.",
            "# TYPE tempestweb_connections_rejected_total counter",
            f"tempestweb_connections_rejected_total {self._rejected}",
        ]
        if cap is not None:
            lines += [
                "# HELP tempestweb_sessions_max Configured max concurrent sessions.",
                "# TYPE tempestweb_sessions_max gauge",
                f"tempestweb_sessions_max {cap}",
            ]
        return "\n".join(lines) + "\n" + self._observability.prometheus()

    def _install_security_headers(self) -> None:
        """Add hardening response headers to every HTTP response (S6)."""
        if not self._security.wants_headers:
            return
        headers = self._security.header_values()

        @self.api.middleware("http")
        async def _headers(request: Request, call_next: Any) -> Response:  # noqa: ANN401
            """Add the configured hardening headers to every HTTP response.

            Uses ``setdefault`` so a route that deliberately sets its own value
            keeps it — the middleware fills gaps rather than overriding.

            Args:
                request: The incoming request, passed through untouched.
                call_next: The rest of the middleware chain.

            Returns:
                The downstream response, with the missing headers added.
            """
            response: Response = await call_next(request)
            for name, value in headers.items():
                response.headers.setdefault(name, value)
            return response

    def _at_capacity(self) -> bool:
        """Whether the concurrent-session cap is reached (S2)."""
        cap = self._security.max_connections
        return cap is not None and self._live >= cap

    def _rate_ok(self, credentials: Credentials) -> bool:
        """Whether the client IP is within the per-minute connection rate (S2)."""
        if self._rate is None:
            return True
        return self._rate.allow(credentials.client_ip or "unknown")

    def _event_rate_ok(self, credentials: Credentials) -> bool:
        """Whether the client IP is within the per-minute inbound-envelope rate (S2).

        Counts every envelope the client sends on either leg — an SSE ``POST`` or
        a WebSocket frame — against ``max_events_per_minute``. Distinct from
        :meth:`_rate_ok`, which budgets *connections*.

        Args:
            credentials: The credentials extracted from the request or upgrade.

        Returns:
            ``True`` when the envelope is within budget (or no limit is set).
        """
        if self._event_rate is None:
            return True
        return self._event_rate.allow(credentials.client_ip or "unknown")

    def _install_cors(self) -> None:
        """Install CORS for the HTTP/SSE surface when an allowlist is set (S1)."""
        origins = self._security.allowed_origins
        if origins is None:
            return
        from starlette.middleware.cors import CORSMiddleware

        self.api.add_middleware(
            CORSMiddleware,
            allow_origins=origins,
            allow_methods=["GET", "POST"],
            allow_headers=["*"],
            allow_credentials=not self._security.origins_wildcard,
        )

    def _new_session(
        self, transport: PatchTransport, session_id: str | None = None
    ) -> AppSession[S]:
        """Create an isolated session bound to a transport.

        Args:
            transport: The per-connection transport (WS or SSE).
            session_id: The id metrics, logs and traces share for this
                connection. ``None`` lets the session derive one.

        Returns:
            A fresh :class:`AppSession` for this connection.
        """
        return AppSession(
            self._state_factory,
            self._view,
            transport,
            concurrent_dispatch=self._concurrent_dispatch,
            theme=self._theme,
            observability=self._observability,
            session_id=session_id,
        )

    def _register_routes(self) -> None:
        """Mount the ``/health``, ``/ws``, ``/sse`` and ``/sse/{id}`` routes."""

        @self.api.get("/health")
        async def health() -> dict[str, Any]:
            """Liveness/readiness probe (S4): unauthenticated, cheap, no session.

            Returns ``ok`` plus the live session count and, when a cap is set,
            whether the host still has capacity — for load-balancer draining.
            """
            cap = self._security.max_connections
            return {
                "status": "ok",
                "sessions": self._live,
                "ready": cap is None or self._live < cap,
            }

        if self._metrics_enabled:

            @self.api.get("/metrics")
            async def metrics() -> Response:
                """Prometheus-format connection counters (S8)."""
                return Response(
                    content=self._prometheus(),
                    media_type="text/plain; version=0.0.4",
                )

        @self.api.websocket("/ws")
        async def ws_endpoint(websocket: WebSocket) -> None:
            """Serve one client over a WebSocket until it disconnects.

            The auth gate + origin allowlist (Track S) run on the upgrade before
            a session is created; a rejected connection is closed with ``1008``
            (policy violation) and never mounts.

            The connection is wrapped in one observability span and two log lines,
            carrying the same id the patch metrics carry — so "this client was
            slow" is a query, not an archaeology project.
            """
            peer = websocket.client.host if websocket.client else None
            credentials = _credentials_from_headers(
                websocket.headers,
                websocket.query_params,
                peer,
                self._security.trusted_proxies,
            )
            if not self._rate_ok(credentials):
                self._rejected += 1
                await websocket.close(code=1013)  # rate limited
                return
            if not await _authorize(self._security, credentials):
                self._rejected += 1
                await websocket.close(code=1008)
                return
            if self._at_capacity():
                self._rejected += 1
                await websocket.close(code=1013)  # try again later
                return
            await websocket.accept()
            self._live += 1
            self._opened += 1
            transport = WebSocketTransport(
                websocket,
                allow_inbound=lambda: self._event_rate_ok(credentials),
            )
            session = self._new_session(transport)
            try:
                with self._observability.session(session.session_id, transport="ws"):
                    await session.run()
            finally:
                self._live -= 1

        @self.api.get("/sse")
        async def sse_endpoint(request: Request, session: str) -> Response:
            """Open (or resume) the SSE patch stream for ``session``."""
            credentials = self._request_credentials(request)
            new_session = session not in self._sse_sessions
            if new_session and not self._rate_ok(credentials):
                self._rejected += 1
                return JSONResponse({"error": "rate limited"}, status_code=429)
            if not await _authorize(self._security, credentials):
                self._rejected += 1
                return JSONResponse({"error": "unauthorized"}, status_code=401)
            if new_session and self._at_capacity():
                self._rejected += 1
                return JSONResponse({"error": "at capacity"}, status_code=503)
            return await self._open_sse(request, session, credentials)

        @self.api.post("/sse/{session_id}")
        async def sse_post(session_id: str, request: Request) -> Response:
            """Receive one client envelope (event / native_result) for a session.

            The session id alone does not authorize the post: it must come from
            the same principal that opened the session, or the envelope would let
            a third party drive somebody else's screen.
            """
            credentials = self._request_credentials(request)
            if not await self._authorize_request(request):
                return JSONResponse({"error": "unauthorized"}, status_code=401)
            if not self._owns_sse(session_id, credentials):
                return JSONResponse({"error": "forbidden"}, status_code=403)
            if not self._event_rate_ok(credentials):
                self._rejected += 1
                return JSONResponse({"error": "rate limited"}, status_code=429)
            if self._declared_too_large(request):
                return JSONResponse({"error": "payload too large"}, status_code=413)
            body = await self._read_body(request)
            if body is None:
                return JSONResponse({"error": "payload too large"}, status_code=413)
            return await self._handle_sse_post(session_id, body)

    def _declared_too_large(self, request: Request) -> bool:
        """Whether the *declared* body size already exceeds ``max_message_bytes``.

        A cheap pre-check on ``Content-Length`` that rejects an oversized body
        before a single byte is read. It is not sufficient on its own — the
        header is optional under chunked transfer encoding — so
        :meth:`_read_body` enforces the same limit while reading (S2).

        Args:
            request: The incoming request.

        Returns:
            ``True`` when the declared length is over the limit.
        """
        limit = self._security.max_message_bytes
        if limit is None:
            return False
        raw = request.headers.get("content-length")
        try:
            return raw is not None and int(raw) > limit
        except ValueError:
            return False

    async def _read_body(self, request: Request) -> bytes | None:
        """Read the request body, aborting past ``max_message_bytes`` (S2).

        The body is consumed chunk by chunk and the running total is checked
        against the limit, so a client that omits ``Content-Length`` (legal under
        chunked transfer encoding) cannot stream an unbounded body into memory —
        the read stops at the first chunk that crosses the limit.

        Args:
            request: The incoming request.

        Returns:
            The body bytes, or ``None`` when the limit was exceeded.
        """
        limit = self._security.max_message_bytes
        if limit is None:
            return await request.body()
        chunks: list[bytes] = []
        total = 0
        async for chunk in request.stream():
            total += len(chunk)
            if total > limit:
                return None
            chunks.append(chunk)
        return b"".join(chunks)

    def _request_credentials(self, request: Request) -> Credentials:
        """Extract credentials (incl. client IP) from an HTTP request."""
        peer = request.client.host if request.client else None
        return _credentials_from_headers(
            request.headers,
            request.query_params,
            peer,
            self._security.trusted_proxies,
        )

    async def _authorize_request(self, request: Request) -> bool:
        """Run the Track-S auth gate for an HTTP (SSE) request."""
        return await _authorize(self._security, self._request_credentials(request))

    def _owns_sse(self, session_id: str, credentials: Credentials) -> bool:
        """Whether these credentials may act on the SSE session under this id.

        An id the server holds no session for is *not* refused here: the ``GET``
        that opens a stream is what materializes a session, and a ``POST`` for an
        unknown id is already answered ``404`` by the router. Only a live session
        with a different owner is refused.

        Args:
            session_id: The session id from the URL.
            credentials: The credentials extracted from the request.

        Returns:
            ``True`` when the session is unknown or owned by this principal.
        """
        sse = self._sse_sessions.get(session_id)
        if sse is None:
            return True
        return hmac.compare_digest(sse.owner, _session_fingerprint(credentials))

    async def _open_sse(
        self, request: Request, session_id: str, credentials: Credentials
    ) -> Response:
        """Open, resume, or take over an SSE session and return its stream.

        A session id the server already holds may only be resumed by the
        principal that opened it (``403`` otherwise) — the id travels in a URL,
        so on its own it authorizes nothing.

        Resuming is a **takeover**: the new stream becomes the session's owner of
        record, so the stream it replaced can no longer tear the session down when
        its own response finally unwinds. That race used to drop a session the
        client had just successfully reconnected to.

        When the client resumes past a gap the replay buffer has evicted, the
        missed ticks cannot be replayed and no later index-relative patch would
        apply to the tree the client still holds; the session pushes a full
        resync instead and the stream starts from it.

        Args:
            request: The incoming request (for ``Last-Event-ID``).
            session_id: The client-chosen stable session id.
            credentials: The credentials extracted from the request.

        Returns:
            A ``text/event-stream`` streaming response, or ``403`` when the id
            belongs to another principal.
        """
        sse = self._sse_sessions.get(session_id)
        if sse is not None and not self._owns_sse(session_id, credentials):
            self._rejected += 1
            return JSONResponse({"error": "forbidden"}, status_code=403)
        if sse is None:
            transport = SSETransport()
            app_session = self._new_session(transport)
            sse = _SSESession(
                transport=transport,
                session=app_session,
                owner=_session_fingerprint(credentials),
            )
            self._sse_sessions[session_id] = sse
            self._live += 1
            self._opened += 1
            # Route cross-instance inbound events (S4): no-op in-process, Redis
            # pub/sub when configured — so a POST on another instance is delivered.
            sse.teardown = await self._router.bind(session_id, transport)
            sse.task = asyncio.ensure_future(self._run_sse(session_id, app_session))

        last_event_id = _parse_last_event_id(request.headers.get("last-event-id"))
        if last_event_id is not None and sse.transport.missed_since(last_event_id):
            last_event_id = sse.transport.last_id
            await sse.session.resync()
        sse.stream_token += 1
        stream_token = sse.stream_token
        session = sse

        async def body() -> AsyncIterator[str]:
            """Stream SSE frames, releasing the session when this stream ends."""
            try:
                async for chunk in session.transport.stream(last_event_id):
                    yield chunk
            finally:
                await self._release_sse(session_id, stream_token)

        return StreamingResponse(
            body(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )

    async def _run_sse(self, session_id: str, session: AppSession[S]) -> None:
        """Drive an SSE-backed session's lifecycle.

        Args:
            session_id: The session id (for cleanup on exit).
            session: The session to mount and run.
        """
        try:
            await session.run()
        finally:
            await self._drop_sse(session_id)

    async def _handle_sse_post(self, session_id: str, body: bytes) -> Response:
        """Route one POSTed client envelope into its SSE session.

        The router feeds a local transport directly, or hands the envelope off to
        the instance holding the stream (Redis); an unroutable session is a
        ``404``.

        Args:
            session_id: The session id from the URL path.
            body: The raw request body, already size-checked, holding the wire
                envelope as JSON.

        Returns:
            ``204 No Content`` on success, ``400`` on malformed JSON, ``404`` if
            the session is unknown.
        """
        try:
            envelope: dict[str, Any] = json.loads(body)
        except (ValueError, UnicodeDecodeError):
            return JSONResponse({"error": "invalid JSON"}, status_code=400)
        sse = self._sse_sessions.get(session_id)
        local = sse.transport if sse is not None else None
        if not await self._router.deliver(session_id, envelope, local):
            return JSONResponse({"error": "unknown session"}, status_code=404)
        return Response(status_code=204)

    async def _release_sse(self, session_id: str, stream_token: int) -> None:
        """Drop an SSE session, unless a newer stream has taken it over.

        Args:
            session_id: The session id whose stream ended.
            stream_token: The token the ending stream was opened with.
        """
        sse = self._sse_sessions.get(session_id)
        if sse is not None and sse.stream_token != stream_token:
            return
        await self._drop_sse(session_id)

    async def _drop_sse(self, session_id: str) -> None:
        """Close and forget an SSE session.

        Args:
            session_id: The session id to tear down.
        """
        sse = self._sse_sessions.pop(session_id, None)
        if sse is not None:
            self._live -= 1
            if sse.teardown is not None:
                await sse.teardown()
            await sse.transport.close()
            await sse.session.close()

Credentials dataclass

The authentication material extracted from a connection.

Attributes:

Name	Type	Description
token	str | None	The bearer token — from Authorization: Bearer <t>, the ?token= query parameter, or None when absent.
origin	str | None	The request Origin header, or None.
headers	Mapping[str, str]	The request headers (lower-cased keys).
query	Mapping[str, str]	The request query parameters.

Source code in tempestweb/server/security.py

@dataclass(slots=True)
class Credentials:
    """The authentication material extracted from a connection.

    Attributes:
        token: The bearer token — from ``Authorization: Bearer <t>``, the
            ``?token=`` query parameter, or ``None`` when absent.
        origin: The request ``Origin`` header, or ``None``.
        headers: The request headers (lower-cased keys).
        query: The request query parameters.
    """

    token: str | None
    origin: str | None
    headers: Mapping[str, str]
    query: Mapping[str, str]
    client_ip: str | None = None

SecurityConfig dataclass

Opt-in security controls for the Mode B host.

Attributes:

Name	Type	Description
authenticate	Authenticate | None	Run on every WS upgrade / SSE request before a session is created; a falsy return or a raised error rejects the connection. None (default) leaves the host open — dev only.
trusted_proxies	list[str] | None	Peer addresses whose X-Forwarded-For header may be believed — the reverse proxies actually in front of this host, or ["*"] to trust every peer. None (default) ignores the header entirely and uses the socket's peer address, because a client can put anything in it: with the header trusted unconditionally, a flood needs only a fresh fake value per request to be counted as a fresh client, which defeats every per-IP limit below.
allowed_origins	list[str] | None	If set, the exact Origin values allowed to connect (installs CORS for HTTP/SSE and checks the WS upgrade). ["*"] allows any origin (CORS wildcard; the WS check is skipped).
max_connections	int | None	Cap on concurrent live sessions (WS + SSE combined). A connection over the cap is refused (WS close 1013; SSE 503). None = unbounded (S2).
max_message_bytes	int | None	Reject an SSE POST body larger than this many bytes with 413 (S2). None = unbounded.
max_connections_per_minute	int | None	Per-client-IP cap on new connections in a rolling 60s window; a flood is refused (WS 1013 / SSE 429). The address comes from the socket's peer, or from X-Forwarded-For when trusted_proxies says the header may be believed. None = no per-IP limit (S2). Pair with a reverse-proxy limiter for defense in depth.
max_events_per_minute	int | None	Per-client-IP cap on inbound envelopes in a rolling 60s window — clicks, input, native_result frames — counted across both legs: an SSE POST /sse/{id} over budget answers 429, and a WebSocket frame over budget closes the socket with 1013. Separate from max_connections_per_minute because the budgets differ by orders of magnitude: one connection per client, but one envelope per interaction. None = unbounded (S2). Size it above the busiest legitimate interaction rate of your app.
security_headers	bool	When True, add hardening response headers (X-Content-Type-Options, Referrer-Policy, X-Frame-Options) to every HTTP response (S6).
hsts	bool	When True (implies security_headers), also send Strict-Transport-Security — enable only behind HTTPS.
content_security_policy	str | None	An explicit Content-Security-Policy value to send when set (app-specific; the shell uses inline module scripts, so a strict CSP needs a nonce/hash you supply here).

Source code in tempestweb/server/security.py

@dataclass(slots=True)
class SecurityConfig:
    """Opt-in security controls for the Mode B host.

    Attributes:
        authenticate: Run on every WS upgrade / SSE request before a session is
            created; a falsy return or a raised error rejects the connection.
            ``None`` (default) leaves the host open — dev only.
        trusted_proxies: Peer addresses whose ``X-Forwarded-For`` header may be
            believed — the reverse proxies actually in front of this host, or
            ``["*"]`` to trust every peer. ``None`` (default) ignores the header
            entirely and uses the socket's peer address, because a client can put
            anything in it: with the header trusted unconditionally, a flood needs
            only a fresh fake value per request to be counted as a fresh client,
            which defeats every per-IP limit below.
        allowed_origins: If set, the exact ``Origin`` values allowed to connect
            (installs CORS for HTTP/SSE and checks the WS upgrade). ``["*"]``
            allows any origin (CORS wildcard; the WS check is skipped).
        max_connections: Cap on concurrent live sessions (WS + SSE combined). A
            connection over the cap is refused (WS close ``1013``; SSE ``503``).
            ``None`` = unbounded (S2).
        max_message_bytes: Reject an SSE ``POST`` body larger than this many
            bytes with ``413`` (S2). ``None`` = unbounded.
        max_connections_per_minute: Per-client-IP cap on new connections in a
            rolling 60s window; a flood is refused (WS ``1013`` / SSE ``429``).
            The address comes from the socket's peer, or from
            ``X-Forwarded-For`` when ``trusted_proxies`` says the header may be
            believed. ``None`` = no per-IP limit (S2). Pair with a reverse-proxy
            limiter for defense in depth.
        max_events_per_minute: Per-client-IP cap on *inbound envelopes* in a
            rolling 60s window — clicks, input, ``native_result`` frames — counted
            across both legs: an SSE ``POST /sse/{id}`` over budget answers
            ``429``, and a WebSocket frame over budget closes the socket with
            ``1013``. Separate from ``max_connections_per_minute`` because the
            budgets differ by orders of magnitude: one connection per client, but
            one envelope per interaction. ``None`` = unbounded (S2). Size it above
            the busiest legitimate interaction rate of your app.
        security_headers: When ``True``, add hardening response headers
            (``X-Content-Type-Options``, ``Referrer-Policy``, ``X-Frame-Options``)
            to every HTTP response (S6).
        hsts: When ``True`` (implies ``security_headers``), also send
            ``Strict-Transport-Security`` — enable only behind HTTPS.
        content_security_policy: An explicit ``Content-Security-Policy`` value to
            send when set (app-specific; the shell uses inline module scripts, so
            a strict CSP needs a nonce/hash you supply here).
    """

    authenticate: Authenticate | None = None
    trusted_proxies: list[str] | None = field(default=None)
    allowed_origins: list[str] | None = field(default=None)
    max_connections: int | None = None
    max_message_bytes: int | None = None
    max_connections_per_minute: int | None = None
    max_events_per_minute: int | None = None
    security_headers: bool = False
    hsts: bool = False
    content_security_policy: str | None = None

    @property
    def wants_headers(self) -> bool:
        """Whether any response-header hardening is enabled (S6)."""
        return (
            self.security_headers
            or self.hsts
            or self.content_security_policy is not None
        )

    def header_values(self) -> dict[str, str]:
        """The hardening response headers implied by this config (S6)."""
        headers: dict[str, str] = {}
        if self.security_headers or self.hsts:
            headers["X-Content-Type-Options"] = "nosniff"
            headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
            headers["X-Frame-Options"] = "DENY"
        if self.hsts:
            headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains"
        if self.content_security_policy is not None:
            headers["Content-Security-Policy"] = self.content_security_policy
        return headers

    @property
    def origins_wildcard(self) -> bool:
        """Whether the origin allowlist is the ``*`` wildcard."""
        return self.allowed_origins is not None and "*" in self.allowed_origins

    def origin_allowed(self, origin: str | None) -> bool:
        """Whether ``origin`` may connect under this config.

        Args:
            origin: The request ``Origin`` header value, or ``None``.

        Returns:
            ``True`` when no allowlist is configured, the allowlist is ``*``, or
            ``origin`` is explicitly listed.
        """
        if self.allowed_origins is None or self.origins_wildcard:
            return True
        return origin in self.allowed_origins

wants_headers property

wants_headers: bool

Whether any response-header hardening is enabled (S6).

origins_wildcard property

origins_wildcard: bool

Whether the origin allowlist is the * wildcard.

header_values

header_values() -> dict[str, str]

The hardening response headers implied by this config (S6).

Source code in tempestweb/server/security.py

def header_values(self) -> dict[str, str]:
    """The hardening response headers implied by this config (S6)."""
    headers: dict[str, str] = {}
    if self.security_headers or self.hsts:
        headers["X-Content-Type-Options"] = "nosniff"
        headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        headers["X-Frame-Options"] = "DENY"
    if self.hsts:
        headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains"
    if self.content_security_policy is not None:
        headers["Content-Security-Policy"] = self.content_security_policy
    return headers

origin_allowed

origin_allowed(origin: str | None) -> bool

Whether origin may connect under this config.

Parameters:

Name	Type	Description	Default
origin	str | None	The request Origin header value, or None.	required

Returns:

Type	Description
bool	True when no allowlist is configured, the allowlist is *, or
bool	origin is explicitly listed.

Source code in tempestweb/server/security.py

def origin_allowed(self, origin: str | None) -> bool:
    """Whether ``origin`` may connect under this config.

    Args:
        origin: The request ``Origin`` header value, or ``None``.

    Returns:
        ``True`` when no allowlist is configured, the allowlist is ``*``, or
        ``origin`` is explicitly listed.
    """
    if self.allowed_origins is None or self.origins_wildcard:
        return True
    return origin in self.allowed_origins

InProcessRouter

Single-instance router: an inbound event feeds the local transport only.

Source code in tempestweb/server/sessions.py

class InProcessRouter:
    """Single-instance router: an inbound event feeds the local transport only."""

    async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
        """No cross-instance delivery is needed in-process."""

        async def _teardown() -> None:
            """Release nothing: in-process binding acquired nothing to release.

            The no-op still has to exist so callers can treat every router the
            same way and always await a teardown.
            """
            return None

        return _teardown

    async def deliver(
        self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
    ) -> bool:
        """Feed the local transport; report ``False`` when the session is remote."""
        if local is None:
            return False
        local.feed_inbound(envelope)
        return True

bind async

bind(session_id: str, transport: SSETransport) -> Teardown

No cross-instance delivery is needed in-process.

Source code in tempestweb/server/sessions.py

async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
    """No cross-instance delivery is needed in-process."""

    async def _teardown() -> None:
        """Release nothing: in-process binding acquired nothing to release.

        The no-op still has to exist so callers can treat every router the
        same way and always await a teardown.
        """
        return None

    return _teardown

deliver async

deliver(session_id: str, envelope: dict[str, Any], local: SSETransport | None) -> bool

Feed the local transport; report False when the session is remote.

Source code in tempestweb/server/sessions.py

async def deliver(
    self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
) -> bool:
    """Feed the local transport; report ``False`` when the session is remote."""
    if local is None:
        return False
    local.feed_inbound(envelope)
    return True

RedisSessionRouter

Cross-instance SSE router over Redis pub/sub (drops the sticky need).

Each session maps to a channel <prefix><session_id>. The instance holding the stream subscribes and feeds its transport; a POST on any instance publishes to the channel (or feeds directly when the session is local).

Source code in tempestweb/server/sessions.py

class RedisSessionRouter:
    """Cross-instance SSE router over Redis pub/sub (drops the sticky need).

    Each session maps to a channel ``<prefix><session_id>``. The instance holding
    the stream subscribes and feeds its transport; a ``POST`` on any instance
    publishes to the channel (or feeds directly when the session is local).
    """

    def __init__(self, client: Any, *, prefix: str = "tw:sse:") -> None:  # noqa: ANN401 - a redis.asyncio client (duck-typed)
        """Initialize with a redis.asyncio-compatible client.

        Args:
            client: A client exposing ``publish(channel, data)`` and
                ``pubsub()`` (redis.asyncio, or a compatible fake in tests).
            prefix: The channel key prefix.
        """
        self._client: Any = client
        self._prefix: str = prefix

    @classmethod
    def from_url(cls, url: str, *, prefix: str = "tw:sse:") -> RedisSessionRouter:
        """Build a router from a Redis URL (requires the ``[cache]`` extra).

        Args:
            url: A ``redis://`` connection URL.
            prefix: The channel key prefix.

        Returns:
            A configured :class:`RedisSessionRouter`.

        Raises:
            RuntimeError: If ``redis`` is not installed.
        """
        try:
            import redis.asyncio as redis  # type: ignore[import-not-found]
        except ImportError as exc:  # pragma: no cover - exercised via the error path
            raise RuntimeError(
                "redis is required for RedisSessionRouter; install "
                'tempest-fastapi-sdk[cache] or "redis".'
            ) from exc
        return cls(redis.from_url(url, decode_responses=True), prefix=prefix)

    async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
        """Subscribe to the session's channel and feed the transport."""
        channel = self._prefix + session_id
        pubsub = self._client.pubsub()
        await pubsub.subscribe(channel)

        async def _reader() -> None:
            """Drain the session's channel into the local SSE transport, forever.

            Runs as a background task for the life of the binding. Subscription
            bookkeeping frames are skipped — only ``type == "message"`` carries
            an envelope. A payload that is not decodable JSON is dropped rather
            than killing the reader, since one malformed publish must not take
            the session's inbound leg down with it.
            """
            async for message in pubsub.listen():
                if message.get("type") != "message":
                    continue
                data = message.get("data")
                try:
                    transport.feed_inbound(json.loads(data))
                except (ValueError, TypeError):  # pragma: no cover - defensive
                    continue

        task = asyncio.ensure_future(_reader())

        async def _teardown() -> None:
            """Stop the reader and give the channel's pubsub connection back.

            The unsubscribe is wrapped so the connection is closed even when it
            fails — a dropped Redis link would otherwise leak the pubsub object
            for every session that ended after the outage. ``aclose`` is
            preferred when present, falling back to ``close`` for older redis
            clients.
            """
            task.cancel()
            try:
                await pubsub.unsubscribe(channel)
            finally:
                aclose = getattr(pubsub, "aclose", None) or pubsub.close
                await aclose()

        return _teardown

    async def deliver(
        self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
    ) -> bool:
        """Feed a local transport directly, else publish for the holding instance.

        ``PUBLISH`` answers with the number of subscribers that received the
        message, and that is the only evidence available that some instance still
        holds the session. Reporting success unconditionally (as this used to)
        turned every post for a session that had already ended into a silent
        ``204``: the client believed its click had been delivered and the event
        was gone.
        """
        if local is not None:
            local.feed_inbound(envelope)
            return True
        receivers = await self._client.publish(
            self._prefix + session_id, json.dumps(envelope)
        )
        return bool(receivers)

from_url classmethod

from_url(url: str, *, prefix: str = 'tw:sse:') -> RedisSessionRouter

Build a router from a Redis URL (requires the [cache] extra).

Parameters:

Name	Type	Description	Default
url	str	A redis:// connection URL.	required
prefix	str	The channel key prefix.	'tw:sse:'

Returns:

Type	Description
RedisSessionRouter	A configured :class:RedisSessionRouter.

Raises:

Type	Description
RuntimeError	If redis is not installed.

Source code in tempestweb/server/sessions.py

@classmethod
def from_url(cls, url: str, *, prefix: str = "tw:sse:") -> RedisSessionRouter:
    """Build a router from a Redis URL (requires the ``[cache]`` extra).

    Args:
        url: A ``redis://`` connection URL.
        prefix: The channel key prefix.

    Returns:
        A configured :class:`RedisSessionRouter`.

    Raises:
        RuntimeError: If ``redis`` is not installed.
    """
    try:
        import redis.asyncio as redis  # type: ignore[import-not-found]
    except ImportError as exc:  # pragma: no cover - exercised via the error path
        raise RuntimeError(
            "redis is required for RedisSessionRouter; install "
            'tempest-fastapi-sdk[cache] or "redis".'
        ) from exc
    return cls(redis.from_url(url, decode_responses=True), prefix=prefix)

bind async

bind(session_id: str, transport: SSETransport) -> Teardown

Subscribe to the session's channel and feed the transport.

Source code in tempestweb/server/sessions.py

async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
    """Subscribe to the session's channel and feed the transport."""
    channel = self._prefix + session_id
    pubsub = self._client.pubsub()
    await pubsub.subscribe(channel)

    async def _reader() -> None:
        """Drain the session's channel into the local SSE transport, forever.

        Runs as a background task for the life of the binding. Subscription
        bookkeeping frames are skipped — only ``type == "message"`` carries
        an envelope. A payload that is not decodable JSON is dropped rather
        than killing the reader, since one malformed publish must not take
        the session's inbound leg down with it.
        """
        async for message in pubsub.listen():
            if message.get("type") != "message":
                continue
            data = message.get("data")
            try:
                transport.feed_inbound(json.loads(data))
            except (ValueError, TypeError):  # pragma: no cover - defensive
                continue

    task = asyncio.ensure_future(_reader())

    async def _teardown() -> None:
        """Stop the reader and give the channel's pubsub connection back.

        The unsubscribe is wrapped so the connection is closed even when it
        fails — a dropped Redis link would otherwise leak the pubsub object
        for every session that ended after the outage. ``aclose`` is
        preferred when present, falling back to ``close`` for older redis
        clients.
        """
        task.cancel()
        try:
            await pubsub.unsubscribe(channel)
        finally:
            aclose = getattr(pubsub, "aclose", None) or pubsub.close
            await aclose()

    return _teardown

deliver async

deliver(session_id: str, envelope: dict[str, Any], local: SSETransport | None) -> bool

Feed a local transport directly, else publish for the holding instance.

PUBLISH answers with the number of subscribers that received the message, and that is the only evidence available that some instance still holds the session. Reporting success unconditionally (as this used to) turned every post for a session that had already ended into a silent 204: the client believed its click had been delivered and the event was gone.

Source code in tempestweb/server/sessions.py

async def deliver(
    self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
) -> bool:
    """Feed a local transport directly, else publish for the holding instance.

    ``PUBLISH`` answers with the number of subscribers that received the
    message, and that is the only evidence available that some instance still
    holds the session. Reporting success unconditionally (as this used to)
    turned every post for a session that had already ended into a silent
    ``204``: the client believed its click had been delivered and the event
    was gone.
    """
    if local is not None:
        local.feed_inbound(envelope)
        return True
    receivers = await self._client.publish(
        self._prefix + session_id, json.dumps(envelope)
    )
    return bool(receivers)

SessionRouter

Bases: Protocol

Routes SSE inbound events to the transport holding the stream.

Source code in tempestweb/server/sessions.py

class SessionRouter(Protocol):
    """Routes SSE inbound events to the transport holding the stream."""

    async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
        """Start delivering inbound events for ``session_id`` to ``transport``.

        Called when an SSE stream opens on this instance. Returns a teardown
        coroutine to call when the stream closes.
        """
        ...

    async def deliver(
        self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
    ) -> bool:
        """Deliver one inbound envelope for ``session_id``.

        Args:
            session_id: The target session.
            envelope: The wire envelope (event / native_result).
            local: The transport on this instance, or ``None`` if not local.

        Returns:
            ``True`` if the event was delivered or handed off; ``False`` if it
            could not be routed (the caller returns ``404``).
        """
        ...

bind async

bind(session_id: str, transport: SSETransport) -> Teardown

Start delivering inbound events for session_id to transport.

Called when an SSE stream opens on this instance. Returns a teardown coroutine to call when the stream closes.

Source code in tempestweb/server/sessions.py

async def bind(self, session_id: str, transport: SSETransport) -> Teardown:
    """Start delivering inbound events for ``session_id`` to ``transport``.

    Called when an SSE stream opens on this instance. Returns a teardown
    coroutine to call when the stream closes.
    """
    ...

deliver async

deliver(session_id: str, envelope: dict[str, Any], local: SSETransport | None) -> bool

Deliver one inbound envelope for session_id.

Parameters:

Name	Type	Description	Default
session_id	str	The target session.	required
envelope	dict[str, Any]	The wire envelope (event / native_result).	required
local	SSETransport | None	The transport on this instance, or None if not local.	required

Returns:

Type	Description
bool	True if the event was delivered or handed off; False if it
bool	could not be routed (the caller returns 404).

Source code in tempestweb/server/sessions.py

async def deliver(
    self, session_id: str, envelope: dict[str, Any], local: SSETransport | None
) -> bool:
    """Deliver one inbound envelope for ``session_id``.

    Args:
        session_id: The target session.
        envelope: The wire envelope (event / native_result).
        local: The transport on this instance, or ``None`` if not local.

    Returns:
        ``True`` if the event was delivered or handed off; ``False`` if it
        could not be routed (the caller returns ``404``).
    """
    ...

InMemorySubscriptionStore dataclass

A simple in-memory subscription store (dev/tests).

Subscriptions are keyed by their push endpoint so re-subscribing the same browser replaces, never duplicates. Each stored record carries its owner.

Attributes:

Name	Type	Description
_by_endpoint	dict[str, dict[str, Any]]	Internal endpoint -> {owner, subscription} mapping.

Source code in tempestweb/server/webpush.py

@dataclass(slots=True)
class InMemorySubscriptionStore:
    """A simple in-memory subscription store (dev/tests).

    Subscriptions are keyed by their push ``endpoint`` so re-subscribing the same
    browser replaces, never duplicates. Each stored record carries its ``owner``.

    Attributes:
        _by_endpoint: Internal endpoint -> {owner, subscription} mapping.
    """

    _by_endpoint: dict[str, dict[str, Any]] = field(default_factory=dict)

    def add(self, owner: str, subscription: SubscriptionInfo) -> None:
        """Persist (or replace) a subscription for an owner.

        Args:
            owner: The owning user/topic identifier.
            subscription: The browser push subscription JSON (needs ``endpoint``).

        Raises:
            ValueError: If the subscription has no ``endpoint``.
        """
        endpoint = subscription.get("endpoint")
        if not endpoint:
            raise ValueError("subscription must include an endpoint")
        self._by_endpoint[endpoint] = {"owner": owner, "subscription": subscription}

    def remove(self, endpoint: str) -> bool:
        """Remove a subscription by endpoint.

        Args:
            endpoint: The push endpoint URL.

        Returns:
            True when a subscription was removed, False when absent.
        """
        return self._by_endpoint.pop(endpoint, None) is not None

    def list_for(self, owner: str) -> list[SubscriptionInfo]:
        """Return all subscriptions for an owner.

        Args:
            owner: The owning identifier.

        Returns:
            The subscriptions ([] when the owner has none).
        """
        return [
            rec["subscription"]
            for rec in self._by_endpoint.values()
            if rec["owner"] == owner
        ]

    def all(self) -> list[SubscriptionInfo]:
        """Return every stored subscription.

        Returns:
            All subscriptions ([] when empty).
        """
        return [rec["subscription"] for rec in self._by_endpoint.values()]

add

add(owner: str, subscription: SubscriptionInfo) -> None

Persist (or replace) a subscription for an owner.

Parameters:

Name	Type	Description	Default
owner	str	The owning user/topic identifier.	required
subscription	SubscriptionInfo	The browser push subscription JSON (needs endpoint).	required

Raises:

Type	Description
ValueError	If the subscription has no endpoint.

Source code in tempestweb/server/webpush.py

def add(self, owner: str, subscription: SubscriptionInfo) -> None:
    """Persist (or replace) a subscription for an owner.

    Args:
        owner: The owning user/topic identifier.
        subscription: The browser push subscription JSON (needs ``endpoint``).

    Raises:
        ValueError: If the subscription has no ``endpoint``.
    """
    endpoint = subscription.get("endpoint")
    if not endpoint:
        raise ValueError("subscription must include an endpoint")
    self._by_endpoint[endpoint] = {"owner": owner, "subscription": subscription}

remove

remove(endpoint: str) -> bool

Remove a subscription by endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The push endpoint URL.	required

Returns:

Type	Description
bool	True when a subscription was removed, False when absent.

Source code in tempestweb/server/webpush.py

def remove(self, endpoint: str) -> bool:
    """Remove a subscription by endpoint.

    Args:
        endpoint: The push endpoint URL.

    Returns:
        True when a subscription was removed, False when absent.
    """
    return self._by_endpoint.pop(endpoint, None) is not None

list_for

list_for(owner: str) -> list[SubscriptionInfo]

Return all subscriptions for an owner.

Parameters:

Name	Type	Description	Default
owner	str	The owning identifier.	required

Returns:

Type	Description
list[SubscriptionInfo]	The subscriptions ([] when the owner has none).

Source code in tempestweb/server/webpush.py

def list_for(self, owner: str) -> list[SubscriptionInfo]:
    """Return all subscriptions for an owner.

    Args:
        owner: The owning identifier.

    Returns:
        The subscriptions ([] when the owner has none).
    """
    return [
        rec["subscription"]
        for rec in self._by_endpoint.values()
        if rec["owner"] == owner
    ]

all

all() -> list[SubscriptionInfo]

Return every stored subscription.

Returns:

Type	Description
list[SubscriptionInfo]	All subscriptions ([] when empty).

Source code in tempestweb/server/webpush.py

def all(self) -> list[SubscriptionInfo]:
    """Return every stored subscription.

    Returns:
        All subscriptions ([] when empty).
    """
    return [rec["subscription"] for rec in self._by_endpoint.values()]

SendOutcome dataclass

The result of attempting to send to one subscription.

Attributes:

Name	Type	Description
endpoint	str	The target push endpoint.
ok	bool	Whether the push was accepted.
status_code	int | None	The push service HTTP status (when known), and None when it is not: a sender that returns no response object at all (an injected fake) or a str (pywebpush.webpush(curl=True)) leaves it unset rather than reporting a status nobody answered.
gone	bool	Whether the endpoint is dead (HTTP 410/404) and was pruned.
error	str | None	A human-readable error when ok is False.

Source code in tempestweb/server/webpush.py

@dataclass(slots=True)
class SendOutcome:
    """The result of attempting to send to one subscription.

    Attributes:
        endpoint: The target push endpoint.
        ok: Whether the push was accepted.
        status_code: The push service HTTP status (when known), and ``None`` when
            it is not: a sender that returns no response object at all (an
            injected fake) or a ``str`` (``pywebpush.webpush(curl=True)``) leaves
            it unset rather than reporting a status nobody answered.
        gone: Whether the endpoint is dead (HTTP 410/404) and was pruned.
        error: A human-readable error when ``ok`` is False.
    """

    endpoint: str
    ok: bool
    status_code: int | None = None
    gone: bool = False
    error: str | None = None

SubscriptionStore

Bases: Protocol

Storage protocol for push subscriptions keyed by endpoint.

A host app supplies a real implementation (SQLAlchemy, Redis, …); the default InMemorySubscriptionStore covers tests and single-process dev.

Source code in tempestweb/server/webpush.py

class SubscriptionStore(Protocol):
    """Storage protocol for push subscriptions keyed by endpoint.

    A host app supplies a real implementation (SQLAlchemy, Redis, …); the default
    ``InMemorySubscriptionStore`` covers tests and single-process dev.
    """

    def add(self, owner: str, subscription: SubscriptionInfo) -> None:
        """Persist a subscription for an owner."""
        ...

    def remove(self, endpoint: str) -> bool:
        """Remove a subscription by endpoint. Returns whether one was removed."""
        ...

    def list_for(self, owner: str) -> list[SubscriptionInfo]:
        """Return all subscriptions for an owner ([] when none)."""
        ...

    def all(self) -> list[SubscriptionInfo]:
        """Return every stored subscription ([] when none)."""
        ...

add

add(owner: str, subscription: SubscriptionInfo) -> None

Persist a subscription for an owner.

Source code in tempestweb/server/webpush.py

def add(self, owner: str, subscription: SubscriptionInfo) -> None:
    """Persist a subscription for an owner."""
    ...

remove

remove(endpoint: str) -> bool

Remove a subscription by endpoint. Returns whether one was removed.

Source code in tempestweb/server/webpush.py

def remove(self, endpoint: str) -> bool:
    """Remove a subscription by endpoint. Returns whether one was removed."""
    ...

list_for

list_for(owner: str) -> list[SubscriptionInfo]

Return all subscriptions for an owner ([] when none).

Source code in tempestweb/server/webpush.py

def list_for(self, owner: str) -> list[SubscriptionInfo]:
    """Return all subscriptions for an owner ([] when none)."""
    ...

all

all() -> list[SubscriptionInfo]

Return every stored subscription ([] when none).

Source code in tempestweb/server/webpush.py

def all(self) -> list[SubscriptionInfo]:
    """Return every stored subscription ([] when none)."""
    ...

VapidConfig dataclass

VAPID credentials and contact for signing WebPush requests.

Attributes:

Name	Type	Description
public_key	str	Base64url VAPID public key (shared with the browser client).
private_key	str	Base64url VAPID private key (secret; empty disables sending).
subject	str	VAPID sub claim — a mailto: or https: contact.

Source code in tempestweb/server/webpush.py

@dataclass(slots=True)
class VapidConfig:
    """VAPID credentials and contact for signing WebPush requests.

    Attributes:
        public_key: Base64url VAPID public key (shared with the browser client).
        private_key: Base64url VAPID private key (secret; empty disables sending).
        subject: VAPID ``sub`` claim — a ``mailto:`` or ``https:`` contact.
    """

    public_key: str = ""
    private_key: str = ""
    subject: str = "mailto:admin@example.com"

    @property
    def enabled(self) -> bool:
        """Whether sending is enabled (a private key is configured).

        Returns:
            True when a non-empty private key is present.
        """
        return bool(self.private_key)

    @classmethod
    def from_env(cls, prefix: str = "VAPID_") -> VapidConfig:
        """Build a config from environment variables.

        Reads ``<prefix>PUBLIC_KEY``, ``<prefix>PRIVATE_KEY`` and
        ``<prefix>SUBJECT``. Missing values fall back to the dataclass defaults.

        Args:
            prefix: Environment variable prefix.

        Returns:
            The populated config.
        """
        return cls(
            public_key=os.environ.get(f"{prefix}PUBLIC_KEY", ""),
            private_key=os.environ.get(f"{prefix}PRIVATE_KEY", ""),
            subject=os.environ.get(f"{prefix}SUBJECT", "mailto:admin@example.com"),
        )

enabled property

enabled: bool

Whether sending is enabled (a private key is configured).

Returns:

Type	Description
bool	True when a non-empty private key is present.

from_env classmethod

from_env(prefix: str = 'VAPID_') -> VapidConfig

Build a config from environment variables.

Reads <prefix>PUBLIC_KEY, <prefix>PRIVATE_KEY and <prefix>SUBJECT. Missing values fall back to the dataclass defaults.

Parameters:

Name	Type	Description	Default
prefix	str	Environment variable prefix.	'VAPID_'

Returns:

Type	Description
VapidConfig	The populated config.

Source code in tempestweb/server/webpush.py

@classmethod
def from_env(cls, prefix: str = "VAPID_") -> VapidConfig:
    """Build a config from environment variables.

    Reads ``<prefix>PUBLIC_KEY``, ``<prefix>PRIVATE_KEY`` and
    ``<prefix>SUBJECT``. Missing values fall back to the dataclass defaults.

    Args:
        prefix: Environment variable prefix.

    Returns:
        The populated config.
    """
    return cls(
        public_key=os.environ.get(f"{prefix}PUBLIC_KEY", ""),
        private_key=os.environ.get(f"{prefix}PRIVATE_KEY", ""),
        subject=os.environ.get(f"{prefix}SUBJECT", "mailto:admin@example.com"),
    )

VapidKeys dataclass

A freshly generated VAPID keypair (base64url, unpadded).

Attributes:

Name	Type	Description
public_key	str	The application server key the browser subscribes with (65-byte uncompressed P-256 point).
private_key	str	The signing key the server keeps secret (32-byte scalar).

Source code in tempestweb/server/webpush.py

@dataclass(slots=True)
class VapidKeys:
    """A freshly generated VAPID keypair (base64url, unpadded).

    Attributes:
        public_key: The application server key the browser subscribes with
            (65-byte uncompressed P-256 point).
        private_key: The signing key the server keeps secret (32-byte scalar).
    """

    public_key: str
    private_key: str

WebPushError

Bases: Exception

Raised by a sender to signal a push delivery failure.

Attributes:

Name	Type	Description
status_code		The push service HTTP status, when available.

Source code in tempestweb/server/webpush.py

class WebPushError(Exception):
    """Raised by a sender to signal a push delivery failure.

    Attributes:
        status_code: The push service HTTP status, when available.
    """

    def __init__(self, message: str, status_code: int | None = None) -> None:
        """Initialize the error.

        Args:
            message: The failure message.
            status_code: The push service HTTP status, when known.
        """
        super().__init__(message)
        self.status_code = status_code

WebPushService

Owns the subscription store and sends VAPID-signed push messages.

Parameters:

Name	Type	Description	Default
vapid	VapidConfig	The VAPID configuration.	required
store	SubscriptionStore | None	The subscription store (defaults to in-memory).	None
sender	WebPushSender | None	The push sender callable (defaults to pywebpush.webpush, resolved lazily on first send; injected in tests).	None
timeout	float	Per-send HTTP timeout in seconds, forwarded to the sender. pywebpush.webpush declares timeout: float | None = None and passes that None down to requests.post, so its own kwargs.pop("timeout", 10000) fallback never applies and an unbounded send is the default — a push endpoint that accepts the TCP connection and then never answers hangs the caller forever. 10 seconds is the default here: FCM answered in ~1.0 s in the device measurement, so it leaves 10x headroom for a slow service while bounding the worst case to something a request can absorb.	10.0

Source code in tempestweb/server/webpush.py

class WebPushService:
    """Owns the subscription store and sends VAPID-signed push messages.

    Args:
        vapid: The VAPID configuration.
        store: The subscription store (defaults to in-memory).
        sender: The push sender callable (defaults to ``pywebpush.webpush``,
            resolved lazily on first send; injected in tests).
        timeout: Per-send HTTP timeout in seconds, forwarded to the sender.
            ``pywebpush.webpush`` declares ``timeout: float | None = None`` and
            passes that ``None`` down to ``requests.post``, so its own
            ``kwargs.pop("timeout", 10000)`` fallback never applies and an
            unbounded send is the default — a push endpoint that accepts the TCP
            connection and then never answers hangs the caller forever. **10
            seconds** is the default here: FCM answered in ~1.0 s in the device
            measurement, so it leaves 10x headroom for a slow service while
            bounding the worst case to something a request can absorb.
    """

    def __init__(
        self,
        vapid: VapidConfig,
        store: SubscriptionStore | None = None,
        sender: WebPushSender | None = None,
        *,
        timeout: float = 10.0,
    ) -> None:
        """Initialize the service."""
        self.vapid = vapid
        self.store: SubscriptionStore = store or InMemorySubscriptionStore()
        self.timeout = timeout
        self._sender = sender

    def _prune_dead_endpoint(self, endpoint: str) -> bool:
        """Drop a dead endpoint from the store, never failing the send batch.

        This is the one place in this module where an exception is swallowed, and
        here that is the correct answer: the push service already said the
        endpoint is dead (410/404), and the fact the caller needs is already in
        :attr:`SendOutcome.gone` — whether the row actually went away changes
        nothing about *this* delivery.

        The store is host-supplied (SQLAlchemy, Redis), so its ``remove`` can
        raise on a dropped connection. Letting that propagate cancelled delivery
        to every **live** subscription queued behind the dead one in the same
        ``send_to_owner``/``broadcast`` batch, and answered ``POST
        {prefix}/send`` with a 500. Measured with a store raising on
        ``remove()`` and the dead endpoint first of two: the live endpoint was
        never even attempted.

        Args:
            endpoint: The push endpoint URL to remove.

        Returns:
            True when the store removed a subscription; False when it held none
            or the store itself failed.
        """
        try:
            return self.store.remove(endpoint)
        except Exception:  # noqa: BLE001 - a broken store must not stop delivery
            return False

    def _resolve_sender(self) -> WebPushSender:
        """Return the configured sender, resolving the default lazily.

        Returns:
            The sender callable.
        """
        if self._sender is None:
            self._sender = _default_sender()
        return self._sender

    def add_subscription(self, owner: str, subscription: SubscriptionInfo) -> None:
        """Persist a browser push subscription for an owner (POST /webpush/subscribe).

        Args:
            owner: The owning user/topic identifier.
            subscription: The browser push subscription JSON.
        """
        self.store.add(owner, subscription)

    def remove_subscription(self, endpoint: str) -> bool:
        """Remove a subscription by endpoint (DELETE /webpush/my).

        Args:
            endpoint: The push endpoint URL.

        Returns:
            Whether a subscription was removed.
        """
        return self.store.remove(endpoint)

    def send(
        self, subscription: SubscriptionInfo, payload: dict[str, Any]
    ) -> SendOutcome:
        """Send one VAPID-signed push to a single subscription.

        A dead endpoint (HTTP 410/404) is pruned from the store and reported with
        ``gone=True``; a store that fails while pruning is ignored, so one dead
        endpoint cannot cancel delivery to the live ones. Sending is a no-op
        success-free outcome when VAPID is disabled (no private key), so dev
        environments degrade cleanly.

        The call **blocks** (``pywebpush`` posts with ``requests``) and is
        bounded by ``self.timeout``. An ``async`` caller must run it off the
        loop; ``webpush_router``'s ``/send`` route goes through
        ``run_in_threadpool`` for exactly that reason.

        Args:
            subscription: The target subscription JSON (needs ``endpoint``).
            payload: The JSON-able notification payload (title/body/data/...).

        Returns:
            The send outcome, carrying the status the push service answered (or
            None when the sender exposes no response).
        """
        endpoint = str(subscription.get("endpoint", ""))
        if not self.vapid.enabled:
            return SendOutcome(
                endpoint=endpoint, ok=False, error="VAPID disabled (no private key)"
            )

        sender = self._resolve_sender()
        try:
            response = sender(
                subscription_info=subscription,
                data=json.dumps(payload),
                vapid_private_key=self.vapid.private_key,
                vapid_claims={"sub": self.vapid.subject},
                timeout=self.timeout,
            )
        except WebPushError as exc:
            gone = exc.status_code in (404, 410)
            if gone:
                self._prune_dead_endpoint(endpoint)
            return SendOutcome(
                endpoint=endpoint,
                ok=False,
                status_code=exc.status_code,
                gone=gone,
                error=str(exc),
            )
        except Exception as exc:  # noqa: BLE001 - report any sender failure
            return SendOutcome(endpoint=endpoint, ok=False, error=str(exc))
        return SendOutcome(endpoint=endpoint, ok=True, status_code=_status_of(response))

    def send_to_owner(self, owner: str, payload: dict[str, Any]) -> list[SendOutcome]:
        """Send a push to every subscription an owner has.

        Returns [] when the owner has no subscriptions (never an error).

        Args:
            owner: The owning identifier.
            payload: The notification payload.

        Returns:
            One outcome per subscription ([] when the owner has none).
        """
        return [self.send(sub, payload) for sub in self.store.list_for(owner)]

    def broadcast(self, payload: dict[str, Any]) -> list[SendOutcome]:
        """Send a push to every stored subscription.

        Args:
            payload: The notification payload.

        Returns:
            One outcome per subscription ([] when the store is empty).
        """
        return [self.send(sub, payload) for sub in self.store.all()]

add_subscription

add_subscription(owner: str, subscription: SubscriptionInfo) -> None

Persist a browser push subscription for an owner (POST /webpush/subscribe).

Parameters:

Name	Type	Description	Default
owner	str	The owning user/topic identifier.	required
subscription	SubscriptionInfo	The browser push subscription JSON.	required

Source code in tempestweb/server/webpush.py

def add_subscription(self, owner: str, subscription: SubscriptionInfo) -> None:
    """Persist a browser push subscription for an owner (POST /webpush/subscribe).

    Args:
        owner: The owning user/topic identifier.
        subscription: The browser push subscription JSON.
    """
    self.store.add(owner, subscription)

remove_subscription

remove_subscription(endpoint: str) -> bool

Remove a subscription by endpoint (DELETE /webpush/my).

Parameters:

Name	Type	Description	Default
endpoint	str	The push endpoint URL.	required

Returns:

Type	Description
bool	Whether a subscription was removed.

Source code in tempestweb/server/webpush.py

def remove_subscription(self, endpoint: str) -> bool:
    """Remove a subscription by endpoint (DELETE /webpush/my).

    Args:
        endpoint: The push endpoint URL.

    Returns:
        Whether a subscription was removed.
    """
    return self.store.remove(endpoint)

send

send(subscription: SubscriptionInfo, payload: dict[str, Any]) -> SendOutcome

Send one VAPID-signed push to a single subscription.

A dead endpoint (HTTP 410/404) is pruned from the store and reported with gone=True; a store that fails while pruning is ignored, so one dead endpoint cannot cancel delivery to the live ones. Sending is a no-op success-free outcome when VAPID is disabled (no private key), so dev environments degrade cleanly.

The call blocks (pywebpush posts with requests) and is bounded by self.timeout. An async caller must run it off the loop; webpush_router's /send route goes through run_in_threadpool for exactly that reason.

Parameters:

Name	Type	Description	Default
subscription	SubscriptionInfo	The target subscription JSON (needs endpoint).	required
payload	dict[str, Any]	The JSON-able notification payload (title/body/data/...).	required

Returns:

Type	Description
SendOutcome	The send outcome, carrying the status the push service answered (or
SendOutcome	None when the sender exposes no response).

Source code in tempestweb/server/webpush.py

def send(
    self, subscription: SubscriptionInfo, payload: dict[str, Any]
) -> SendOutcome:
    """Send one VAPID-signed push to a single subscription.

    A dead endpoint (HTTP 410/404) is pruned from the store and reported with
    ``gone=True``; a store that fails while pruning is ignored, so one dead
    endpoint cannot cancel delivery to the live ones. Sending is a no-op
    success-free outcome when VAPID is disabled (no private key), so dev
    environments degrade cleanly.

    The call **blocks** (``pywebpush`` posts with ``requests``) and is
    bounded by ``self.timeout``. An ``async`` caller must run it off the
    loop; ``webpush_router``'s ``/send`` route goes through
    ``run_in_threadpool`` for exactly that reason.

    Args:
        subscription: The target subscription JSON (needs ``endpoint``).
        payload: The JSON-able notification payload (title/body/data/...).

    Returns:
        The send outcome, carrying the status the push service answered (or
        None when the sender exposes no response).
    """
    endpoint = str(subscription.get("endpoint", ""))
    if not self.vapid.enabled:
        return SendOutcome(
            endpoint=endpoint, ok=False, error="VAPID disabled (no private key)"
        )

    sender = self._resolve_sender()
    try:
        response = sender(
            subscription_info=subscription,
            data=json.dumps(payload),
            vapid_private_key=self.vapid.private_key,
            vapid_claims={"sub": self.vapid.subject},
            timeout=self.timeout,
        )
    except WebPushError as exc:
        gone = exc.status_code in (404, 410)
        if gone:
            self._prune_dead_endpoint(endpoint)
        return SendOutcome(
            endpoint=endpoint,
            ok=False,
            status_code=exc.status_code,
            gone=gone,
            error=str(exc),
        )
    except Exception as exc:  # noqa: BLE001 - report any sender failure
        return SendOutcome(endpoint=endpoint, ok=False, error=str(exc))
    return SendOutcome(endpoint=endpoint, ok=True, status_code=_status_of(response))

send_to_owner

send_to_owner(owner: str, payload: dict[str, Any]) -> list[SendOutcome]

Send a push to every subscription an owner has.

Returns [] when the owner has no subscriptions (never an error).

Parameters:

Name	Type	Description	Default
owner	str	The owning identifier.	required
payload	dict[str, Any]	The notification payload.	required

Returns:

Type	Description
list[SendOutcome]	One outcome per subscription ([] when the owner has none).

Source code in tempestweb/server/webpush.py

def send_to_owner(self, owner: str, payload: dict[str, Any]) -> list[SendOutcome]:
    """Send a push to every subscription an owner has.

    Returns [] when the owner has no subscriptions (never an error).

    Args:
        owner: The owning identifier.
        payload: The notification payload.

    Returns:
        One outcome per subscription ([] when the owner has none).
    """
    return [self.send(sub, payload) for sub in self.store.list_for(owner)]

broadcast

broadcast(payload: dict[str, Any]) -> list[SendOutcome]

Send a push to every stored subscription.

Parameters:

Name	Type	Description	Default
payload	dict[str, Any]	The notification payload.	required

Returns:

Type	Description
list[SendOutcome]	One outcome per subscription ([] when the store is empty).

Source code in tempestweb/server/webpush.py

def broadcast(self, payload: dict[str, Any]) -> list[SendOutcome]:
    """Send a push to every stored subscription.

    Args:
        payload: The notification payload.

    Returns:
        One outcome per subscription ([] when the store is empty).
    """
    return [self.send(sub, payload) for sub in self.store.all()]

create_app

create_app(state_factory: Callable[[], S], view: Callable[[App[S]], Widget], *, title: str = 'tempestweb', security: SecurityConfig | None = None, metrics: bool = False, observability: ServerObservability | None = None, sse_backend: SessionRouter | None = None, concurrent_dispatch: bool = False, theme: Theme | None = None) -> FastAPI

Build a Mode B FastAPI app for a view and state factory.

Parameters:

Name	Type	Description	Default
state_factory	Callable[[], S]	Builds a fresh state per connection (isolation).	required
view	Callable[[App[S]], Widget]	The shared view function rendered for each session.	required
title	str	OpenAPI title for the FastAPI app.	'tempestweb'
security	SecurityConfig | None	Opt-in auth + origin controls (Track S — S0/S1/S3). None leaves the host open (dev); pass a :class:SecurityConfig with an authenticate predicate and/or allowed_origins for production.	None
metrics	bool	When True, mount GET /metrics (Prometheus text) — S8.	False
observability	ServerObservability | None	Patch latency, structured session logs and tracing (S8). None keeps the inert default: nothing collected, no exporter imported. Pass ServerObservability(metrics=PatchMetrics()) to get the latency histogram alongside the connection counters, a logger for one JSON line per session lifecycle event, and tracer= otel_tracer() for a span per session and per patch round.	None
sse_backend	SessionRouter | None	SSE inbound router (Track S — S4). None is in-process (sticky sessions); a RedisSessionRouter scales SSE without sticky.	None
concurrent_dispatch	bool	Dispatch each event as its own task (ordered per widget key) instead of one at a time, so a slow handler cannot freeze the connection. Off by default; tempestweb.runtime.spawn handles the common case without changing dispatch semantics.	False
theme	Theme | None	The palette every component resolves its colors against. None keeps the Material baseline. Pair it with :func:~tempestweb.html.theme_css in the page head: the CSS variables cover what the base stylesheet paints, and this covers what components resolve in Python.	None

Returns:

Type	Description
FastAPI	The configured FastAPI application with WS and SSE routes mounted.

Source code in tempestweb/server/app.py

def create_app(
    state_factory: Callable[[], S],
    view: Callable[[App[S]], Widget],
    *,
    title: str = "tempestweb",
    security: SecurityConfig | None = None,
    metrics: bool = False,
    observability: ServerObservability | None = None,
    sse_backend: SessionRouter | None = None,
    concurrent_dispatch: bool = False,
    theme: Theme | None = None,
) -> FastAPI:
    """Build a Mode B FastAPI app for a ``view`` and state factory.

    Args:
        state_factory: Builds a fresh state per connection (isolation).
        view: The shared ``view`` function rendered for each session.
        title: OpenAPI title for the FastAPI app.
        security: Opt-in auth + origin controls (Track S — S0/S1/S3). ``None``
            leaves the host open (dev); pass a :class:`SecurityConfig` with an
            ``authenticate`` predicate and/or ``allowed_origins`` for production.
        metrics: When ``True``, mount ``GET /metrics`` (Prometheus text) — S8.
        observability: Patch latency, structured session logs and tracing (S8).
            ``None`` keeps the inert default: nothing collected, no exporter
            imported. Pass ``ServerObservability(metrics=PatchMetrics())`` to get
            the latency histogram alongside the connection counters, a ``logger``
            for one JSON line per session lifecycle event, and ``tracer=
            otel_tracer()`` for a span per session and per patch round.
        sse_backend: SSE inbound router (Track S — S4). ``None`` is in-process
            (sticky sessions); a ``RedisSessionRouter`` scales SSE without sticky.
        concurrent_dispatch: Dispatch each event as its own task (ordered per
            widget key) instead of one at a time, so a slow handler cannot freeze
            the connection. Off by default; ``tempestweb.runtime.spawn`` handles
            the common case without changing dispatch semantics.
        theme: The palette every component resolves its colors against.
            ``None`` keeps the Material baseline. Pair it with
            :func:`~tempestweb.html.theme_css` in the page head: the CSS
            variables cover what the base stylesheet paints, and this covers
            what components resolve in Python.

    Returns:
        The configured FastAPI application with WS and SSE routes mounted.
    """
    return TempestWebServer(
        state_factory,
        view,
        title=title,
        security=security,
        metrics=metrics,
        observability=observability,
        sse_backend=sse_backend,
        concurrent_dispatch=concurrent_dispatch,
        theme=theme,
    ).api

jwt_authenticator

jwt_authenticator(key: str, *, algorithms: tuple[str, ...] = ('HS256',), audience: str | None = None, issuer: str | None = None, require_expiry: bool = True) -> Authenticate

Build an authenticate callable that verifies a bearer JWT (S3).

Parameters:

Name	Type	Description	Default
key	str	The signing key / secret.	required
algorithms	tuple[str, ...]	Accepted signing algorithms.	('HS256',)
audience	str | None	Expected aud claim, if any.	None
issuer	str | None	Expected iss claim, if any.	None
require_expiry	bool	Refuse a token with no exp claim (see :func:verify_jwt).	True

Returns:

Type	Description
Authenticate	A predicate that accepts a connection with a valid, unexpired JWT.

Source code in tempestweb/server/security.py

def jwt_authenticator(
    key: str,
    *,
    algorithms: tuple[str, ...] = ("HS256",),
    audience: str | None = None,
    issuer: str | None = None,
    require_expiry: bool = True,
) -> Authenticate:
    """Build an ``authenticate`` callable that verifies a bearer JWT (S3).

    Args:
        key: The signing key / secret.
        algorithms: Accepted signing algorithms.
        audience: Expected ``aud`` claim, if any.
        issuer: Expected ``iss`` claim, if any.
        require_expiry: Refuse a token with no ``exp`` claim (see
            :func:`verify_jwt`).

    Returns:
        A predicate that accepts a connection with a valid, unexpired JWT.
    """

    def _authenticate(credentials: Credentials) -> bool:
        """Accept the connection when its bearer token is a valid JWT.

        A missing token is refused without attempting verification. Both
        failures ``verify_jwt`` can raise — a malformed or expired token
        (``ValueError``) and a missing signing dependency (``RuntimeError``) —
        are answered as "not authenticated" rather than propagating, so the
        handshake never turns into a 500 for an unauthenticated client.

        Args:
            credentials: The connection's token, origin, headers and query.

        Returns:
            ``True`` when the token verifies against the enclosing key.
        """
        if not credentials.token:
            return False
        try:
            verify_jwt(
                credentials.token,
                key,
                algorithms=algorithms,
                audience=audience,
                issuer=issuer,
                require_expiry=require_expiry,
            )
        except (ValueError, RuntimeError):
            return False
        return True

    return _authenticate

token_authenticator

token_authenticator(secret: str) -> Authenticate

Build an authenticate callable for a shared-secret token.

Compares the connection's bearer token to secret with a constant-time check (the X-Token convention). An empty secret disables the gate (always allows) — dev-only, matching the framework's "empty secret disables auth" rule.

Parameters:

Name	Type	Description	Default
secret	str	The shared secret; empty disables the gate.	required

Returns:

Type	Description
Authenticate	A predicate that accepts a connection whose token equals secret.

Source code in tempestweb/server/security.py

def token_authenticator(secret: str) -> Authenticate:
    """Build an ``authenticate`` callable for a shared-secret token.

    Compares the connection's bearer token to ``secret`` with a constant-time
    check (the ``X-Token`` convention). An **empty** secret disables the gate
    (always allows) — dev-only, matching the framework's "empty secret disables
    auth" rule.

    Args:
        secret: The shared secret; empty disables the gate.

    Returns:
        A predicate that accepts a connection whose token equals ``secret``.
    """

    def _authenticate(credentials: Credentials) -> bool:
        """Accept the connection when its bearer token equals the shared secret.

        The comparison is constant-time, so a rejected token leaks nothing about
        how much of it was right. An empty enclosing secret short-circuits to
        ``True``, which is the documented dev-only way to disable the gate.

        Args:
            credentials: The connection's token, origin, headers and query.

        Returns:
            ``True`` when the gate is disabled or the token matches.
        """
        if not secret:
            return True
        token = credentials.token or ""
        return hmac.compare_digest(token, secret)

    return _authenticate

verify_jwt

verify_jwt(token: str, key: str, *, algorithms: tuple[str, ...] = ('HS256',), audience: str | None = None, issuer: str | None = None, require_expiry: bool = True) -> dict[str, Any]

Verify a JWT's signature and expiry, returning its claims.

Unlike observability.auth.decode_jwt (which only base64url-decodes the payload), this validates the signature and standard time claims.

Parameters:

Name	Type	Description	Default
token	str	The compact-serialization JWT.	required
key	str	The signing key / secret.	required
algorithms	tuple[str, ...]	Accepted signing algorithms.	('HS256',)
audience	str | None	Expected aud claim, if any.	None
issuer	str | None	Expected iss claim, if any.	None
require_expiry	bool	Refuse a token that carries no exp claim. PyJWT only checks an expiry that is present, so without this a token minted without exp is accepted forever — which is not what "verifies the expiry" can mean. Set False only for a token whose lifetime something else bounds.	True

Returns:

Type	Description
dict[str, Any]	The verified claims.

Raises:

Type	Description
RuntimeError	If PyJWT is not installed.
ValueError	If the token is invalid, expired, missing a required claim, or fails a claim check.

Source code in tempestweb/server/security.py

def verify_jwt(
    token: str,
    key: str,
    *,
    algorithms: tuple[str, ...] = ("HS256",),
    audience: str | None = None,
    issuer: str | None = None,
    require_expiry: bool = True,
) -> dict[str, Any]:
    """Verify a JWT's signature and expiry, returning its claims.

    Unlike ``observability.auth.decode_jwt`` (which only base64url-decodes the
    payload), this validates the signature and standard time claims.

    Args:
        token: The compact-serialization JWT.
        key: The signing key / secret.
        algorithms: Accepted signing algorithms.
        audience: Expected ``aud`` claim, if any.
        issuer: Expected ``iss`` claim, if any.
        require_expiry: Refuse a token that carries no ``exp`` claim. PyJWT only
            checks an expiry that is *present*, so without this a token minted
            without ``exp`` is accepted forever — which is not what "verifies
            the expiry" can mean. Set ``False`` only for a token whose lifetime
            something else bounds.

    Returns:
        The verified claims.

    Raises:
        RuntimeError: If PyJWT is not installed.
        ValueError: If the token is invalid, expired, missing a required claim,
            or fails a claim check.
    """
    try:
        import jwt  # type: ignore[import-not-found]  # optional [auth] extra
    except ImportError as exc:  # pragma: no cover - exercised via the error path
        raise RuntimeError(
            "PyJWT is required for verify_jwt; install "
            'tempest-fastapi-sdk[auth] or "pyjwt".'
        ) from exc
    required: list[str] = ["exp"] if require_expiry else []
    try:
        return dict(
            jwt.decode(
                token,
                key,
                algorithms=list(algorithms),
                audience=audience,
                issuer=issuer,
                options={"require": required},
            )
        )
    except jwt.PyJWTError as exc:
        raise ValueError(f"invalid token: {exc}") from exc

generate_vapid_keys

generate_vapid_keys() -> VapidKeys

Generate a P-256 VAPID keypair for WebPush.

The public key is the browser applicationServerKey and the private key signs push requests server-side. Store the private key as a secret (env var) — never commit it.

Returns:

Type	Description
VapidKeys	The base64url-encoded :class:VapidKeys.

Raises:

Type	Description
RuntimeError	If cryptography is not installed.

Source code in tempestweb/server/webpush.py

def generate_vapid_keys() -> VapidKeys:
    """Generate a P-256 VAPID keypair for WebPush.

    The public key is the browser ``applicationServerKey`` and the private key
    signs push requests server-side. Store the private key as a secret (env var)
    — never commit it.

    Returns:
        The base64url-encoded :class:`VapidKeys`.

    Raises:
        RuntimeError: If ``cryptography`` is not installed.
    """
    try:
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric import ec
    except ImportError as exc:  # pragma: no cover - exercised via the error path
        raise RuntimeError(
            "cryptography is required to generate VAPID keys; install "
            'tempest-fastapi-sdk[webpush] or "cryptography".'
        ) from exc
    private = ec.generate_private_key(ec.SECP256R1())
    public_point = private.public_key().public_bytes(
        serialization.Encoding.X962,
        serialization.PublicFormat.UncompressedPoint,
    )
    private_scalar = private.private_numbers().private_value.to_bytes(32, "big")
    return VapidKeys(
        public_key=_b64url(public_point),
        private_key=_b64url(private_scalar),
    )

webpush_router

webpush_router(service: WebPushService, *, owner: str = 'default', prefix: str = '/webpush') -> Any

Build a FastAPI router exposing the WebPush subscribe/send endpoints.

Mount it on a host app with app.include_router(webpush_router(service)). Endpoints (all JSON):

• GET {prefix}/vapid-public-key → {"public_key": ...} for the client to subscribe with.
• POST {prefix}/subscribe (body: the browser subscription JSON) → stores it under owner.
• POST {prefix}/unsubscribe (body: {"endpoint": ...}) → removes it, only when this owner holds it; a body with no endpoint answers 400, like /subscribe already did.
• POST {prefix}/send (body: the notification payload) → pushes to every subscription of owner; returns {"sent", "total"}. The blocking send runs in a worker thread, so it never stalls the event loop.

A single fixed owner keeps the default multi-tenant-free; an app with real users wires its own routes around the same :class:WebPushService, resolving the owner from auth.

Parameters:

Name	Type	Description	Default
service	WebPushService	The WebPush service (store + sender).	required
owner	str	The owner every subscription is filed under (default "default").	'default'
prefix	str	The route prefix.	'/webpush'

Returns:

Type	Description
Any	A configured fastapi.APIRouter.

Raises:

Type	Description
RuntimeError	If FastAPI is not installed.

Source code in tempestweb/server/webpush.py

def webpush_router(
    service: WebPushService,
    *,
    owner: str = "default",
    prefix: str = "/webpush",
) -> Any:  # noqa: ANN401 - a FastAPI APIRouter (imported lazily)
    """Build a FastAPI router exposing the WebPush subscribe/send endpoints.

    Mount it on a host app with ``app.include_router(webpush_router(service))``.
    Endpoints (all JSON):

    - ``GET  {prefix}/vapid-public-key`` → ``{"public_key": ...}`` for the client
      to subscribe with.
    - ``POST {prefix}/subscribe`` (body: the browser subscription JSON) → stores
      it under ``owner``.
    - ``POST {prefix}/unsubscribe`` (body: ``{"endpoint": ...}``) → removes it,
      **only when this ``owner`` holds it**; a body with no ``endpoint`` answers
      400, like ``/subscribe`` already did.
    - ``POST {prefix}/send`` (body: the notification payload) → pushes to every
      subscription of ``owner``; returns ``{"sent", "total"}``. The blocking
      send runs in a worker thread, so it never stalls the event loop.

    A single fixed ``owner`` keeps the default multi-tenant-free; an app with
    real users wires its own routes around the same :class:`WebPushService`,
    resolving the owner from auth.

    Args:
        service: The WebPush service (store + sender).
        owner: The owner every subscription is filed under (default ``"default"``).
        prefix: The route prefix.

    Returns:
        A configured ``fastapi.APIRouter``.

    Raises:
        RuntimeError: If FastAPI is not installed.
    """
    try:
        from fastapi import APIRouter, Body, HTTPException
        from starlette.concurrency import run_in_threadpool
    except ImportError as exc:  # pragma: no cover - server extra always ships it
        raise RuntimeError(
            "FastAPI is required for webpush_router; install "
            "tempest-fastapi-sdk or the server extra."
        ) from exc

    router = APIRouter(prefix=prefix, tags=["webpush"])

    @router.get("/vapid-public-key")
    async def vapid_public_key() -> dict[str, str]:
        """Return the VAPID public key the browser subscribes with."""
        return {"public_key": service.vapid.public_key}

    @router.post("/subscribe")
    async def subscribe(
        subscription: SubscriptionInfo = Body(...),  # noqa: B008 - FastAPI param
    ) -> dict[str, bool]:
        """Persist a browser push subscription under the router's owner.

        A body without an ``endpoint`` is the caller's mistake, so it answers
        **400** naming it. It used to reach the store's ``ValueError`` uncaught
        and come back as a 500 with a traceback — measured while wiring a probe
        that posted the wrong shape.

        Raises:
            HTTPException: 400, when the body carries no push endpoint.
        """
        try:
            service.add_subscription(owner, subscription)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"ok": True}

    @router.post("/unsubscribe")
    async def unsubscribe(
        body: dict[str, Any] = Body(...),  # noqa: B008 - FastAPI param
    ) -> dict[str, bool]:
        """Remove one of **this owner's** subscriptions, by endpoint.

        The scope is the point. The store is keyed by endpoint alone, and the
        signature invites two routers over one service
        (``webpush_router(svc, owner="alice", prefix="/webpush/alice")`` and the
        same for ``"bob"``), so an unscoped remove let
        ``POST /webpush/alice/unsubscribe`` carrying **bob's** endpoint delete
        bob's subscription and answer ``{"removed": true}``. An endpoint this
        owner does not hold now answers ``{"removed": false}`` — the same answer
        as one already gone, so the route never reveals that somebody else holds
        it.

        Pruning a dead endpoint inside :meth:`WebPushService.send` stays
        unscoped, and that is correct: there the push service itself reported the
        endpoint dead, so it is dead for every owner.

        Raises:
            HTTPException: 400, when the body carries no push endpoint. It used
                to fall through to ``remove("")`` and answer
                ``{"removed": false}`` in silence, while ``/subscribe`` already
                answered 400 to the very same malformed body.
        """
        endpoint = str(body.get("endpoint") or "")
        if not endpoint:
            raise HTTPException(status_code=400, detail="body must include an endpoint")
        held = any(
            str(sub.get("endpoint", "")) == endpoint
            for sub in service.store.list_for(owner)
        )
        if not held:
            return {"removed": False}
        return {"removed": service.remove_subscription(endpoint)}

    @router.post("/send")
    async def send(
        payload: dict[str, Any] = Body(...),  # noqa: B008 - FastAPI param
    ) -> dict[str, int]:
        """Push a notification payload to every subscription of the owner.

        The fan-out runs in a worker thread because it **blocks**:
        ``pywebpush`` posts with ``requests``, and this is an ``async`` route on
        the same event loop that streams patches over WebSocket/SSE. Measured
        with a sender sleeping 1 s and three subscriptions: called inline the
        request took 3.00 s and a 10 ms heartbeat on the same loop got **zero**
        ticks — every connected app frozen for the whole fan-out. Through
        ``run_in_threadpool`` the same request takes 3.01 s and the heartbeat
        keeps ticking (296 ticks, worst lateness 0.01 s).
        """
        outcomes = await run_in_threadpool(service.send_to_owner, owner, payload)
        return {"sent": sum(1 for o in outcomes if o.ok), "total": len(outcomes)}

    return router