From 4a3f96acad57ea91d61b68cfa6c93cebbfc6f113 Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:11:22 +0700 Subject: [PATCH 1/6] feat(2.0)!: invert WsMessage into a backend-independent WireMessage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before this, `WsMessage` was a re-export of `tungstenite::Message` whenever the `ws` feature was on, leaking a backend type through session, toolbox, subs, push, conn, server and client. The `#[cfg(not(feature = "ws"))]` inner enum was already the right shape; promote it to the canonical type. - `WireMessage` is now unconditional, `#[non_exhaustive]`, and carries the same five variants. `pub type WsMessage = WireMessage` keeps type positions compiling; it does NOT preserve tungstenite's inherent methods. - Conversions to/from tungstenite live only in the tungstenite backend module and are applied at two edges: the server upgrader's WsStream impl and the client's private stream helpers. - Swept client.rs alongside the modules named in the plan — it was an unlisted leak site (stream_send took tungstenite's Message directly). - `WireMessage::as_text` folds the old Text/Binary UTF-8 handling into one place; `is_close` replaces a common inherent-method call. - tungstenite's `Frame` variant degrades to an empty Binary rather than panicking: this crate never uses the low-level frame API, and an unexpected raw frame is not worth killing a live session over. The ws-wtx backend is deliberately untouched — it is dead code (compile_error deprecation, denylisted, no longer implements WsUpgrader). Verified: types / ws-core / full all build, ws-core pulls no tungstenite, clippy clean, 67 tests green (65 baseline + 2 new round-trip tests). Co-Authored-By: Claude Fable 5 --- PLAN-2.0.md | 463 ++++++++++++++++++++++++++++ PLAN-2.1.md | 333 ++++++++++++++++++++ src/libs/ws/client.rs | 11 +- src/libs/ws/message.rs | 85 +++-- src/libs/ws/tungstenite.rs | 1 + src/libs/ws/tungstenite/message.rs | 82 +++++ src/libs/ws/tungstenite/upgrader.rs | 9 +- 7 files changed, 959 insertions(+), 25 deletions(-) create mode 100644 PLAN-2.0.md create mode 100644 PLAN-2.1.md create mode 100644 src/libs/ws/tungstenite/message.rs diff --git a/PLAN-2.0.md b/PLAN-2.0.md new file mode 100644 index 0000000..9c7573f --- /dev/null +++ b/PLAN-2.0.md @@ -0,0 +1,463 @@ +# endpoint-libs 2.0 — implementation plan (brief for Claude Code) + +Repo: `~/code/endpoint-libs` (currently 1.9.1, edition 2024). Goal of 2.0: make the +schema/handler/MCP machinery **transport-agnostic** so the same server core runs over +TCP+TLS+WebSocket (today's path, unchanged behavior) *and* over local attested +transports (Unix sockets, Windows named pipes, macOS XPC) implemented later in a +sibling crate. Three designs are deliberately stolen from tarpc 0.37 (MIT): its +`Transport` trait shape, its `request_hook` Before/After pattern, and its +`serde_transport` framing stack. Do NOT add tarpc as a dependency. + +Ground rules: + +- Existing consumers that only use `add_handler` + `listen()` + `RequestHandler` must + compile with zero or trivial changes. Breaking changes are confined to the types + named in §3 (peer identity) and §3b (schema model) — nowhere else. +- No new required dependencies in default features. New deps (`tokio-util`, + `tokio-serde`) go behind a new feature flag. +- Every phase lands green. **NB (corrected against the repo):** `cargo test + --all-features` can never pass — `ws` and `ws-wtx` are mutually exclusive via + `compile_error!` guards (`src/lib.rs:4`). CI uses `cargo all-features test`, which + respects the `cargo-all-features` denylist in `Cargo.toml:230`. Use that, or the + explicit matrix in §9. Baseline on `main` at 1.9.1: `--features full` = 65 tests + green. +- The `ws-wtx` backend is **dead code** — `compile_error!` deprecation + (`src/libs/ws/wtx.rs:1`), denylisted, and no longer implements the current + `WsUpgrader` trait. **Decision: leave it untouched.** Phase 1 does NOT add + `WireMessage` conversions for it (the original §2 said to; that was written before + the rot was known). Do not "fix" it in passing. +- Work in phases, one PR-sized commit series each, in the order given. Phase 1+2 are + this brief's scope; §7 defines contracts a sibling crate implements later — expose + the traits, don't implement platform code here. + +--- + +## 1. The stolen pieces (reference shapes) + +### 1a. tarpc's Transport: a blanket alias over Sink + Stream + +```rust +// tarpc/src/transport.rs (shape to replicate, adapted names) +pub trait Transport +where + Self: Stream>::Error>>, + Self: Sink>::TransportError>, +{ + type TransportError: std::error::Error + Send + Sync + 'static; +} +impl Transport for T +where + T: ?Sized + Stream> + Sink, + E: std::error::Error + Send + Sync + 'static, +{ type TransportError = E; } +``` + +Key property: implementors never name the trait — anything that is `Sink + Stream` of +the right item types *is* a transport. This composes with `tokio_util::codec::Framed`, +`tokio_serde`, and hand-rolled adapters (XPC) for free. + +### 1b. tarpc's serde_transport: framing = LengthDelimitedCodec + serde codec + +```rust +// shape: Framed wrapped by tokio_serde::Framed +// endpoint-libs version: serde_json value = one length-delimited frame +pub fn framed(io: S) -> impl Transport +``` + +### 1c. tarpc's request_hook: BeforeRequest / AfterRequest + +Hooks run before execution (may reject with a typed error, may mutate context) and +after completion (observe result). This is the seam where mission-token verification +will plug in without the transport or handler layers knowing about it. + +--- + +## 2. Phase 1 — `WireMessage` inversion (non-breaking) + +**Problem:** with feature `ws` on, `WsMessage` is a re-export of +`tokio_tungstenite::tungstenite::Message`, so a backend type leaks through session, +toolbox, subs, push, and conn. The `#[cfg(not(feature = "ws"))] mod inner` enum in +`src/libs/ws/message.rs` is already the shape we want. + +**Change:** + +- Promote the inner enum to the unconditional canonical type, rename `WireMessage` + (keep `pub type WsMessage = WireMessage;` alias for compat). Variants: `Text(String)`, + `Binary(Vec)`, `Ping(Vec)`, `Pong(Vec)`, `Close(Option)`. +- Add `From for tungstenite::Message` and the reverse, gated on the + tungstenite backend. Conversions live in the tungstenite backend module only. + **Not the wtx backend** — it is dead code, see ground rules. +- Sweep: session.rs, toolbox.rs, subs.rs, push.rs, conn.rs, server.rs **and client.rs** + use `WireMessage` exclusively. Grep for `tungstenite::Message` outside the two backend + dirs → must be zero. + - `client.rs` is easy to miss and is a real leak site: `WsClient::stream_send` + (`client.rs:108`) takes tungstenite's `Message`, and the private `enum WsStream` + (`client.rs:55`) holds tungstenite stream types. Convert the helpers to + `WireMessage` now — Phase 3 adds a local-transport client constructor on top of + this, and doing the sweep later means re-opening and re-testing the same paths. +- The `pub type WsMessage = WireMessage;` alias covers *type positions only*. It does + **not** preserve tungstenite's inherent methods (`.into_text()`, `.is_close()`, + `.into_data()`). Grep consumers for those call sites and list them in the migration + doc individually — "the `From` impls cover most uses" is not true for method calls. + +**Acceptance:** all tests/examples pass with `--features full`; `ws-core` builds +without tungstenite in the tree (`cargo tree -e features` proof). + +## 3. Phase 2 — `PeerIdentity` (breaking change 1 of 2; see also §3b) + +New module `src/libs/peer.rs`: + +```rust +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum PeerIdentity { + /// TCP/TLS network peer (today's behavior). + Network(SocketAddr), + /// Same-machine peer over a local transport. + Local(LocalPeer), + Unknown, +} + +#[derive(Debug, Clone)] +pub struct LocalPeer { + pub pid: Option, + pub uid: Option, // unix only; None on Windows + pub attestation: Attestation, +} + +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum Attestation { + /// Transport did not verify code identity. + None, + /// OS/kernel or transport-level verification succeeded. + Verified { + /// e.g. "xpc-codesign-requirement", "pidfd-exe-sha256", "pipe-sid-dacl" + mechanism: &'static str, + /// e.g. the requirement string, digest, or SID that matched + subject: String, + }, +} + +impl PeerIdentity { + /// Best-effort IP for logging; loopback for local peers. + pub fn ip_addr(&self) -> IpAddr { ... } +} +``` + +**Threading it through (breaking):** + +- `WsConnection.address: SocketAddr` → `WsConnection.peer: PeerIdentity`. Add + `pub fn address(&self) -> SocketAddr` compat accessor (Network addr, else + `127.0.0.1:0`), mark `#[deprecated(note = "use .peer")]`. +- `RequestContext`: keep `ip_addr` field **populated from `peer.ip_addr()`** (most + consumers only log it — they keep compiling), add `pub peer: PeerIdentity`. +- `WsConnection` and `RequestContext` gain `extensions: Extensions` — implement a + minimal typed map (`anymap` pattern over `HashMap>`, + ~60 lines, no new dep; model after `http::Extensions`). Connection-scoped extensions + carry the attestation; request-scoped extensions will carry verified mission claims. +- Logging call sites that format `?addr` switch to `%conn.peer_display()` (add a + compact Display). + +**Acceptance:** compile + tests; a grep inventory of every `ip_addr`/`address` consumer +goes in the PR description (there are ~15 sites: server.rs, session.rs, toolbox +logging, headers.rs real-IP handling — the header-derived real-IP override must keep +working for the Network variant). + +## 3b. Phase 2b — schema-model future-proofing (breaking, ~20 lines, do it with Phase 2) + +**Why this is in 2.0 even though nothing uses it yet.** OpenAPI/AsyncAPI emission is a +2.1 feature living in *endpointgen* (see `PLAN-2.1.md`). It touches no runtime code and +is purely additive — **provided** the schema model can absorb new information without a +breaking change. Today it cannot. This phase buys that, and it is only free while 2.0 is +already breaking. Skipping it means OpenAPI enrichment forces a 3.0. + +**What already works — protect it.** Generated consumer code does *not* struct-literal +schema types; endpointgen emits `serde_json::from_str(schema).unwrap()` +(`endpointgen/src/rust.rs:513`). Therefore **adding a field to `EndpointSchema` is +already non-breaking for generated code, as long as it carries `#[serde(default)]`.** +Record this as a load-bearing invariant in the module docs so nobody "optimises" the +generated schema into a struct literal later. + +**Changes in `src/model/`:** + +- `#[non_exhaustive]` on `Type` (`model/types.rs:82`), `Field` (`types.rs:5`), + `EnumVariant` (`types.rs:44`), `EndpointSchema` (`model/endpoint.rs:11`), and + `EndpointErrorSchema`. `Type` is the critical one: it is a plain public enum, so any + future variant (a decimal with precision/scale, a constrained string, a format + carrier) is breaking without this. +- Add a forward-compatible metadata slot, populated by nobody in 2.0: + +```rust +// on both Field and EndpointSchema +/// Emitter-facing annotations (examples, constraints, tags, deprecation, +/// protocol-binding hints). Unused in 2.0; consumed by the OpenAPI/AsyncAPI +/// emitters in 2.1. Unknown keys must round-trip untouched. +#[serde(default, skip_serializing_if = "BTreeMap::is_empty")] +pub meta: BTreeMap, +``` + +- Because `Field`/`EndpointSchema`/`EnumVariant` already have `new*` constructors, keep + them as the supported construction path and add `with_meta`. + +**Known fallout, fix in the same lockstep release:** `#[non_exhaustive]` forbids struct +literals *outside* the defining crate, which breaks endpointgen's +`impl From for EndpointSchema` +(`endpointgen/src/definitions.rs:430`). One-line fix — switch it to +`EndpointSchema::new(...)` plus setters. Nothing in consumer projects constructs these +by literal, so the blast radius is endpointgen only. + +**Acceptance:** `cargo test --all-features`; a round-trip test proving an +`EndpointSchema` JSON blob containing an *unknown* future field deserializes, and that +a populated `meta` map survives serialize → deserialize unchanged. That test is the +contract 2.1 depends on. + +## 4. Phase 3 — transport seam + +New module `src/libs/transport.rs` (in `ws-core`): + +```rust +/// STOLEN SHAPE (tarpc): blanket transport alias. +pub trait Transport: /* as §1a, items = WireMessage */ { ... } + +/// Object-safe session-facing stream. The existing `WsStream` trait keeps this +/// exact API; rename to `MessageStream` with `pub use ... as WsStream` alias. +#[async_trait(?Send)] +pub trait MessageStream: Unpin + Send { + async fn send(&mut self, msg: WireMessage) -> Result<(), StreamError>; + async fn recv(&mut self) -> Option>; +} + +/// Adapter: any Sink+Stream transport is a MessageStream. +pub struct TransportStream(pub T); +#[async_trait(?Send)] +impl MessageStream for TransportStream +where T: Transport + Unpin + Send { ... } +``` + +New module `src/libs/transport/framed.rs` behind new feature +`framed-transport = ["ws-core", "dep:tokio-util", "dep:tokio-serde"]`: + +```rust +/// STOLEN SHAPE (tarpc serde_transport): length-delimited JSON frames over any +/// byte stream. Text frames only (the legacy protocol is JSON either way); +/// Ping/Pong map to empty control frames w/ a 1-byte tag; Close = clean EOF. +pub fn framed_json(io: S) -> impl Transport +where S: AsyncRead + AsyncWrite + Unpin + Send + 'static; +``` + +Frame format: `u32 BE length | u8 kind (0=Text,1=Binary,2=Ping,3=Pong,4=Close) | payload`. +Configure `LengthDelimitedCodec` max frame length from a parameter (default 16 MiB). +Document the format in the module docs — the sibling crate and any non-Rust peer must +be able to implement it. + +**Server entry seam** in `server.rs`: + +```rust +impl WebsocketServer { + /// Public, transport-agnostic entry: runs auth + session for one + /// already-established connection. Generalizes post_upgrade_connection. + pub async fn serve_connection( + self: Arc, + peer: PeerIdentity, + states: Arc, + stream: Box, + auth_protocol: Option, // WS subprotocol today; token handoff for local + ); + + /// Accept loop over any listener (see trait below). `listen()` becomes a + /// thin wrapper: TCP/TLS accept + HTTP upgrade → serve_connection. + pub async fn serve_with(self, listener: L) -> Result<()>; +} + +#[async_trait] +pub trait SessionListener: Send + Sync + 'static { + async fn accept(&self) -> Result<(Box, PeerIdentity)>; +} +``` + +**Client-side seam (same shape, do it here — AgentOne is a client):** + +```rust +impl WsClient { + /// Transport-agnostic constructor: drive the existing request/reply + + /// seq-correlation logic over any MessageStream. Mirrors serve_connection. + pub fn from_stream(stream: Box) -> Self; +} +``` + +The private `enum WsStream` (`client.rs:55`) gains a `Message(Box)` +variant. Both changes are **additive and non-breaking** (the enum is private, +`WsClient::new` is untouched), so this could technically wait — do it now anyway: +Phase 1 already rewrites these exact functions, the A0↔A1 sidecar needs a client over +XPC/UDS/pipe, and the alternative is the sibling crate reimplementing seq correlation, +MCP framing and reconnect. Ship `from_stream` even though no local transport exists yet +— the duplex acceptance test below exercises it. + +Notes: + +- `post_upgrade_connection` refactors to call `serve_connection`; the + upgrader/H1/H2/TLS/shard machinery stays exactly where it is, feeding the same + entry. The shard model is a property of `listen()`, not of `serve_with` — document + that `serve_with` runs single-runtime (fine for local IPC's connection counts). +- **`MessageStream` is `#[async_trait(?Send)]`**, so its futures are not `Send` and + `serve_with`/`serve_connection` must run inside a `LocalSet` (consistent with the + existing `spawn_local` dispatch). State this explicitly in the module docs and in + §7's contract — otherwise sibling-crate authors will discover it via a confusing + `Send` bound error at integration time. +- `AuthController::auth` currently receives the WS subprotocol string; local + transports pass their token via `auth_protocol` — no trait change needed. Document it. + +**Acceptance test (the proof of the whole seam):** an in-process test using +`tokio::io::duplex` + `framed_json` + `serve_connection` that (a) round-trips a legacy +`{method,seq,params}` request through a real registered handler, (b) completes an MCP +`initialize` → `tools/call` on the same connection, (c) never touches a TCP socket, +and (d) drives the client half through `WsClient::from_stream` on the other end of the +duplex — so both seams are proven by the same test. This test is the definition of done +for 2.0's core claim. + +## 5. Phase 4 — request hooks (STOLEN SHAPE: tarpc request_hook) + +New module `src/libs/hooks.rs` (ws-core): + +```rust +#[async_trait(?Send)] +pub trait BeforeRequest: Send + Sync { + /// Runs after role-check, before dispatch. Err → typed error to client, + /// handler never runs. May mutate ctx (e.g. insert verified claims into + /// ctx.extensions). + async fn before( + &self, + ctx: &mut RequestContext, + endpoint: &EndpointSchema, + params: &serde_json::Value, + ) -> Result<(), CustomError>; +} + +#[async_trait(?Send)] +pub trait AfterRequest: Send + Sync { + async fn after(&self, ctx: &RequestContext, endpoint: &EndpointSchema, + outcome: &RequestOutcome); // Ok / PublicErr(code) / InternalErr +} + +impl WebsocketServer { + pub fn add_before_hook(&mut self, hook: impl BeforeRequest + 'static); + pub fn add_after_hook(&mut self, hook: impl AfterRequest + 'static); +} +``` + +Wiring — this is the subtle part, get both paths: + +- `session.rs::handle_message` (legacy path): after `check_roles`, before + `spawn_local(handler.handle(...))`. Hooks run inside the spawned task (they're + async; don't block the session loop). On `Err(custom)`, emit via the existing + `WsResponseError` shape with the hook's code/params. +- `session.rs::handle_mcp_frame` → `McpAction::ToolCall` path: same placement inside + the spawned task, error emitted as MCP tool error (`encode_tool_error`), mirroring + `HandlerError::Public` handling in `handler.rs::handle_mcp`. +- Hooks execute in registration order; first error short-circuits. Store as + `Vec>` on the server, snapshot into the spawned task. + +**Also add `OnConnect` while you are here (cheap, additive):** + +```rust +#[async_trait] +pub trait OnConnect: Send + Sync { + /// Runs after auth, before the session loop. Err → connection refused, + /// no messages exchanged. May populate connection-scoped extensions. + async fn on_connect(&self, peer: &PeerIdentity, ext: &mut Extensions) + -> Result<(), CustomError>; +} +``` + +Strictly additive (new trait + new register method), so it is not a 2.0-or-never item — +but it is ~30 lines next to the hooks you are already writing, and without it every +`BeforeRequest` has to re-read attestation from extensions per request instead of +rejecting an unattested peer once at connect. The XPC spike's finding that rejected +peers are invisible to the listener process makes this the natural place for +attestation telemetry. + +**Acceptance:** test registering a hook that denies method X with a custom code — +assert the exact error frame on the legacy path AND the MCP path; test a hook that +inserts a claim into `ctx.extensions` and a handler that reads it. + +## 6. Phase 5 — release mechanics + +- Version `2.0.0-alpha.1`; CHANGELOG section listing the four breaking items + (`WsConnection.address`; `WsMessage` no longer tungstenite's type — `From` impls + cover type positions but *not* inherent method calls, see §2; `WsStream` renamed + w/ alias; the model types now `#[non_exhaustive]` per §3b) and the compat shims. +- `docs/2.0-migration.md`: per-symbol table (old → new → action), modeled on the + existing mcp-migration.md. +- README: new "Transports" section — framed_json format spec, `serve_with`/ + `serve_connection` examples, one worked UDS example (`examples/uds_echo.rs`, unix + cfg, plain UnixListener + framed_json — no attestation; ~40 lines) proving the seam + end-to-end on a real OS transport. +- endpointgen lockstep (separate repo — note it, don't do it here): + 1. Bump `ENDPOINT_LIBS_REQUIREMENT` to `2.0`. + 2. Fix the `EndpointSchema` struct literal broken by Phase 2b's `#[non_exhaustive]` + (`endpointgen/src/definitions.rs:430` → `EndpointSchema::new(...)` + setters). + 3. **Add a `--check` mode** (borrowed from Oxide's `dropshot-api-manager` + discipline): regenerate every artifact into a temp dir, diff against what is + committed, exit non-zero on drift, and wire it into CI. endpointgen today has + `check_compatibility` for *version* lockstep (`endpointgen/src/main.rs:412`) but + nothing that catches "someone edited the RON and forgot to regenerate." This is + independently worth doing for the Rust/docs/MCP artifacts, and it is the + prerequisite for trusting committed OpenAPI/AsyncAPI documents in 2.1. + +## 7. Contracts for the sibling crate (define here, implement later — NOT in this repo) + +`endpoint-transport-local` will provide `SessionListener`/`Transport` impls: + +| Backend | Rendezvous | Attestation → `Attestation::Verified` | +|---|---|---| +| `uds` (linux) | named socket or inherited socketpair | `SO_PEERPIDFD` + exe SHA-256 (`mechanism: "pidfd-exe-sha256"`); `SO_PEERCRED` fills pid/uid | +| `pipe` (windows) | named pipe (SID DACL, first-instance flag) or inherited handle | DACL admission + impersonation SID check (`"pipe-sid-dacl"`) | +| `xpc` (macos) | private mach service via rama-net-apple-xpc | `PeerSecurityRequirement` (`"xpc-codesign-requirement"`); XpcMessage dict wrapping one `WireMessage` frame per message | + +Phase 1–4 must not require ANY change for these to slot in — that's the design test. +If while implementing you find a needed hook that isn't in this plan (e.g. connection- +close callbacks for attestation telemetry), add it in the same style and note it in +the migration doc. + +## 8. Explicit non-goals for 2.0 + +- No platform attestation code in endpoint-libs (sibling crate). +- No change to the wire protocols (legacy JSON frames + MCP are untouched). +- No mission-token semantics (that's a `BeforeRequest` impl in AgencyZero, not here). +- No client-side *feature* work: `WsClient::new` (TCP/TLS) is untouched and no local + transport ships here. `WsClient::from_stream` **is** in scope (§4) — it is the seam + only, additive, and the sibling crate supplies the actual transports. +- No OpenAPI/AsyncAPI emission — that is 2.1, in endpointgen, and Phase 2b is what + keeps it a minor release. See `PLAN-2.1.md`. +- Don't fix the double-serialization TODO in `handler.rs` unless it falls out free. + +## 9. Order of work & verification loop + +1. Phase 1 (WireMessage, incl. client.rs) → full test suite + both examples. +2. Phase 2 (PeerIdentity) → suite + grep inventory in PR notes. +3. Phase 2b (schema-model future-proofing) → unknown-field round-trip test. Land it + *with* Phase 2 so all model breakage is in one commit series. +4. Phase 3 (transport seam + `WsClient::from_stream`) → the duplex acceptance test + (both halves) + uds example. +5. Phase 4 (hooks, incl. `OnConnect`) → both-path hook tests. +6. Phase 5 (docs/release) → `cargo semver-checks` if available; alpha tag. + +**Forward-compatibility invariants that must survive 2.0** (2.1 depends on all four — +breaking any of them turns the OpenAPI release into a 3.0): + +1. Generated schemas stay JSON-deserialized at runtime, never struct literals. +2. New model fields always carry `#[serde(default)]`. +3. `Type` and the schema structs stay `#[non_exhaustive]`. +4. `Field.meta` / `EndpointSchema.meta` round-trip unknown keys untouched. + +After each phase, run this matrix (NOT `--all-features`, see ground rules): + +```bash +cargo build --no-default-features --features types +cargo build --features ws-core # must not pull tungstenite +cargo build --features full +cargo test --features full # baseline: 65 tests green at 1.9.1 +cargo clippy --all-targets --features full -- -D warnings +cargo all-features test # what CI runs; respects the denylist +``` diff --git a/PLAN-2.1.md b/PLAN-2.1.md new file mode 100644 index 0000000..d7167fe --- /dev/null +++ b/PLAN-2.1.md @@ -0,0 +1,333 @@ +# endpoint-libs 2.1 / endpointgen 2.1 — OpenAPI + AsyncAPI emission (brief for Claude Code) + +Repos: `~/code/endpoint-libs` (2.0.x after `PLAN-2.0.md` lands) and `~/code/endpointgen` +(lockstep). Goal of 2.1: teach the RON pipeline to emit **OpenAPI 3.1** and +**AsyncAPI 3.0** documents as *additional artifacts* alongside the Rust/docs/MCP output +it already produces, so the project collects the standard-format dividends (third-party +client SDKs, hosted docs, spec-driven fuzzing, OpenAPI→MCP bridging) without giving up +the RON source of truth, the role/error model, or the deployed WS protocol. + +**Read first:** `~/code/iaai-27/rpc-crate-survey.md` §A — the research this release +comes from, including why the alternative (adopt an OpenAPI-first framework) was +rejected and why the codegen arrow has to keep pointing RON → artifacts. + +Ground rules: + +- **This is a MINOR release. Nothing here may break a 2.0 consumer.** If you find + yourself needing a breaking change to `Type`, `Field`, or `EndpointSchema`, stop and + re-read §1 — the 2.0 Phase 2b groundwork exists precisely so you don't have to. +- The RON stays the single source of truth. These emitters are **outputs**. Never add + an OpenAPI/AsyncAPI *input* path (that is the "adopt OpenAPI" plan that was + rejected). +- No new required dependencies in endpoint-libs default features. The emitters live in + endpointgen, which may take `serde_yaml` (or emit JSON only — see §4.4). +- Every phase lands green: `cargo clippy --all-features`, `cargo test --all-features`, + and the generated documents validate against a real spec validator (§7). + +--- + +## 1. What 2.0 already gave you (do not redo this work) + +Confirm these hold before starting; if any is false, the 2.0 plan did not land as +written and this release will be breaking: + +| Invariant | Where | Why 2.1 needs it | +|---|---|---| +| Generated schemas are JSON-deserialized at runtime, not struct literals | `endpointgen/src/rust.rs:513` emits `serde_json::from_str(schema)` | New model fields don't break generated code | +| New model fields carry `#[serde(default)]` | `model/endpoint.rs` | Old committed schema JSON still deserializes | +| `Type`, `Field`, `EnumVariant`, `EndpointSchema` are `#[non_exhaustive]` | `model/types.rs`, `model/endpoint.rs` | New `Type` variants / fields stay additive | +| `Field.meta` and `EndpointSchema.meta` round-trip unknown keys | 2.0 Phase 2b | Per-field examples/constraints/tags land without a model change | +| endpointgen has `--check` (regenerate → diff → non-zero on drift) | 2.0 lockstep item 3 | Committed spec documents can be trusted in CI | + +**The single most important existing asset:** `Type::to_json_schema` +(`endpoint-libs/src/model/json_schema.rs:161`) already emits **JSON Schema 2020-12** — +`$defs`, `$ref`, `format: uuid`, `contentEncoding: base64`, `pattern` for blockchain +addresses/hashes, `minimum`/`maximum` for sized ints, `anyOf: [T, null]` for +`Optional`. **OpenAPI 3.1 is a superset of JSON Schema 2020-12**, so these schema +objects drop into an OpenAPI document essentially verbatim. AsyncAPI 3.0 also uses +JSON Schema for payloads. You are not writing a type-to-schema converter — you are +writing two document *envelopes* around an existing one. + +The only structural mismatch is location: `to_json_schema` puts shared definitions in +`#/$defs/X`, while OpenAPI wants `#/components/schemas/X` and AsyncAPI wants +`#/components/schemas/X` too. That is a mechanical `$ref` rewrite (§2.2). + +--- + +## 2. Phase 1 — shared document plumbing (endpoint-libs, additive) + +New module `src/model/api_document.rs`, exported from `model`. This lives in +endpoint-libs (not endpointgen) because the MCP server already needs the same +registry-walking logic at startup, and both emitters plus any future OpenRPC emitter +should share one implementation. + +### 2.1 Collect every shared definition once + +```rust +/// Walks all endpoints in a service, emitting each referenced Struct/Enum exactly +/// once into a component map. Mirrors what `to_mcp_input_schema` does per-endpoint, +/// but hoisted to document scope so `$ref`s are shared across operations. +pub struct SchemaComponents { + pub schemas: BTreeMap, +} + +impl SchemaComponents { + pub fn collect( + endpoints: &[EndpointSchema], + registry: &TypeRegistry, + ) -> Result; +} +``` + +Implementation note: reuse `Type::to_json_schema` with a shared `defs` map across all +endpoints instead of a fresh `BTreeMap` per endpoint (which is what +`to_mcp_input_schema`/`to_mcp_output_schema` do today). Do **not** change those two +methods — MCP tool schemas are self-contained by design and consumers depend on that. + +### 2.2 `$ref` relocation + +```rust +/// Rewrites `#/$defs/X` → `#/components/schemas/X` throughout a schema value. +/// Recursive over objects and arrays; only touches string values under a `$ref` key. +pub fn relocate_refs(value: &mut serde_json::Value, prefix: &str); +``` + +Unit-test this against a deeply nested case (Vec> inside a Struct +field) and a recursive struct — `to_json_schema` reserves a slot to terminate +recursion (`json_schema.rs:211`), so the rewrite must not loop. + +### 2.3 The `meta` passthrough + +Any key in `Field.meta` / `EndpointSchema.meta` that starts with `x-` is copied +verbatim onto the corresponding schema/operation object (both OpenAPI and AsyncAPI +allow arbitrary `x-` extensions). Recognised non-`x-` keys are mapped explicitly: +`example`, `examples`, `deprecated`, `tags`, plus the JSON Schema constraint keywords +(`minimum`, `maximum`, `minLength`, `maxLength`, `pattern`, `enum`). Unrecognised +non-`x-` keys are a **hard error** with the endpoint and field name — silently dropping +metadata is how these documents rot. + +**Acceptance:** unit tests only; no emitter yet. `cargo test -p endpoint-libs`. + +--- + +## 3. Phase 2 — OpenAPI 3.1 emitter (endpointgen) + +New file `endpointgen/src/openapi.rs`, modelled directly on +`endpointgen/src/docs.rs::gen_mcp_tools_json` (`docs.rs:275`) — same registry +construction, same per-service loop, same `docs/` output directory. + +```rust +pub fn gen_openapi(data: &Data) -> eyre::Result<()>; +// writes docs/{service}_openapi.json (and .yaml if the yaml feature is on) +``` + +Call it from `main.rs` next to the existing emitters (`main.rs:83-89`). + +### 3.1 The modelling decision you must make first + +A WS RPC method has no URL. OpenAPI needs paths. **Synthesize them** — do not try to +be clever: + +``` +POST /rpc/{EndpointName} operationId: endpointName (camelCase) + requestBody: application/json → object schema over `parameters` + responses: + 200: application/json → object schema over `returns` + 4xx: application/json → the endpoint's error catalog (§3.3) +``` + +Document prominently in the generated file's `info.description` **and** in +`docs/openapi-README.md` that this is a *projection for tooling purposes*: the real +transport is a persistent WebSocket carrying `{method, seq, params}` frames, and the +authoritative description of that is the AsyncAPI document (§4). Generating an HTTP +client from this and pointing it at the server will not work. This warning is not +optional — an undocumented synthetic path map is worse than no document, because it +looks usable. + +Fields that carry over directly: + +| RON / `EndpointSchema` | OpenAPI | +|---|---| +| `name` | `operationId` (camelCase), path segment | +| `code` | `x-endpoint-code` extension | +| `description` | `summary` (first line) + `description` (full) | +| `parameters` | `requestBody` object schema, non-`Optional` → `required` | +| `returns` | `200` response schema | +| `stream_response` | `x-stream-response` extension + a note in `description` | +| `roles` | `security` + `x-required-roles` (§3.2) | +| `errors` | error responses (§3.3) | +| `frontend_facing` (on the element, not the schema) | `x-frontend-facing`; also drives `--public-only` filtering | + +### 3.2 Roles → security + +Emit one `securitySchemes` entry describing the WS subprotocol auth token: + +```json +"securitySchemes": { + "sessionToken": { "type": "apiKey", "in": "header", "name": "Sec-WebSocket-Protocol", + "description": "Auth token passed as WS subprotocol; see AuthController." } +} +``` + +Each operation gets `"security": [{"sessionToken": []}]` plus +`"x-required-roles": ["Admin", "User"]` from `schema.roles`. OpenAPI has no native +role concept — do not attempt to encode roles as scopes, it misleads generators into +emitting OAuth2 flows that do not exist. + +### 3.3 Error catalog → responses + +`EndpointSchema.errors` (`Vec`) plus the global error-code catalog +(`endpointgen/src/error_codes.rs`) become response entries. One response object per +distinct HTTP-ish class is enough — the wire protocol has no status codes, so: + +- `default` response → the standard error envelope schema (code, message, params), + with `x-error-codes` listing the codes this endpoint may return, each with its + description from the catalog. + +Do not invent per-code HTTP statuses. The envelope is the contract. + +### 3.4 Filtering + +`--public-only` (or a config key) emits only `frontend_facing` endpoints, for the +document you would hand to a third party. Default emits everything. + +**Acceptance:** the emitted document validates (§7); an endpoint with a recursive +struct, an enum ref, an optional vec, and two error codes round-trips into readable +schemas; `--public-only` drops exactly the non-frontend-facing operations. + +--- + +## 4. Phase 3 — AsyncAPI 3.0 emitter (endpointgen) + +New file `endpointgen/src/asyncapi.rs`. **This is the document that actually describes +your protocol** — the OpenAPI one is a tooling projection, this one is the truth. + +```rust +pub fn gen_asyncapi(data: &Data) -> eyre::Result<()>; +// writes docs/{service}_asyncapi.json +``` + +### 4.1 Channel and operation model + +AsyncAPI 3.0 separates channels (where messages flow), operations (send/receive), and +messages (payload shapes). Map as: + +- **One channel** per service: `ws`, with `address: "/"` and a `ws` binding recording + the subprotocol used for auth. +- **Two operations**: `sendRequest` (client → server, `action: send`) and + `receiveResponse` (server → client, `action: receive`). +- **Messages**: `Request`, `Response`, `Error`, and — because they share the socket — + `McpJsonRpc`. The `Request` payload is the envelope: + +```json +{ "type": "object", + "properties": { + "method": { "type": "integer", "description": "endpoint code" }, + "seq": { "type": "integer" }, + "params": { "oneOf": [ /* $ref per endpoint parameter schema */ ] } }, + "required": ["method", "seq", "params"] } +``` + +Use `oneOf` over the per-endpoint parameter schemas with a `discriminator` on `method` +if the generator you test with supports it; otherwise emit the `oneOf` plus an +`x-method-map` extension mapping code → schema name. Note which you did in the file +header. + +### 4.2 Per-endpoint detail + +Each endpoint contributes a `components.messages.{Name}Request` / +`{Name}Response` pair with the same descriptions, roles extensions, and error lists as +the OpenAPI operations. Reuse `SchemaComponents` from §2.1 — both documents must +reference **identical** schema objects, and a test should assert that +(`assert_eq!(openapi.components.schemas, asyncapi.components.schemas)`). + +### 4.3 The framed_json binding + +2.0 defines a length-delimited frame format for non-WS transports +(`u32 BE length | u8 kind | payload`, `PLAN-2.0.md` §4). Record it in the AsyncAPI +document as a second channel entry with a custom `x-framing` extension describing the +byte layout, so a non-Rust peer implementing the local transport has one authoritative +reference. This is the only place that format is machine-readable. + +### 4.4 YAML + +JSON is mandatory; YAML is nice-to-have for humans. If you add it, put `serde_yaml` +behind an endpointgen feature — do not make it a default dependency for a cosmetic +output. + +**Acceptance:** validates against an AsyncAPI 3.0 validator (§7); the shared-components +equality test passes; a hand-written peer can reconstruct the frame layout from +`x-framing` alone. + +--- + +## 5. Phase 4 — wire it into the build and CI + +- `main.rs`: call `openapi::gen_openapi` and `asyncapi::gen_asyncapi` after + `docs::gen_mcp_tools_json` (`main.rs:87`). +- Both documents are **committed artifacts**, like the existing generated Rust/docs. +- Extend the 2.0 `--check` mode to cover them: regenerate → diff → non-zero exit. + This is the dropshot `dropshot-api-manager` discipline; the whole value of a + committed spec is that CI proves it matches the RON. +- `docs/openapi-README.md`: what each document is, the synthetic-path warning (§3.1), + and the three consumption recipes in §6. + +--- + +## 6. What this unlocks (validate at least the first one) + +1. **OpenAPI → MCP bridging**: point `rmcp-openapi` + (`gitlab.com/lx-industries/rmcp-openapi`) at the emitted document and confirm the + tool list matches endpoint-libs' own `tools/list` output for the same service. + **This is the highest-value check in the release** — a mismatch means the hand-rolled + MCP metadata and the emitted spec disagree, and one of them is lying to an agent. +2. **Third-party client SDKs**: `openapi-generator` (any of 50+ languages) against the + `--public-only` document. Expect the synthetic paths to be wrong for real use — + that is exactly why §3.1's warning exists; validate that it *generates*, not that + it *connects*. +3. **Spec-driven fuzzing**: Schemathesis against the document is the interesting one + long-term, but it needs an HTTP surface the server does not have. Note it as future + work behind a REST adapter; do not build the adapter here. + +--- + +## 7. Validation tooling + +- OpenAPI 3.1: `redocly lint` or the `oas3` Rust crate for a parse check in a test. + Prefer a real linter in CI over a parse check. +- AsyncAPI 3.0: the official `@asyncapi/cli validate`. +- Both: a test that every `$ref` in the document resolves against + `components.schemas` (catches the §2.2 rewrite regressing) and that no `$defs` key + survives anywhere. + +--- + +## 8. Explicit non-goals for 2.1 + +- **No OpenAPI/AsyncAPI as input.** No spec → Rust codegen, ever, in this direction. + The RON is the source of truth. +- **No runtime behaviour change.** endpoint-libs serves the same frames; these are + build-time artifacts. Nothing in `src/libs/` changes except the additive + `model/api_document.rs`. +- **No REST/HTTP adapter.** The synthetic paths are for tooling, not for serving. If a + real REST surface is ever wanted, that is its own release with its own plan. +- **No OpenRPC emitter yet.** It is arguably the best-fitting standard (MCP is + JSON-RPC, and so is half your protocol) but Rust/ecosystem tooling for it is thin — + see `rpc-crate-survey.md` §A.5. Revisit if OpenRPC tooling matures; the + `SchemaComponents` plumbing in §2 is deliberately emitter-agnostic so adding it later + is one more file. +- **No breaking changes.** If one seems necessary, it belongs in 3.0 and needs its own + plan — not a quiet bump here. + +## 9. Order of work & verification loop + +1. Phase 1 (`api_document.rs` + `relocate_refs` + meta passthrough) → unit tests. +2. Phase 2 (OpenAPI emitter) → validator + recursive/enum/optional round-trip test. +3. Phase 3 (AsyncAPI emitter) → validator + shared-components equality test. +4. Phase 4 (build wiring + `--check` extension + docs). +5. Consumption check §6.1 (`rmcp-openapi` tool-list parity) — treat a mismatch as a + release blocker, not a curiosity. + +After each phase: `cargo clippy --all-targets --all-features -- -D warnings`, +`cargo test --all-features` in both repos, and `endpointgen --check` clean on a real +service RON (use `api.support.cafe` or `web3.trading-backend` as the corpus). diff --git a/src/libs/ws/client.rs b/src/libs/ws/client.rs index 0ade068..3a2ad72 100644 --- a/src/libs/ws/client.rs +++ b/src/libs/ws/client.rs @@ -18,12 +18,13 @@ use tokio_rustls::TlsConnector; use tokio_tungstenite::MaybeTlsStream; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::connect_async; -use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::Message as TMessage; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::protocol::Role; use tracing::*; use crate::libs::log::LogLevel; +use crate::libs::ws::WireMessage as Message; use crate::libs::ws::{WsLogResponse, WsRequest, WsRequestGeneric, WsResponseGeneric}; // --------------------------------------------------------------------------- @@ -106,6 +107,9 @@ impl WsClient { // --- Private stream helpers ------------------------------------------- async fn stream_send(&mut self, msg: Message) -> Result<()> { + // Backend edge: the client speaks WireMessage; tungstenite's type exists + // only inside these helpers. + let msg: TMessage = msg.into(); match &mut self.stream { WsStream::H1(s) => s.send(msg).await?, WsStream::H2(s) => s.send(msg).await?, @@ -116,10 +120,11 @@ impl WsClient { async fn stream_next( &mut self, ) -> Option> { - match &mut self.stream { + let next = match &mut self.stream { WsStream::H1(s) => s.next().await, WsStream::H2(s) => s.next().await, - } + }; + next.map(|res| res.map(Into::into)) } async fn stream_close(&mut self) -> Result<()> { diff --git a/src/libs/ws/message.rs b/src/libs/ws/message.rs index 622bbaf..e7b2dfe 100644 --- a/src/libs/ws/message.rs +++ b/src/libs/ws/message.rs @@ -1,29 +1,74 @@ -#[cfg(feature = "ws")] -pub use tokio_tungstenite::tungstenite::Message as WsMessage; +//! The canonical, backend-independent WebSocket message type. +//! +//! # 2.0 change +//! +//! Before 2.0, `WsMessage` was a *re-export of `tungstenite::Message`* whenever the +//! `ws` feature was on, which leaked a backend type through the session, toolbox, +//! subscription, push and connection layers. [`WireMessage`] is now the canonical type +//! in every configuration; the tungstenite (and any future) backend converts at its own +//! edge via the `From` impls in that backend's module. +//! +//! `pub type WsMessage = WireMessage` remains as a compatibility alias, but note it +//! only covers *type positions*. Code that called tungstenite's inherent methods +//! (`.into_text()`, `.into_data()`, `.is_close()`, …) must migrate — see +//! `docs/2.0-migration.md`. -#[cfg(not(feature = "ws"))] -pub use inner::*; +/// A WebSocket close frame: status code plus a human-readable reason. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CloseFrame { + pub code: u16, + pub reason: String, +} + +/// A protocol-level message, independent of any WebSocket backend. +/// +/// This is the item type carried by [`MessageStream`](crate::libs::ws::WsStream) and, +/// from 2.0 on, by any [`Transport`](crate::libs::ws::Transport) — including non-WS +/// local transports (Unix sockets, named pipes, XPC), where `Ping`/`Pong`/`Close` are +/// mapped onto whatever that transport's control mechanism is. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum WireMessage { + Text(String), + Binary(Vec), + Ping(Vec), + Pong(Vec), + Close(Option), +} + +/// Compatibility alias for the pre-2.0 name. +/// +/// Covers type positions only — not tungstenite's inherent methods. +pub type WsMessage = WireMessage; -#[cfg(not(feature = "ws"))] -mod inner { - #[derive(Debug, Clone, PartialEq, Eq)] - pub struct CloseFrame { - pub code: u16, - pub reason: String, +impl From for WireMessage { + fn from(s: String) -> Self { + Self::Text(s) } +} - #[derive(Debug, Clone, PartialEq, Eq)] - pub enum WsMessage { - Text(String), - Binary(Vec), - Ping(Vec), - Pong(Vec), - Close(Option), +impl From<&str> for WireMessage { + fn from(s: &str) -> Self { + Self::Text(s.to_owned()) } +} - impl From for WsMessage { - fn from(s: String) -> Self { - WsMessage::Text(s) +impl WireMessage { + /// Borrow the payload as text, if this message carries UTF-8. + /// + /// `Text` always succeeds; `Binary` succeeds when the bytes are valid UTF-8 + /// (the legacy protocol and MCP both accept either framing). Control frames + /// return `None`. + pub fn as_text(&self) -> Option<&str> { + match self { + Self::Text(t) => Some(t.as_str()), + Self::Binary(b) => std::str::from_utf8(b).ok(), + _ => None, } } + + /// True for `Close`. + pub fn is_close(&self) -> bool { + matches!(self, Self::Close(_)) + } } diff --git a/src/libs/ws/tungstenite.rs b/src/libs/ws/tungstenite.rs index 6843b41..500b39e 100644 --- a/src/libs/ws/tungstenite.rs +++ b/src/libs/ws/tungstenite.rs @@ -1,3 +1,4 @@ +mod message; pub mod upgrader; pub use upgrader::HyperTungsteniteUpgrader; diff --git a/src/libs/ws/tungstenite/message.rs b/src/libs/ws/tungstenite/message.rs new file mode 100644 index 0000000..431e799 --- /dev/null +++ b/src/libs/ws/tungstenite/message.rs @@ -0,0 +1,82 @@ +//! Conversions between the canonical [`WireMessage`] and tungstenite's `Message`. +//! +//! These are the *only* place the tungstenite message type is allowed to meet the rest +//! of the crate. Everything above the backend edge — session, toolbox, subs, push, +//! conn, server, client — deals in [`WireMessage`] exclusively. + +use tokio_tungstenite::tungstenite::Message as TMessage; +use tokio_tungstenite::tungstenite::protocol::frame::CloseFrame as TCloseFrame; + +use crate::libs::ws::message::{CloseFrame, WireMessage}; + +impl From for TMessage { + fn from(msg: WireMessage) -> Self { + match msg { + WireMessage::Text(t) => Self::Text(t.into()), + WireMessage::Binary(b) => Self::Binary(b.into()), + WireMessage::Ping(b) => Self::Ping(b.into()), + WireMessage::Pong(b) => Self::Pong(b.into()), + WireMessage::Close(frame) => Self::Close(frame.map(|f| TCloseFrame { + code: f.code.into(), + reason: f.reason.into(), + })), + } + } +} + +impl From for WireMessage { + fn from(msg: TMessage) -> Self { + match msg { + TMessage::Text(t) => Self::Text(t.as_str().to_owned()), + TMessage::Binary(b) => Self::Binary(b.into()), + TMessage::Ping(b) => Self::Ping(b.into()), + TMessage::Pong(b) => Self::Pong(b.into()), + TMessage::Close(frame) => Self::Close(frame.map(|f| CloseFrame { + code: f.code.into(), + reason: f.reason.as_str().to_owned(), + })), + // tungstenite's `Frame` variant is only produced by its low-level frame + // API, which this crate never uses. Map it to an empty binary payload + // rather than panicking: an unexpected raw frame is not worth aborting a + // live session over, and the session layer will simply ignore it. + TMessage::Frame(_) => Self::Binary(Vec::new()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_trips_every_variant() { + let cases = vec![ + WireMessage::Text("hello".into()), + WireMessage::Binary(vec![1, 2, 3]), + WireMessage::Ping(vec![4]), + WireMessage::Pong(vec![5]), + WireMessage::Close(None), + WireMessage::Close(Some(CloseFrame { + code: 1000, + reason: "bye".into(), + })), + ]; + for case in cases { + let there: TMessage = case.clone().into(); + let back: WireMessage = there.into(); + assert_eq!(case, back, "round trip changed the message"); + } + } + + #[test] + fn tungstenite_raw_frame_degrades_to_empty_binary() { + use tokio_tungstenite::tungstenite::protocol::frame::Frame; + use tokio_tungstenite::tungstenite::protocol::frame::coding::{Data, OpCode}; + let raw = TMessage::Frame(Frame::message( + bytes::Bytes::from_static(b"x"), + OpCode::Data(Data::Binary), + true, + )); + assert_eq!(WireMessage::from(raw), WireMessage::Binary(Vec::new())); + } +} diff --git a/src/libs/ws/tungstenite/upgrader.rs b/src/libs/ws/tungstenite/upgrader.rs index 50b7ee4..5455c0e 100644 --- a/src/libs/ws/tungstenite/upgrader.rs +++ b/src/libs/ws/tungstenite/upgrader.rs @@ -350,11 +350,16 @@ struct HyperWsStream { #[async_trait(?Send)] impl WsStream for HyperWsStream { async fn send(&mut self, msg: Message) -> Result<(), StreamError> { - self.inner.send(msg).await.map_err(map_err) + // Backend edge: convert the canonical WireMessage into tungstenite's type + // here and nowhere else. + self.inner.send(msg.into()).await.map_err(map_err) } async fn recv(&mut self) -> Option> { - self.inner.next().await.map(|r| r.map_err(map_err)) + self.inner + .next() + .await + .map(|r| r.map(Into::into).map_err(map_err)) } } From 5f1aad56965d22706e523546f1c2d565c04e2c3c Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:17:59 +0700 Subject: [PATCH 2/6] feat(2.0)!: replace SocketAddr peers with PeerIdentity + Extensions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bare SocketAddr only describes a TCP/TLS peer. Local transports (Unix sockets, named pipes, XPC) identify peers by process and by verified code identity, and that has to reach the request context for authz and logging. - New `libs::peer`: PeerIdentity (Network/Local/Unknown, #[non_exhaustive]), LocalPeer{pid,uid,attestation}, Attestation (None/Verified{mechanism,subject}). - `WsConnection.address: SocketAddr` → `WsConnection.peer: PeerIdentity`, with a #[deprecated] `address()` accessor returning a loopback placeholder for non-network peers. - `RequestContext` keeps `ip_addr` (populated via `peer.ip_addr()`, so logging consumers keep compiling) and gains `peer` + `extensions`. - Both types gain `extensions: Extensions`, a type-keyed map implemented here rather than depending on `http`. Extensions stores `Box` rather than `Box` because RequestContext derives Clone and consumers rely on it; values must therefore be Clone, the same trade http::Extensions makes. Note the three explicit derefs marked in that module: `Box` itself satisfies the blanket impl's bounds, so `boxed.as_any()` / `self.clone_box()` silently resolve to the box's own impl — the former never downcasts, the latter recurses into a stack overflow. Both were caught by the tests in this commit. Sites updated (the plan's grep inventory): server.rs construction + logging, session.rs run/handle_message, headers.rs real-IP context, toolbox.rs RequestContext::from_conn, and the ws-echo example (now the reference migration for `conn.address` → `conn.peer`). Verified: types / ws-core / full build, clippy clean, 72 tests green. Co-Authored-By: Claude Fable 5 --- examples/ws-echo/main.rs | 5 +- src/libs.rs | 1 + src/libs/peer.rs | 314 +++++++++++++++++++++++++++++++++++++++ src/libs/ws/basics.rs | 17 ++- src/libs/ws/headers.rs | 4 +- src/libs/ws/server.rs | 6 +- src/libs/ws/session.rs | 4 +- src/libs/ws/toolbox.rs | 16 +- 8 files changed, 359 insertions(+), 8 deletions(-) create mode 100644 src/libs/peer.rs diff --git a/examples/ws-echo/main.rs b/examples/ws-echo/main.rs index 59f6ec4..a341daa 100644 --- a/examples/ws-echo/main.rs +++ b/examples/ws-echo/main.rs @@ -179,7 +179,10 @@ impl AuthController for AllowAllAuthController { let conn_id = conn.connection_id; tracing::info!( conn_id = %conn_id, - ip = %conn.address, + // 2.0: `conn.address` (SocketAddr) → `conn.peer` (PeerIdentity), + // which also describes local transports. Display is compact and + // includes attestation for local peers. + peer = %conn.peer, header_len = header.len(), "New connection — granting role 1 (allow-all auth)" ); diff --git a/src/libs.rs b/src/libs.rs index 6d9e339..077dca3 100644 --- a/src/libs.rs +++ b/src/libs.rs @@ -5,6 +5,7 @@ pub mod log_reader; pub mod scheduler; #[cfg(feature = "signal")] pub mod signal; +pub mod peer; pub mod types; pub mod utils; #[cfg(feature = "ws-core")] diff --git a/src/libs/peer.rs b/src/libs/peer.rs new file mode 100644 index 0000000..8fbe20c --- /dev/null +++ b/src/libs/peer.rs @@ -0,0 +1,314 @@ +//! Peer identity for a connection, independent of transport. +//! +//! Before 2.0 a connection's peer was a bare `SocketAddr`, which only makes sense for +//! TCP/TLS. Local transports (Unix sockets, Windows named pipes, macOS XPC) identify +//! peers by process and by *code identity* — a codesign requirement, an executable +//! digest, a SID — and that has to survive all the way into the request context so +//! authorization and logging can see it. +//! +//! [`PeerIdentity`] is `#[non_exhaustive]`: new transports may add variants without a +//! breaking release. + +use std::any::{Any, TypeId}; +use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + +/// Who is on the other end of a connection. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum PeerIdentity { + /// A TCP/TLS network peer — the pre-2.0 behaviour. + Network(SocketAddr), + /// A same-machine peer reached over a local transport. + Local(LocalPeer), + /// Transport could not determine a peer (in-process tests, exotic transports). + Unknown, +} + +/// A same-machine peer: OS-level process identity plus whatever code identity the +/// transport was able to verify. +#[derive(Debug, Clone)] +pub struct LocalPeer { + pub pid: Option, + /// Effective uid. Unix only — always `None` on Windows. + pub uid: Option, + pub attestation: Attestation, +} + +/// Whether — and how — the transport verified the peer's *code* identity. +/// +/// This is deliberately separate from `pid`/`uid`: knowing which process connected is +/// not the same as knowing it runs the binary you expect. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum Attestation { + /// The transport did not verify code identity. Treat the peer as untrusted. + None, + /// The OS or the transport verified code identity before the connection was + /// handed over. + Verified { + /// How it was verified, e.g. `"xpc-codesign-requirement"`, + /// `"pidfd-exe-sha256"`, `"pipe-sid-dacl"`. + mechanism: &'static str, + /// What matched: the requirement string, digest, or SID. + subject: String, + }, +} + +impl Attestation { + /// True only for [`Attestation::Verified`]. + pub fn is_verified(&self) -> bool { + matches!(self, Self::Verified { .. }) + } +} + +impl PeerIdentity { + /// Best-effort IP for logging and for the pre-2.0 `ip_addr` field. + /// + /// Local and unknown peers report loopback — they have no IP, and loopback is + /// both truthful about locality and safe for code that only formats this value. + pub fn ip_addr(&self) -> IpAddr { + match self { + Self::Network(addr) => addr.ip(), + Self::Local(_) | Self::Unknown => IpAddr::V4(Ipv4Addr::LOCALHOST), + } + } + + /// The socket address for network peers; a loopback placeholder otherwise. + /// + /// Provided for the deprecated `WsConnection::address` accessor. Prefer matching + /// on the enum. + pub fn socket_addr(&self) -> SocketAddr { + match self { + Self::Network(addr) => *addr, + Self::Local(_) | Self::Unknown => SocketAddr::from(([127, 0, 0, 1], 0)), + } + } + + /// Attestation for local peers; `None` for network peers (TLS client certs are + /// not modelled here). + pub fn attestation(&self) -> Option<&Attestation> { + match self { + Self::Local(peer) => Some(&peer.attestation), + _ => None, + } + } + + /// Compact one-line form for logs: `1.2.3.4:5678`, `local(pid=42,xpc-codesign-requirement)`. + pub fn display(&self) -> String { + match self { + Self::Network(addr) => addr.to_string(), + Self::Local(peer) => { + let pid = peer + .pid + .map_or_else(|| "?".to_owned(), |pid| pid.to_string()); + match &peer.attestation { + Attestation::Verified { mechanism, .. } => { + format!("local(pid={pid},{mechanism})") + } + Attestation::None => format!("local(pid={pid},unattested)"), + } + } + Self::Unknown => "unknown".to_owned(), + } + } +} + +impl std::fmt::Display for PeerIdentity { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.display()) + } +} + +impl From for PeerIdentity { + fn from(addr: SocketAddr) -> Self { + Self::Network(addr) + } +} + +// --------------------------------------------------------------------------- +// Extensions +// --------------------------------------------------------------------------- + +/// Object-safe `Any` that can also clone itself. +/// +/// `RequestContext` derives `Clone` and consumers rely on that, so a plain +/// `Box` map cannot live inside it. Requiring `Clone` on stored values (the +/// same trade `http::Extensions` makes) keeps the containing types cloneable. +trait CloneAny: Any + Send + Sync { + fn clone_box(&self) -> Box; + fn as_any(&self) -> &dyn Any; + fn as_any_mut(&mut self) -> &mut dyn Any; + fn into_any(self: Box) -> Box; +} + +impl CloneAny for T { + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + fn as_any(&self) -> &dyn Any { + self + } + fn as_any_mut(&mut self) -> &mut dyn Any { + self + } + fn into_any(self: Box) -> Box { + self + } +} + +impl Clone for Box { + fn clone(&self) -> Self { + // `(**self)` is load-bearing. `Box` satisfies the blanket impl's + // bounds, so `self.clone_box()` would resolve to the box's own `clone_box`, + // which calls `clone` — infinite recursion and a stack overflow. Deref to the + // trait object so the inner value's `clone_box` runs instead. + (**self).clone_box() + } +} + +/// A type-keyed map for attaching arbitrary data to a connection or request. +/// +/// Modelled on `http::Extensions`, implemented here to avoid the dependency. Used to +/// carry attestation on a connection and verified mission-token claims on a request, +/// without either concept leaking into this crate's core types. +/// +/// Stored values must be `Clone` so that the containing `RequestContext` stays +/// `Clone`. +#[derive(Default, Clone)] +pub struct Extensions { + map: HashMap>, +} + +impl Extensions { + pub fn new() -> Self { + Self::default() + } + + /// Insert a value, returning the previous one of the same type, if any. + pub fn insert(&mut self, value: T) -> Option { + self.map + .insert(TypeId::of::(), Box::new(value)) + .and_then(|prev| prev.into_any().downcast().ok().map(|boxed| *boxed)) + } + + pub fn get(&self) -> Option<&T> { + // `(**boxed)` is load-bearing: `Box` itself satisfies + // `Any + Send + Sync + Clone`, so it matches the blanket impl below and + // `boxed.as_any()` would resolve to the *box's* impl — yielding a `dyn Any` + // whose concrete type is the box, which never downcasts to `T`. + self.map + .get(&TypeId::of::()) + .and_then(|boxed| (**boxed).as_any().downcast_ref()) + } + + pub fn get_mut(&mut self) -> Option<&mut T> { + // See the note in `get` — the explicit deref is required here too. + self.map + .get_mut(&TypeId::of::()) + .and_then(|boxed| (**boxed).as_any_mut().downcast_mut()) + } + + pub fn remove(&mut self) -> Option { + self.map + .remove(&TypeId::of::()) + .and_then(|boxed| boxed.into_any().downcast().ok().map(|b| *b)) + } + + pub fn contains(&self) -> bool { + self.map.contains_key(&TypeId::of::()) + } + + pub fn is_empty(&self) -> bool { + self.map.is_empty() + } + + pub fn len(&self) -> usize { + self.map.len() + } +} + +impl std::fmt::Debug for Extensions { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Values are `dyn Any` and cannot be formatted; report the count so logs + // still show whether anything was attached. + f.debug_struct("Extensions") + .field("len", &self.map.len()) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn network_peer_reports_its_own_ip() { + let addr: SocketAddr = "203.0.113.7:9000".parse().unwrap(); + let peer = PeerIdentity::Network(addr); + assert_eq!(peer.ip_addr(), addr.ip()); + assert_eq!(peer.socket_addr(), addr); + assert_eq!(peer.display(), "203.0.113.7:9000"); + assert!(peer.attestation().is_none()); + } + + #[test] + fn local_peer_reports_loopback_and_keeps_attestation() { + let peer = PeerIdentity::Local(LocalPeer { + pid: Some(42), + uid: Some(501), + attestation: Attestation::Verified { + mechanism: "xpc-codesign-requirement", + subject: "identifier agentone".to_owned(), + }, + }); + assert_eq!(peer.ip_addr(), IpAddr::V4(Ipv4Addr::LOCALHOST)); + assert!(peer.attestation().unwrap().is_verified()); + assert_eq!(peer.display(), "local(pid=42,xpc-codesign-requirement)"); + } + + #[test] + fn unattested_local_peer_is_visible_as_such() { + let peer = PeerIdentity::Local(LocalPeer { + pid: None, + uid: None, + attestation: Attestation::None, + }); + assert!(!peer.attestation().unwrap().is_verified()); + assert_eq!(peer.display(), "local(pid=?,unattested)"); + } + + #[test] + fn extensions_survive_a_clone() { + #[derive(Debug, Clone, PartialEq)] + struct Claims(&'static str); + + let mut ext = Extensions::new(); + ext.insert(Claims("mission-1")); + // RequestContext derives Clone; extensions must come along intact. + let copy = ext.clone(); + assert_eq!(copy.get::(), Some(&Claims("mission-1"))); + } + + #[test] + fn extensions_round_trip_by_type() { + #[derive(Debug, Clone, PartialEq)] + struct Claims(&'static str); + + let mut ext = Extensions::new(); + assert!(ext.is_empty()); + assert!(ext.get::().is_none()); + + assert!(ext.insert(Claims("mission-1")).is_none()); + assert_eq!(ext.get::(), Some(&Claims("mission-1"))); + assert!(ext.contains::()); + + // Re-inserting returns the previous value rather than silently dropping it. + let prev = ext.insert(Claims("mission-2")); + assert_eq!(prev, Some(Claims("mission-1"))); + assert_eq!(ext.get::(), Some(&Claims("mission-2"))); + + assert_eq!(ext.remove::(), Some(Claims("mission-2"))); + assert!(ext.is_empty()); + } +} diff --git a/src/libs/ws/basics.rs b/src/libs/ws/basics.rs index 9c98a1f..9e028cb 100644 --- a/src/libs/ws/basics.rs +++ b/src/libs/ws/basics.rs @@ -10,6 +10,7 @@ use std::sync::Arc; use std::sync::atomic::AtomicU64; use crate::libs::error_code::ErrorCode; +use crate::libs::peer::{Extensions, PeerIdentity}; use crate::libs::handler::RequestHandlerErased; use crate::libs::log::{CustomEyreHandler, LogLevel}; use crate::libs::toolbox::{CustomError, RequestContext}; @@ -48,7 +49,12 @@ pub struct WsConnection { pub connection_id: ConnectionId, pub user_id: AtomicU64, pub roles: Arc>>>, - pub address: SocketAddr, + /// Who is on the other end. Replaces the pre-2.0 `address: SocketAddr`, which + /// could not describe a local (Unix socket / named pipe / XPC) peer. + pub peer: PeerIdentity, + /// Connection-scoped data. Local transports attach attestation details here; + /// `OnConnect` hooks may attach anything else. + pub extensions: Extensions, pub log_id: u64, } impl WsConnection { @@ -56,6 +62,15 @@ impl WsConnection { self.user_id.load(std::sync::atomic::Ordering::Acquire) } + /// The peer's socket address. + /// + /// Returns a loopback placeholder (`127.0.0.1:0`) for local and unknown peers, + /// which have no socket address at all. + #[deprecated(note = "use `.peer` — a SocketAddr cannot describe a local peer")] + pub fn address(&self) -> SocketAddr { + self.peer.socket_addr() + } + pub fn get_roles(&self) -> Arc> { self.roles.read().clone() } diff --git a/src/libs/ws/headers.rs b/src/libs/ws/headers.rs index 8516b9a..b0a6e74 100644 --- a/src/libs/ws/headers.rs +++ b/src/libs/ws/headers.rs @@ -227,7 +227,9 @@ impl AuthController for EndpointAuthController { method: endpoint.schema.code, log_id: conn.log_id, roles: roles.clone(), - ip_addr: conn.address.ip(), + ip_addr: conn.peer.ip_addr(), + peer: conn.peer.clone(), + extensions: conn.extensions.clone(), }; endpoint .handler diff --git a/src/libs/ws/server.rs b/src/libs/ws/server.rs index e2ee0b1..035f4d1 100644 --- a/src/libs/ws/server.rs +++ b/src/libs/ws/server.rs @@ -17,6 +17,7 @@ use tracing::*; #[cfg(feature = "ws")] use crate::libs::error_code::ErrorCode; use crate::libs::handler::{RequestHandler, RequestHandlerErased}; +use crate::libs::peer::{Extensions, PeerIdentity}; use crate::libs::toolbox::{ArcToolbox, RequestContext, TOOLBOX, Toolbox}; #[cfg(feature = "ws")] use crate::libs::utils::{get_conn_id, get_log_id}; @@ -195,7 +196,8 @@ impl WebsocketServer { connection_id: get_conn_id(), user_id: Default::default(), roles: Arc::new(RwLock::new(Arc::new(Vec::new()))), - address: addr, + peer: PeerIdentity::Network(addr), + extensions: Extensions::new(), log_id: get_log_id(), }); debug!( @@ -239,7 +241,7 @@ impl WebsocketServer { stream: Box, rx: mpsc::Receiver, ) { - let addr = conn.address; + let addr = conn.peer.display(); let context = RequestContext::from_conn(&conn); let conn_id = context.connection_id; diff --git a/src/libs/ws/session.rs b/src/libs/ws/session.rs index 273d6b4..2e89c4f 100644 --- a/src/libs/ws/session.rs +++ b/src/libs/ws/session.rs @@ -44,7 +44,7 @@ impl WsClientSession { } pub async fn run(mut self) { - let addr = self.conn_info.address; + let addr = self.conn_info.peer.display(); let conn_id = self.conn_info.connection_id; if let Err(err) = self.run_loop().await { error!( @@ -58,7 +58,7 @@ impl WsClientSession { } fn handle_message(&mut self, msg: Message) -> Result { - let addr = &self.conn_info.address; + let addr = self.conn_info.peer.display(); let mut context = RequestContext::from_conn(&self.conn_info); // MCP: route JSON-RPC 2.0 frames to the MCP adapter when enabled. diff --git a/src/libs/ws/toolbox.rs b/src/libs/ws/toolbox.rs index 58ad0a3..3e07be6 100644 --- a/src/libs/ws/toolbox.rs +++ b/src/libs/ws/toolbox.rs @@ -1,3 +1,4 @@ +use crate::libs::peer::{Extensions, PeerIdentity}; use crate::libs::ws::WsMessage as Message; use dashmap::DashMap; use eyre::Result; @@ -100,7 +101,16 @@ pub struct RequestContext { pub method: u32, pub log_id: u64, pub roles: Arc>, + /// Best-effort IP, kept for compatibility: most consumers only log it. + /// + /// Populated from [`PeerIdentity::ip_addr`], so local peers report loopback. + /// Prefer [`Self::peer`] when the distinction matters. pub ip_addr: IpAddr, + /// Who issued this request. Carries attestation for local transports. + pub peer: PeerIdentity, + /// Request-scoped data. `BeforeRequest` hooks attach verified claims here; + /// handlers read them back. + pub extensions: Extensions, } impl RequestContext { @@ -113,6 +123,8 @@ impl RequestContext { log_id: 0, roles: Arc::new(Vec::new()), ip_addr: IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), + peer: PeerIdentity::Unknown, + extensions: Extensions::new(), } } pub fn from_conn(conn: &WsConnection) -> Self { @@ -124,7 +136,9 @@ impl RequestContext { method: 0, log_id: conn.log_id, roles, - ip_addr: conn.address.ip(), + ip_addr: conn.peer.ip_addr(), + peer: conn.peer.clone(), + extensions: conn.extensions.clone(), } } } From 82687dc8aaf64d2b0cb1c0c47a748c559661b9dc Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:20:30 +0700 Subject: [PATCH 3/6] feat(2.0)!: future-proof the schema model so OpenAPI can land in 2.1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenAPI/AsyncAPI emission (PLAN-2.1.md) is an endpointgen feature that touches no runtime code — but only stays a MINOR release if the schema model can absorb new information without breaking. Today it cannot. This is the ~20-line insurance premium, and it is only free while 2.0 is already breaking. - #[non_exhaustive] on Type, Field, EnumVariant, EndpointSchema and EndpointErrorSchema. Type matters most: it is a plain public enum, so any future variant (a decimal with precision/scale, a constrained string, a format carrier) would otherwise be breaking. - New `meta` slot on Field and EndpointSchema, plus with_meta setters. Empty in 2.0; the 2.1 emitters read examples, constraints, tags and deprecation from it. `meta` is a MetaMap newtype rather than a bare BTreeMap because Field derives Hash/Ord/Eq and serde_json::Value implements none of them. The manual impls compare and hash by canonical JSON text, which is deterministic given BTreeMap's key ordering. Five tests pin the forward-compatibility contract 2.1 depends on: unknown future fields deserialize, absent meta defaults empty, meta round-trips including keys this version assigns no meaning to, empty meta never serializes (so 2.0 artifacts stay byte-identical and endpointgen --check sees no spurious drift), and Field's Hash/Ord derives still work with meta populated. Note for the lockstep release: #[non_exhaustive] breaks endpointgen's struct literal at definitions.rs:430 — fixed in that repo in this same session. Verified: types / ws-core / full build, clippy clean, 77 tests green. Co-Authored-By: Claude Fable 5 --- src/model/endpoint.rs | 103 +++++++++++++++++++++++++++++++++++++++++- src/model/types.rs | 73 ++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 1 deletion(-) diff --git a/src/model/endpoint.rs b/src/model/endpoint.rs index e7f1554..1889171 100644 --- a/src/model/endpoint.rs +++ b/src/model/endpoint.rs @@ -1,4 +1,4 @@ -use crate::model::{Field, Type}; +use crate::model::{Field, MetaMap, Type}; use convert_case::{Case, Casing}; use eyre::{ContextCompat, Result}; use serde::de::{Error, Unexpected}; @@ -8,6 +8,7 @@ use std::fmt::Write; /// `EndpointSchema` is a struct that represents a single endpoint in the API. #[derive(Debug, Serialize, Deserialize, Default, Clone)] +#[non_exhaustive] pub struct EndpointSchema { /// The name of the endpoint (e.g. `UserListSymbols`) pub name: String, @@ -39,6 +40,11 @@ pub struct EndpointSchema { /// Public error variants that handlers may return for this endpoint. #[serde(default)] pub errors: Vec, + + /// Emitter annotations — see [`MetaMap`](crate::model::MetaMap). Empty in 2.0; + /// consumed by the OpenAPI/AsyncAPI emitters in 2.1. + #[serde(default, skip_serializing_if = "MetaMap::is_empty")] + pub meta: MetaMap, } impl EndpointSchema { @@ -59,9 +65,17 @@ impl EndpointSchema { json_schema: Default::default(), roles: Vec::new(), errors: Vec::new(), + meta: MetaMap::default(), } } + /// Attach emitter annotations. See [`MetaMap`](crate::model::MetaMap). + #[must_use] + pub fn with_meta(mut self, meta: MetaMap) -> Self { + self.meta = meta; + self + } + /// Adds a stream response type field to the endpoint. pub fn with_stream_response_type(mut self, stream_response: Type) -> Self { self.stream_response = Some(stream_response); @@ -88,6 +102,7 @@ impl EndpointSchema { } #[derive(Clone, Debug, Serialize, Deserialize, Hash, PartialEq, PartialOrd, Eq, Ord)] +#[non_exhaustive] pub struct EndpointErrorSchema { pub name: String, pub code: EndpointErrorCodeRef, @@ -212,3 +227,89 @@ pub fn encode_header(v: T, schema: EndpointSchema) -> Result); + +impl MetaMap { + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + pub fn get(&self, key: &str) -> Option<&serde_json::Value> { + self.0.get(key) + } + + pub fn insert(&mut self, key: impl Into, value: serde_json::Value) { + self.0.insert(key.into(), value); + } +} + +/// Deterministic because `BTreeMap` iterates in key order and `Value`'s `Display` +/// is a canonical JSON rendering. +impl Hash for MetaMap { + fn hash(&self, state: &mut H) { + for (key, value) in &self.0 { + key.hash(state); + value.to_string().hash(state); + } + } +} + +impl Ord for MetaMap { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.0 + .iter() + .map(|(k, v)| (k, v.to_string())) + .cmp(other.0.iter().map(|(k, v)| (k, v.to_string()))) + } +} + +impl PartialOrd for MetaMap { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} /// `Field` is a struct that represents the parameters and returns in an endpoint schema. #[derive(Clone, Debug, Hash, Serialize, Deserialize, Ord, PartialOrd, Eq, PartialEq)] +#[non_exhaustive] pub struct Field { /// The name of the field (e.g. `user_id`) pub name: String, @@ -12,6 +70,10 @@ pub struct Field { /// The type of the field (e.g. `Type::BigInt`) pub ty: Type, + + /// Emitter annotations — see [`MetaMap`]. Empty in 2.0. + #[serde(default, skip_serializing_if = "MetaMap::is_empty")] + pub meta: MetaMap, } impl Field { @@ -22,6 +84,7 @@ impl Field { name: name.into(), description: "".into(), ty, + meta: MetaMap::default(), } } @@ -35,12 +98,21 @@ impl Field { name: name.into(), description: description.into(), ty, + meta: MetaMap::default(), } } + + /// Attach emitter annotations. See [`MetaMap`]. + #[must_use] + pub fn with_meta(mut self, meta: MetaMap) -> Self { + self.meta = meta; + self + } } /// `EnumVariant` is a struct that represents the variants of an enum. #[derive(Clone, Debug, Hash, Serialize, Deserialize, Ord, PartialOrd, Eq, PartialEq)] +#[non_exhaustive] pub struct EnumVariant { /// The name of the variant (e.g. `UniSwap`) pub name: String, @@ -79,6 +151,7 @@ impl EnumVariant { /// `Type` is an enum that represents the types of the fields in an endpoint schema. #[derive(Clone, Debug, Serialize, Deserialize, Hash, PartialEq, PartialOrd, Eq, Ord)] +#[non_exhaustive] pub enum Type { UInt32, Int32, From 6ef9df3f6f9cee69bc6f7ba4b5f630a3328594bd Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:42:45 +0700 Subject: [PATCH 4/6] =?UTF-8?q?feat(2.0):=20add=20the=20transport=20seam?= =?UTF-8?q?=20=E2=80=94=20serve=20over=20anything,=20not=20just=20WebSocke?= =?UTF-8?q?ts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The point of 2.0: the session/dispatch/MCP machinery no longer assumes a WebSocket underneath it. - `libs::ws::transport`: `Transport` (STOLEN SHAPE from tarpc — a blanket alias over Sink + Stream, so implementors never name it) and `TransportStream`, which adapts any such transport into the object-safe `MessageStream`. - `WsStream` trait renamed to `MessageStream` (alias kept) — it was never WebSocket-specific, only named that way. - `transport::framed` behind the new `framed-transport` feature: length- delimited WireMessage framing over any byte stream. Format is documented in the module and pinned by a test, because non-Rust peers must implement it. Uses tokio-util's LengthDelimitedCodec but NOT tokio-serde: the kind byte means the payload is not a bare serde value, so that layer buys nothing. - Server: `serve_connection` (transport-agnostic entry) + `serve_with` + `SessionListener`. `post_upgrade_connection` is now a thin WS-specific wrapper over `serve_connection`; the upgrader/TLS/shard machinery is untouched and feeds the same entry. - Client: `WsClient::from_stream`, the mirror of `serve_connection`. The private WsStream enum gains a Message variant — both additive. The acceptance test (tests/transport_seam.rs) is the definition of done: over an in-memory duplex pipe with no TCP socket, it round-trips a legacy {method,seq,params} request through a real registered handler AND completes an MCP initialize -> tools/list -> tools/call, driving the server through serve_connection and the client through from_stream. examples/uds_echo.rs proves the same on a real OS transport. Two things the feature matrix caught that a `full`-only build would not: ErrorCode/get_conn_id/get_log_id were imported under #[cfg(feature = "ws")], so serve_connection did not compile for ws-core — exactly the configuration a local transport uses. And the acceptance test silently ran zero tests until ws-client was added to the feature set. Known follow-up (non-breaking, deliberately not in scope): `from_stream` lives behind `ws-client`, which pulls tungstenite. Splitting the client so it is available from ws-core alone is additive and can come with the sibling crate. Verified: types / ws-core / framed-transport / full all build, ws-core pulls no tungstenite, clippy clean, 83 lib + 2 acceptance tests green, uds_echo runs. Co-Authored-By: Claude Fable 5 --- Cargo.toml | 11 + examples/uds_echo.rs | 175 +++++++++++++++ src/libs/ws.rs | 2 + src/libs/ws/client.rs | 62 ++++-- src/libs/ws/listener.rs | 20 ++ src/libs/ws/server.rs | 106 +++++++-- src/libs/ws/traits.rs | 14 +- src/libs/ws/transport.rs | 93 ++++++++ src/libs/ws/transport/framed.rs | 371 ++++++++++++++++++++++++++++++++ tests/transport_seam.rs | 258 ++++++++++++++++++++++ 10 files changed, 1084 insertions(+), 28 deletions(-) create mode 100644 examples/uds_echo.rs create mode 100644 src/libs/ws/transport.rs create mode 100644 src/libs/ws/transport/framed.rs create mode 100644 tests/transport_seam.rs diff --git a/Cargo.toml b/Cargo.toml index bcaa388..5f614ad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,12 @@ ws-core = [ "dep:httpdate", "dep:crossfire", ] +framed-transport = [ + # Length-delimited WireMessage framing over any byte stream (Unix sockets, + # named pipes, inherited socketpairs). No WebSocket, no TLS, no HTTP. + "ws-core", + "dep:tokio-util", +] ws-client = [ # WS client (WsClient, WsClientBuilder) - standalone "ws-core", @@ -199,6 +205,11 @@ uuid = { version = "1", features = ["v4", "serde"] } rcgen = "0.14" cert-provider = {git = "https://github.com/dVeon-loch/cert-provider.git", features = ["dns01"]} +[[example]] +name = "uds_echo" +path = "examples/uds_echo.rs" +required-features = ["full", "framed-transport", "ws-client"] + [[example]] name = "mcp_echo" required-features = ["ws-http1"] diff --git a/examples/uds_echo.rs b/examples/uds_echo.rs new file mode 100644 index 0000000..bd9b83d --- /dev/null +++ b/examples/uds_echo.rs @@ -0,0 +1,175 @@ +//! The transport seam on a real OS transport: the ordinary endpoint machinery served +//! over a Unix domain socket, with no TCP, no TLS and no HTTP upgrade. +//! +//! ```bash +//! cargo run --example uds_echo --features full,framed-transport,ws-client +//! ``` +//! +//! This is the shape a platform-transport crate implements for real. Note what is +//! *not* here: no attestation. A plain `UnixListener` cannot tell you what code is on +//! the other end — that is what `SO_PEERPIDFD` + an executable digest (Linux), a SID +//! DACL (Windows), or an XPC code-signing requirement (macOS) are for, and why they +//! belong in `endpoint-transport-local` rather than in this crate. + +use std::sync::Arc; + +use async_trait::async_trait; +use endpoint_libs::libs::handler::{RequestHandler, Response}; +use endpoint_libs::libs::peer::{Attestation, LocalPeer, PeerIdentity}; +use endpoint_libs::libs::toolbox::{ArcToolbox, CustomError, RequestContext}; +use endpoint_libs::libs::ws::transport::{TransportStream, framed_json}; +use endpoint_libs::libs::ws::{ + AuthController, MessageStream, SessionListener, WebsocketServer, WsClient, WsConnection, + WsRequest, WsResponse, WsServerConfig, +}; +use eyre::Result; +use futures::FutureExt; +use futures::future::LocalBoxFuture; +use serde::{Deserialize, Serialize}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::task::LocalSet; + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct EchoRequest { + pub message: String, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct EchoResponse { + pub message: String, +} + +impl WsRequest for EchoRequest { + type Response = EchoResponse; + const METHOD_ID: u32 = 1; + const ROLES: &'static [u32] = &[1]; + const SCHEMA: &'static str = r#"{ + "name": "Echo", + "code": 1, + "parameters": [{"name": "message", "ty": "String"}], + "returns": [{"name": "message", "ty": "String"}], + "description": "Echoes the message back.", + "roles": [] + }"#; +} + +impl WsResponse for EchoResponse { + type Request = EchoRequest; +} + +struct MethodEcho; + +#[async_trait(?Send)] +impl RequestHandler for MethodEcho { + type Request = EchoRequest; + type Error = CustomError; + + async fn handle(&self, ctx: RequestContext, req: EchoRequest) -> Response { + println!("[server] handling request from {}", ctx.peer); + Ok(EchoResponse { + message: format!("echo: {}", req.message), + }) + } +} + +struct AllowAllAuthController; + +impl AuthController for AllowAllAuthController { + fn auth( + self: Arc, + _toolbox: &ArcToolbox, + _header: String, + conn: Arc, + ) -> LocalBoxFuture<'static, Result<()>> { + async move { + conn.set_roles(Arc::new(vec![1])); + Ok(()) + } + .boxed_local() + } +} + +/// A `SessionListener` over a Unix socket — the seam a platform crate implements. +struct UdsListener { + inner: UnixListener, +} + +#[async_trait] +impl SessionListener for UdsListener { + async fn accept(&self) -> Result<(Box, PeerIdentity)> { + let (stream, _addr) = self.inner.accept().await?; + + // SO_PEERCRED gives pid/uid for free on Unix. It identifies the *process*, + // not the *code* — hence Attestation::None. Upgrading this to + // Attestation::Verified is exactly what the sibling crate adds. + let peer = PeerIdentity::Local(LocalPeer { + pid: stream.peer_cred().ok().and_then(|c| c.pid()).map(|p| p as u32), + uid: stream.peer_cred().ok().map(|c| c.uid()), + attestation: Attestation::None, + }); + + let framed: Box = Box::new(TransportStream::new(framed_json(stream))); + Ok((framed, peer)) + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let path = std::env::temp_dir().join(format!("endpoint-libs-uds-{}.sock", std::process::id())); + let _ = std::fs::remove_file(&path); + + let listener = UdsListener { + inner: UnixListener::bind(&path)?, + }; + println!("[server] listening on {}", path.display()); + + let config = WsServerConfig { + insecure: true, + ..Default::default() + }; + let mut server = WebsocketServer::new(config); + server.set_auth_controller(AllowAllAuthController); + server.add_handler(MethodEcho); + + // MessageStream's futures are not Send, so everything runs on a LocalSet. + let local = LocalSet::new(); + let client_path = path.clone(); + local + .run_until(async move { + tokio::task::spawn_local(async move { + if let Err(err) = server.serve_with(listener).await { + eprintln!("[server] stopped: {err}"); + } + }); + + // Give the listener a moment, then connect as a client over the same + // socket using the transport-agnostic constructor. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let stream = UnixStream::connect(&client_path).await?; + let client_stream: Box = + Box::new(TransportStream::new(framed_json(stream))); + let mut client = WsClient::from_stream(client_stream); + + let resp: EchoResponse = client + .request(EchoRequest { + message: "over a unix socket".into(), + }) + .await?; + println!("[client] got: {}", resp.message); + assert_eq!(resp.message, "echo: over a unix socket"); + println!("[client] OK — endpoint machinery ran with no TCP, TLS or HTTP"); + Ok::<_, eyre::Error>(()) + }) + .await?; + + let _ = std::fs::remove_file(&path); + Ok(()) +} diff --git a/src/libs/ws.rs b/src/libs/ws.rs index 56c3e64..bae9408 100644 --- a/src/libs/ws.rs +++ b/src/libs/ws.rs @@ -13,6 +13,7 @@ mod subs; mod tls; pub mod toolbox; mod traits; +pub mod transport; #[cfg(feature = "ws-client")] mod client; @@ -32,6 +33,7 @@ pub use subs::*; #[cfg(any(feature = "ws", feature = "ws-wtx"))] pub use tls::*; pub use traits::*; +pub use transport::*; #[cfg(feature = "ws-client")] pub use client::*; diff --git a/src/libs/ws/client.rs b/src/libs/ws/client.rs index 3a2ad72..6095b13 100644 --- a/src/libs/ws/client.rs +++ b/src/libs/ws/client.rs @@ -56,6 +56,9 @@ pub struct WsConnectResponse { enum WsStream { H1(Box>>), H2(Box>>), + /// Any transport-agnostic message channel — a framed Unix socket, a named pipe, + /// an XPC connection. Added in 2.0 alongside [`WsClient::from_stream`]. + Message(Box), } // --------------------------------------------------------------------------- @@ -104,33 +107,68 @@ impl WsClient { )) } + /// Build a client over any [`MessageStream`], bypassing TCP/TLS entirely. + /// + /// The transport-agnostic counterpart to [`Self::new`], and the client-side mirror + /// of [`WebsocketServer::serve_connection`](crate::libs::ws::WebsocketServer::serve_connection). + /// All the request/reply machinery — sequence correlation, response routing, MCP + /// framing — is shared with the WebSocket path; only the byte plumbing differs. + /// + /// Use with [`framed_json`](crate::libs::ws::transport::framed_json) over a Unix + /// socket or inherited socketpair, or with a platform transport's own + /// `MessageStream` implementation. + /// + /// Must be driven inside a `tokio::task::LocalSet` — `MessageStream`'s futures are + /// not `Send`. + pub fn from_stream(stream: Box) -> Self { + Self { + stream: WsStream::Message(stream), + seq: 0, + } + } + // --- Private stream helpers ------------------------------------------- async fn stream_send(&mut self, msg: Message) -> Result<()> { // Backend edge: the client speaks WireMessage; tungstenite's type exists // only inside these helpers. - let msg: TMessage = msg.into(); match &mut self.stream { - WsStream::H1(s) => s.send(msg).await?, - WsStream::H2(s) => s.send(msg).await?, + WsStream::H1(s) => s.send(TMessage::from(msg)).await?, + WsStream::H2(s) => s.send(TMessage::from(msg)).await?, + WsStream::Message(s) => s + .send(msg) + .await + .map_err(|err| eyre!("message stream send failed: {err}"))?, } Ok(()) } - async fn stream_next( - &mut self, - ) -> Option> { - let next = match &mut self.stream { - WsStream::H1(s) => s.next().await, - WsStream::H2(s) => s.next().await, - }; - next.map(|res| res.map(Into::into)) + async fn stream_next(&mut self) -> Option> { + match &mut self.stream { + WsStream::H1(s) => s + .next() + .await + .map(|res| res.map(Into::into).map_err(Into::into)), + WsStream::H2(s) => s + .next() + .await + .map(|res| res.map(Into::into).map_err(Into::into)), + WsStream::Message(s) => s + .recv() + .await + .map(|res| res.map_err(|err| eyre!("message stream recv failed: {err}"))), + } } async fn stream_close(&mut self) -> Result<()> { match &mut self.stream { WsStream::H1(s) => s.as_mut().close(None).await?, WsStream::H2(s) => s.as_mut().close(None).await?, + WsStream::Message(s) => { + // No protocol-level close handshake on a plain message channel: + // send the Close frame and let the transport tear down. + let _ = s.send(Message::Close(None)).await; + } } Ok(()) } @@ -145,7 +183,7 @@ impl WsClient { params, })?; debug!("send req: {}", req); - self.stream_send(Message::Text(req.into())).await + self.stream_send(Message::Text(req)).await } /// Send a fully pre-serialized request message. diff --git a/src/libs/ws/listener.rs b/src/libs/ws/listener.rs index 5d93103..5ea6854 100644 --- a/src/libs/ws/listener.rs +++ b/src/libs/ws/listener.rs @@ -1,3 +1,4 @@ +use crate::libs::peer::PeerIdentity; use std::net::SocketAddr; use eyre::Result; @@ -39,3 +40,22 @@ impl ConnectionListener for TcpListener { async move { Ok(channel) }.boxed() } } + +/// Accepts already-framed connections for [`WebsocketServer::serve_with`]. +/// +/// This is the seam a platform-transport crate implements: a Unix socket listener, a +/// Windows named-pipe server, or an XPC mach-service listener each yield a +/// [`MessageStream`] plus the [`PeerIdentity`] they were able to establish — including +/// any code-signature attestation, which is the whole point of the local transports. +/// +/// Distinct from [`ConnectionListener`], which yields *raw byte streams* for the +/// TCP/TLS path and knows nothing about messages or peers. +#[async_trait::async_trait] +pub trait SessionListener: Send + Sync + 'static { + /// Wait for the next peer. + /// + /// Returning `Err` stops `serve_with`, so implementations should handle + /// per-connection failures internally and only surface errors that make the + /// listener itself unusable. + async fn accept(&self) -> eyre::Result<(Box, PeerIdentity)>; +} diff --git a/src/libs/ws/server.rs b/src/libs/ws/server.rs index 035f4d1..8dee95c 100644 --- a/src/libs/ws/server.rs +++ b/src/libs/ws/server.rs @@ -14,12 +14,12 @@ use tokio::sync::mpsc; use tokio::task::LocalSet; use tracing::*; -#[cfg(feature = "ws")] +// Used by serve_connection, which is transport-agnostic (ws-core), so these must +// not be gated on the tungstenite backend. use crate::libs::error_code::ErrorCode; use crate::libs::handler::{RequestHandler, RequestHandlerErased}; use crate::libs::peer::{Extensions, PeerIdentity}; use crate::libs::toolbox::{ArcToolbox, RequestContext, TOOLBOX, Toolbox}; -#[cfg(feature = "ws")] use crate::libs::utils::{get_conn_id, get_log_id}; #[cfg(feature = "ws")] use crate::libs::ws::HyperTungsteniteUpgrader; @@ -29,8 +29,8 @@ use crate::libs::ws::mcp::{McpServerInfo, McpState}; #[cfg(feature = "ws")] use crate::libs::ws::tungstenite::upgrader::create_ws_stream; use crate::libs::ws::{ - BoxedStream, ConnectionListener, TcpListener, WsClientSession, WsConnection, WsRequest, - WsStream, WsUpgrader, + BoxedStream, ConnectionListener, MessageStream, SessionListener, TcpListener, + WsClientSession, WsConnection, WsRequest, WsUpgrader, }; use crate::model::{EndpointSchema, TypeRegistry}; @@ -184,26 +184,39 @@ impl WebsocketServer { ) } - #[cfg(feature = "ws")] - async fn post_upgrade_connection( + /// Run auth and the session loop for one already-established connection. + /// + /// This is the transport-agnostic server entry point. It knows nothing about TCP, + /// TLS or HTTP upgrades: give it a [`MessageStream`] and a [`PeerIdentity`] and it + /// does the rest. The WebSocket path reaches it through + /// `post_upgrade_connection`; local transports (Unix socket, named pipe, XPC) call + /// it directly. + /// + /// `auth_protocol` is whatever the transport uses to carry credentials at connect + /// time — the WebSocket subprotocol string today, a handed-over token for local + /// transports. It is passed to [`AuthController::auth`] unchanged. + /// + /// Must be called inside a `tokio::task::LocalSet`: [`MessageStream`]'s futures + /// are not `Send`. + pub async fn serve_connection( self: Arc, - addr: SocketAddr, + peer: PeerIdentity, states: Arc, - stream: Box, - protocol: String, + stream: Box, + auth_protocol: Option, ) { let conn = Arc::new(WsConnection { connection_id: get_conn_id(), user_id: Default::default(), roles: Arc::new(RwLock::new(Arc::new(Vec::new()))), - peer: PeerIdentity::Network(addr), + peer, extensions: Extensions::new(), log_id: get_log_id(), }); debug!( ws_server = true, - ?addr, - "New connection handshaken {:?}", + peer = %conn.peer, + "New connection established {:?}", conn ); @@ -211,7 +224,11 @@ impl WebsocketServer { states.insert(conn.connection_id, tx, conn.clone()); let auth_result = Arc::clone(&self.auth_controller) - .auth(&self.toolbox, protocol, Arc::clone(&conn)) + .auth( + &self.toolbox, + auth_protocol.unwrap_or_default(), + Arc::clone(&conn), + ) .await; let raw_ctx = RequestContext::from_conn(&conn); if let Err(err) = auth_result { @@ -220,7 +237,7 @@ impl WebsocketServer { error!( ws_server=true, error_code=?ErrorCode::BAD_REQUEST, - ip_addr=%raw_ctx.ip_addr, + peer=%conn.peer, user_id=raw_ctx.user_id, conn_id=raw_ctx.connection_id, roles=?raw_ctx.roles, @@ -234,11 +251,30 @@ impl WebsocketServer { .await; } + /// The WebSocket-specific wrapper: everything TCP/TLS/upgrade-shaped stops here, + /// and the generic path continues in [`Self::serve_connection`]. + #[cfg(feature = "ws")] + async fn post_upgrade_connection( + self: Arc, + addr: SocketAddr, + states: Arc, + stream: Box, + protocol: String, + ) { + self.serve_connection( + PeerIdentity::Network(addr), + states, + stream, + Some(protocol), + ) + .await; + } + pub async fn handle_session_connection( self: Arc, conn: Arc, states: Arc, - stream: Box, + stream: Box, rx: mpsc::Receiver, ) { let addr = conn.peer.display(); @@ -263,6 +299,46 @@ impl WebsocketServer { ); } + /// Accept connections from any [`SessionListener`] and serve each one. + /// + /// The transport-agnostic counterpart to [`Self::listen`]. Unlike `listen`, this + /// runs on a single runtime — the shard-per-core model is a property of the TCP + /// path and buys nothing for a 1:1 sidecar channel. + /// + /// Must be called inside a `tokio::task::LocalSet` (see + /// [`Self::serve_connection`]). + pub async fn serve_with(self, listener: L) -> Result<()> + where + L: SessionListener + 'static, + { + let this = Arc::new(self); + let states = Arc::new(WebsocketStates::new()); + this.toolbox.set_ws_states( + states.clone_states(), + this.config.header_only, + this.config.drop_conn_on_buffer_full, + ); + + loop { + let (stream, peer) = match listener.accept().await { + Ok(accepted) => accepted, + Err(err) => { + error!(ws_server = true, error = %err, "listener accept failed; stopping"); + return Err(err); + } + }; + debug!(ws_server = true, peer = %peer, "accepted connection"); + + let this = Arc::clone(&this); + let states = Arc::clone(&states); + tokio::task::spawn_local(async move { + // Local transports carry credentials out of band (an inherited fd is + // already a capability), so there is no subprotocol string to pass. + this.serve_connection(peer, states, stream, None).await; + }); + } + } + pub async fn listen(self) -> Result<()> { debug!(ws_server = true, "Listening on {}", self.config.address); diff --git a/src/libs/ws/traits.rs b/src/libs/ws/traits.rs index f68e0ea..acb7609 100644 --- a/src/libs/ws/traits.rs +++ b/src/libs/ws/traits.rs @@ -43,12 +43,24 @@ impl std::error::Error for StreamError { } } +/// An object-safe, bidirectional message channel — what the session loop consumes. +/// +/// Renamed from `WsStream` in 2.0 (the alias below keeps old code compiling): it is +/// no longer WebSocket-specific. A Unix socket, a named pipe, or an XPC connection +/// implements this just as well, either directly or via +/// [`TransportStream`](super::TransportStream). +/// +/// Note `?Send`: implementations' futures need not be `Send`, which matches the +/// `spawn_local` dispatch model. Drivers must run inside a `LocalSet`. #[async_trait(?Send)] -pub trait WsStream: Unpin + Send { +pub trait MessageStream: Unpin + Send { async fn send(&mut self, msg: Message) -> Result<(), StreamError>; async fn recv(&mut self) -> Option>; } +/// Compatibility alias for the pre-2.0 name. +pub use MessageStream as WsStream; + /// An upgrade event yielded by the upgrader. /// Contains the on_upgrade future and the negotiated protocol. pub struct UpgradeEvent { diff --git a/src/libs/ws/transport.rs b/src/libs/ws/transport.rs new file mode 100644 index 0000000..7ff59e8 --- /dev/null +++ b/src/libs/ws/transport.rs @@ -0,0 +1,93 @@ +//! The transport seam: what it takes to run the session/dispatch machinery over +//! something that is not a WebSocket. +//! +//! # Design +//! +//! Two layers, deliberately: +//! +//! * [`Transport`] — a *blanket alias* over `Sink + Stream` of [`WireMessage`], +//! modelled on tarpc's `Transport`. Implementors never name it: anything that is a +//! `Sink` and a `Stream` of the right item types already is one. This composes for +//! free with `tokio_util::codec::Framed` and with hand-rolled adapters (XPC +//! dictionaries, for instance). +//! * [`MessageStream`] — the object-safe, `async fn`-based trait the session loop +//! actually consumes. It is the pre-2.0 `WsStream` under a transport-neutral name; +//! `WsStream` remains as an alias. +//! +//! [`TransportStream`] bridges the two, so implementing either one is enough. +//! +//! # Threading +//! +//! [`MessageStream`] is `#[async_trait(?Send)]` — its futures are **not** `Send`, +//! matching the existing `spawn_local` dispatch model. Anything driving a session +//! (`serve_connection`, `serve_with`) must therefore run inside a +//! `tokio::task::LocalSet`. This is not an oversight; it is what lets handlers hold +//! non-`Send` state across await points. + +use eyre::eyre; +use futures::{Sink, SinkExt, Stream, StreamExt}; + +use super::message::WireMessage; +use super::traits::{MessageStream, StreamError}; + +/// A bidirectional, typed message channel. +/// +/// STOLEN SHAPE (tarpc `Transport`): a blanket alias, not a trait to implement. Any +/// `Sink + Stream>` satisfies it. +pub trait Transport +where + Self: Stream>::Error>>, + Self: Sink>::TransportError>, +{ + /// The error type shared by both directions. + type TransportError: std::error::Error + Send + Sync + 'static; +} + +impl Transport for T +where + T: ?Sized + Stream> + Sink, + E: std::error::Error + Send + Sync + 'static, +{ + type TransportError = E; +} + +/// Adapts any [`Transport`] of [`WireMessage`] into a [`MessageStream`]. +/// +/// This is the bridge that lets a `Framed`, an in-memory duplex pipe, +/// or an XPC connection drive the ordinary session loop. +pub struct TransportStream(pub T); + +impl TransportStream { + pub fn new(transport: T) -> Self { + Self(transport) + } + + /// Recover the wrapped transport. + pub fn into_inner(self) -> T { + self.0 + } +} + +#[async_trait::async_trait(?Send)] +impl MessageStream for TransportStream +where + T: Transport + Unpin + Send, +{ + async fn send(&mut self, msg: WireMessage) -> Result<(), StreamError> { + SinkExt::send(&mut self.0, msg) + .await + .map_err(|err| StreamError::Other(eyre!(err))) + } + + async fn recv(&mut self) -> Option> { + StreamExt::next(&mut self.0) + .await + .map(|res| res.map_err(|err| StreamError::Other(eyre!(err)))) + } +} + +#[cfg(feature = "framed-transport")] +pub mod framed; + +#[cfg(feature = "framed-transport")] +pub use framed::{FramedError, framed_json}; diff --git a/src/libs/ws/transport/framed.rs b/src/libs/ws/transport/framed.rs new file mode 100644 index 0000000..5b25e47 --- /dev/null +++ b/src/libs/ws/transport/framed.rs @@ -0,0 +1,371 @@ +//! Length-delimited framing of [`WireMessage`] over any byte stream. +//! +//! This is what carries the ordinary JSON protocol over a Unix socket, a Windows +//! named pipe, or an inherited socketpair — anywhere there is no WebSocket to provide +//! message boundaries. +//! +//! # Wire format +//! +//! Each message is one length-delimited frame: +//! +//! ```text +//! +------------------+--------+------------------------+ +//! | u32 BE length | u8 kind| payload (length-1 bytes)| +//! +------------------+--------+------------------------+ +//! ``` +//! +//! * `length` counts the kind byte plus the payload — i.e. the whole rest of the frame. +//! * `kind` is `0 = Text`, `1 = Binary`, `2 = Ping`, `3 = Pong`, `4 = Close`. +//! * `Text` payloads are UTF-8. `Close` payloads are either empty (no close frame) or +//! `u16 BE code` followed by a UTF-8 reason. +//! +//! The default maximum frame length is 16 MiB; see [`framed_json_with_max_frame`]. +//! +//! **This format is normative for any non-Rust peer.** It is deliberately trivial to +//! implement: a 4-byte length prefix, one tag byte, and a payload. It is also recorded +//! in the AsyncAPI document emitted in 2.1, which is the machine-readable copy. +//! +//! Note the implementation uses `tokio_util`'s `LengthDelimitedCodec` for the length +//! prefix but *not* `tokio_serde`: the kind byte means the payload is not a bare serde +//! value, so the serde codec layer would buy nothing. + +use std::io; + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use futures::{Sink, Stream}; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio_util::codec::{Framed, LengthDelimitedCodec}; + +use super::super::message::{CloseFrame, WireMessage}; +use super::Transport; + +/// Default maximum frame length: 16 MiB. +pub const DEFAULT_MAX_FRAME_BYTES: usize = 16 * 1024 * 1024; + +const KIND_TEXT: u8 = 0; +const KIND_BINARY: u8 = 1; +const KIND_PING: u8 = 2; +const KIND_PONG: u8 = 3; +const KIND_CLOSE: u8 = 4; + +/// Errors from the framed transport. +#[derive(Debug)] +pub enum FramedError { + Io(io::Error), + /// A frame arrived with a `kind` byte this version does not know. + UnknownKind(u8), + /// A frame was empty (not even a kind byte). + EmptyFrame, + /// A `Text` frame's payload was not valid UTF-8. + InvalidUtf8, + /// A `Close` frame's payload was malformed (1 byte, or a bad reason). + MalformedClose, +} + +impl std::fmt::Display for FramedError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(err) => write!(f, "io error: {err}"), + Self::UnknownKind(kind) => write!(f, "unknown frame kind: {kind}"), + Self::EmptyFrame => f.write_str("empty frame"), + Self::InvalidUtf8 => f.write_str("text frame was not valid UTF-8"), + Self::MalformedClose => f.write_str("malformed close frame"), + } + } +} + +impl std::error::Error for FramedError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(err) => Some(err), + _ => None, + } + } +} + +impl From for FramedError { + fn from(err: io::Error) -> Self { + Self::Io(err) + } +} + +fn encode(msg: WireMessage) -> Bytes { + let mut buf = BytesMut::new(); + match msg { + WireMessage::Text(text) => { + buf.put_u8(KIND_TEXT); + buf.put_slice(text.as_bytes()); + } + WireMessage::Binary(data) => { + buf.put_u8(KIND_BINARY); + buf.put_slice(&data); + } + WireMessage::Ping(data) => { + buf.put_u8(KIND_PING); + buf.put_slice(&data); + } + WireMessage::Pong(data) => { + buf.put_u8(KIND_PONG); + buf.put_slice(&data); + } + WireMessage::Close(frame) => { + buf.put_u8(KIND_CLOSE); + if let Some(frame) = frame { + buf.put_u16(frame.code); + buf.put_slice(frame.reason.as_bytes()); + } + } + } + buf.freeze() +} + +fn decode(mut frame: BytesMut) -> Result { + if frame.is_empty() { + return Err(FramedError::EmptyFrame); + } + let kind = frame.get_u8(); + let payload = frame; + Ok(match kind { + KIND_TEXT => WireMessage::Text( + String::from_utf8(payload.to_vec()).map_err(|_| FramedError::InvalidUtf8)?, + ), + KIND_BINARY => WireMessage::Binary(payload.to_vec()), + KIND_PING => WireMessage::Ping(payload.to_vec()), + KIND_PONG => WireMessage::Pong(payload.to_vec()), + KIND_CLOSE => { + if payload.is_empty() { + WireMessage::Close(None) + } else { + if payload.len() < 2 { + return Err(FramedError::MalformedClose); + } + let mut payload = payload; + let code = payload.get_u16(); + let reason = + String::from_utf8(payload.to_vec()).map_err(|_| FramedError::MalformedClose)?; + WireMessage::Close(Some(CloseFrame { code, reason })) + } + } + other => return Err(FramedError::UnknownKind(other)), + }) +} + +/// Wrap a byte stream in the framing described in this module's docs. +/// +/// The result is a [`Transport`] of [`WireMessage`], which +/// [`TransportStream`](super::TransportStream) turns into a +/// [`MessageStream`](super::super::traits::MessageStream) for the session loop. +pub fn framed_json( + io: S, +) -> impl Transport + Unpin + Send +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + framed_json_with_max_frame(io, DEFAULT_MAX_FRAME_BYTES) +} + +/// [`framed_json`] with an explicit maximum frame length. +/// +/// Frames longer than `max_frame_bytes` are rejected rather than buffered, which is +/// what keeps a hostile or broken peer from exhausting memory. +pub fn framed_json_with_max_frame( + io: S, + max_frame_bytes: usize, +) -> impl Transport + Unpin + Send +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + let codec = LengthDelimitedCodec::builder() + .big_endian() + .length_field_length(4) + .max_frame_length(max_frame_bytes) + .new_codec(); + + WireFramed { + inner: Framed::new(io, codec), + } +} + +/// Adapts `Framed` (bytes) to `WireMessage` in both +/// directions. +struct WireFramed { + inner: Framed, +} + +impl Stream for WireFramed +where + S: AsyncRead + AsyncWrite + Unpin, +{ + type Item = Result; + + fn poll_next( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + match std::pin::Pin::new(&mut self.inner).poll_next(cx) { + std::task::Poll::Ready(Some(Ok(frame))) => std::task::Poll::Ready(Some(decode(frame))), + std::task::Poll::Ready(Some(Err(err))) => { + std::task::Poll::Ready(Some(Err(FramedError::Io(err)))) + } + std::task::Poll::Ready(None) => std::task::Poll::Ready(None), + std::task::Poll::Pending => std::task::Poll::Pending, + } + } +} + +impl Sink for WireFramed +where + S: AsyncRead + AsyncWrite + Unpin, +{ + type Error = FramedError; + + fn poll_ready( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner) + .poll_ready(cx) + .map_err(FramedError::Io) + } + + fn start_send(mut self: std::pin::Pin<&mut Self>, item: WireMessage) -> Result<(), Self::Error> { + std::pin::Pin::new(&mut self.inner) + .start_send(encode(item)) + .map_err(FramedError::Io) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner) + .poll_flush(cx) + .map_err(FramedError::Io) + } + + fn poll_close( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner) + .poll_close(cx) + .map_err(FramedError::Io) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::{SinkExt, StreamExt}; + + #[test] + fn every_variant_round_trips_through_the_codec() { + let cases = vec![ + WireMessage::Text("hello".into()), + WireMessage::Text(String::new()), + WireMessage::Binary(vec![0, 1, 2, 255]), + WireMessage::Binary(Vec::new()), + WireMessage::Ping(vec![9]), + WireMessage::Pong(Vec::new()), + WireMessage::Close(None), + WireMessage::Close(Some(CloseFrame { + code: 1001, + reason: "going away".into(), + })), + WireMessage::Close(Some(CloseFrame { + code: 1000, + reason: String::new(), + })), + ]; + for case in cases { + let encoded = encode(case.clone()); + let decoded = decode(BytesMut::from(&encoded[..])).expect("decode"); + assert_eq!(case, decoded, "round trip changed the message"); + } + } + + #[test] + fn frame_layout_is_what_the_docs_promise() { + // Non-Rust peers implement against this. Text "hi" => kind 0, then bytes. + let encoded = encode(WireMessage::Text("hi".into())); + assert_eq!(&encoded[..], &[KIND_TEXT, b'h', b'i']); + + // Close with code 1000 and no reason => kind 4, then u16 BE. + let encoded = encode(WireMessage::Close(Some(CloseFrame { + code: 1000, + reason: String::new(), + }))); + assert_eq!(&encoded[..], &[KIND_CLOSE, 0x03, 0xE8]); + } + + #[test] + fn malformed_frames_are_errors_not_panics() { + assert!(matches!( + decode(BytesMut::new()), + Err(FramedError::EmptyFrame) + )); + assert!(matches!( + decode(BytesMut::from(&[99u8][..])), + Err(FramedError::UnknownKind(99)) + )); + assert!(matches!( + decode(BytesMut::from(&[KIND_TEXT, 0xff, 0xfe][..])), + Err(FramedError::InvalidUtf8) + )); + // A close frame with a single byte cannot hold a u16 code. + assert!(matches!( + decode(BytesMut::from(&[KIND_CLOSE, 0x01][..])), + Err(FramedError::MalformedClose) + )); + } + + #[tokio::test] + async fn duplex_pipe_carries_messages_both_ways() { + let (a, b) = tokio::io::duplex(64 * 1024); + let mut left = framed_json(a); + let mut right = framed_json(b); + + left.send(WireMessage::Text("ping".into())).await.unwrap(); + let got = right.next().await.unwrap().unwrap(); + assert_eq!(got, WireMessage::Text("ping".into())); + + right.send(WireMessage::Binary(vec![7, 7])).await.unwrap(); + let got = left.next().await.unwrap().unwrap(); + assert_eq!(got, WireMessage::Binary(vec![7, 7])); + } + + #[tokio::test] + async fn oversized_outbound_frames_are_refused_by_the_encoder() { + let (a, _b) = tokio::io::duplex(64 * 1024); + let mut left = framed_json_with_max_frame(a, 64); + + // The limit is enforced on the way out too, so we never emit a frame a + // conforming peer would have to reject. + let result = left.send(WireMessage::Binary(vec![0u8; 4096])).await; + assert!( + matches!(result, Err(FramedError::Io(_))), + "expected the encoder to refuse an oversized frame, got {result:?}" + ); + } + + #[tokio::test] + async fn oversized_inbound_frames_are_rejected_rather_than_buffered() { + use tokio::io::AsyncWriteExt; + + let (a, mut b) = tokio::io::duplex(64 * 1024); + let mut left = framed_json_with_max_frame(a, 64); + + // Write the length prefix by hand — a hostile peer is not using our encoder, + // so this is the case that actually protects memory. + b.write_all(&5000u32.to_be_bytes()).await.unwrap(); + b.write_all(&[KIND_BINARY]).await.unwrap(); + b.write_all(&[0u8; 128]).await.unwrap(); + b.flush().await.unwrap(); + + let got = left.next().await; + assert!( + matches!(got, Some(Err(FramedError::Io(_)))), + "expected an io error for an oversized declared length, got {got:?}" + ); + } +} diff --git a/tests/transport_seam.rs b/tests/transport_seam.rs new file mode 100644 index 0000000..1538f6b --- /dev/null +++ b/tests/transport_seam.rs @@ -0,0 +1,258 @@ +//! The acceptance test for 2.0's core claim: the session/dispatch/MCP machinery runs +//! over a transport that is not a WebSocket and never touches a TCP socket. +//! +//! Both halves of the seam are exercised by the same test — the server through +//! [`WebsocketServer::serve_connection`], the client through +//! [`WsClient::from_stream`] — over an in-memory `tokio::io::duplex` pipe framed with +//! [`framed_json`]. +//! +//! If this file fails to compile, the transport seam has regressed. + +#![cfg(all(feature = "framed-transport", feature = "ws-client"))] + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use endpoint_libs::libs::handler::{RequestHandler, Response}; +use endpoint_libs::libs::peer::{Attestation, LocalPeer, PeerIdentity}; +use endpoint_libs::libs::toolbox::{ArcToolbox, CustomError, RequestContext}; +use endpoint_libs::libs::ws::transport::{TransportStream, framed_json}; +use endpoint_libs::libs::ws::{ + AuthController, MessageStream, WebsocketServer, WebsocketStates, WsClient, WsConnection, + WsRequest, WsResponse, WsServerConfig, +}; +use eyre::Result; +use futures::FutureExt; +use futures::future::LocalBoxFuture; +use serde::{Deserialize, Serialize}; +use tokio::task::LocalSet; + +// --- A real endpoint, registered the ordinary way ------------------------- + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct EchoRequest { + pub message: String, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct EchoResponse { + pub message: String, +} + +impl WsRequest for EchoRequest { + type Response = EchoResponse; + const METHOD_ID: u32 = 1; + const ROLES: &'static [u32] = &[1]; + const SCHEMA: &'static str = r#"{ + "name": "Echo", + "code": 1, + "parameters": [{"name": "message", "ty": "String"}], + "returns": [{"name": "message", "ty": "String"}], + "description": "Echoes the message back.", + "roles": [] + }"#; +} + +impl WsResponse for EchoResponse { + type Request = EchoRequest; +} + +struct MethodEcho; + +#[async_trait(?Send)] +impl RequestHandler for MethodEcho { + type Request = EchoRequest; + type Error = CustomError; + + async fn handle(&self, ctx: RequestContext, req: EchoRequest) -> Response { + // Proves the attested peer identity reaches handler code — the whole point + // of threading PeerIdentity through in Phase 2. + let peer = match &ctx.peer { + // Attestation is #[non_exhaustive] (Phase 2b), so an out-of-crate match + // needs a wildcard — future mechanisms must not break this test. + PeerIdentity::Local(local) => match &local.attestation { + Attestation::Verified { mechanism, .. } => format!("local/{mechanism}"), + Attestation::None => "local/unattested".to_owned(), + _ => "local/unknown-attestation".to_owned(), + }, + PeerIdentity::Network(_) => "network".to_owned(), + _ => "unknown".to_owned(), + }; + Ok(EchoResponse { + message: format!("echo[{peer}]: {}", req.message), + }) + } +} + +struct AllowAllAuthController; + +impl AuthController for AllowAllAuthController { + fn auth( + self: Arc, + _toolbox: &ArcToolbox, + _header: String, + conn: Arc, + ) -> LocalBoxFuture<'static, Result<()>> { + async move { + conn.set_roles(Arc::new(vec![1])); + Ok(()) + } + .boxed_local() + } +} + +fn build_server(enable_mcp: bool) -> WebsocketServer { + let config = WsServerConfig { + insecure: true, + ..Default::default() + }; + let mut server = WebsocketServer::new(config); + server.set_auth_controller(AllowAllAuthController); + server.add_handler(MethodEcho); + if enable_mcp { + let mut registry = endpoint_libs::model::TypeRegistry::new(); + let schema: endpoint_libs::model::EndpointSchema = + serde_json::from_str(EchoRequest::SCHEMA).unwrap(); + registry.add_endpoint(&schema); + server + .enable_mcp( + ®istry, + endpoint_libs::libs::ws::mcp::McpServerInfo { + name: "seam-test".into(), + version: "0.0.0".into(), + }, + ) + .expect("enable_mcp"); + } + server +} + +/// An attested local peer, as a platform transport would report one. +fn attested_peer() -> PeerIdentity { + PeerIdentity::Local(LocalPeer { + pid: Some(std::process::id()), + uid: None, + attestation: Attestation::Verified { + mechanism: "test-harness", + subject: "acceptance".to_owned(), + }, + }) +} + +fn spawn_server( + server: WebsocketServer, + server_io: tokio::io::DuplexStream, +) -> impl std::future::Future { + let server = Arc::new(server); + let states = Arc::new(WebsocketStates::new()); + server.toolbox.set_ws_states(states.clone_states(), false, false); + let stream: Box = Box::new(TransportStream::new(framed_json(server_io))); + server.serve_connection(attested_peer(), states, stream, None) +} + +/// (a) A legacy `{method, seq, params}` request reaches a real registered handler and +/// its response comes back — over a duplex pipe, with no TCP socket anywhere. +#[tokio::test(flavor = "current_thread")] +async fn legacy_request_round_trips_over_a_non_websocket_transport() { + let local = LocalSet::new(); + local + .run_until(async { + let (server_io, client_io) = tokio::io::duplex(256 * 1024); + + tokio::task::spawn_local(spawn_server(build_server(false), server_io)); + + let client_stream: Box = + Box::new(TransportStream::new(framed_json(client_io))); + let mut client = WsClient::from_stream(client_stream); + + let resp: EchoResponse = tokio::time::timeout( + Duration::from_secs(5), + client.request(EchoRequest { + message: "hello".into(), + }), + ) + .await + .expect("request timed out") + .expect("request failed"); + + // The handler saw the attestation the transport supplied. + assert_eq!(resp.message, "echo[local/test-harness]: hello"); + }) + .await; +} + +/// (b) MCP `initialize` → `tools/list` → `tools/call` completes on the *same* +/// connection type, proving the JSON-RPC surface is not tied to WebSockets either. +#[tokio::test(flavor = "current_thread")] +async fn mcp_initialize_and_tool_call_work_over_the_same_transport() { + let local = LocalSet::new(); + local + .run_until(async { + let (server_io, client_io) = tokio::io::duplex(256 * 1024); + + tokio::task::spawn_local(spawn_server(build_server(true), server_io)); + + let client_stream: Box = + Box::new(TransportStream::new(framed_json(client_io))); + let mut client = WsClient::from_stream(client_stream); + + let initialize = serde_json::json!({ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "seam-test", "version": "0.0.0"}} + }); + client + .send_raw(initialize.to_string().as_bytes()) + .await + .expect("send initialize"); + let resp = tokio::time::timeout(Duration::from_secs(5), client.recv_raw()) + .await + .expect("initialize timed out") + .expect("initialize failed"); + assert_eq!(resp["id"], 1, "initialize response: {resp}"); + assert!( + resp["result"]["serverInfo"]["name"] == "seam-test", + "unexpected initialize result: {resp}" + ); + + let list = serde_json::json!({ + "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} + }); + client + .send_raw(list.to_string().as_bytes()) + .await + .expect("send tools/list"); + let resp = tokio::time::timeout(Duration::from_secs(5), client.recv_raw()) + .await + .expect("tools/list timed out") + .expect("tools/list failed"); + let tools = resp["result"]["tools"] + .as_array() + .unwrap_or_else(|| panic!("no tools array in {resp}")); + assert_eq!(tools.len(), 1, "expected exactly the echo tool: {resp}"); + assert_eq!(tools[0]["name"], "echo"); + + let call = serde_json::json!({ + "jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "echo", "arguments": {"message": "via-mcp"}} + }); + client + .send_raw(call.to_string().as_bytes()) + .await + .expect("send tools/call"); + let resp = tokio::time::timeout(Duration::from_secs(5), client.recv_raw()) + .await + .expect("tools/call timed out") + .expect("tools/call failed"); + assert_eq!(resp["id"], 3, "tools/call response: {resp}"); + let text = resp["result"]["content"][0]["text"] + .as_str() + .unwrap_or_else(|| panic!("no text content in {resp}")); + assert!( + text.contains("echo[local/test-harness]: via-mcp"), + "tool call did not reach the handler: {text}" + ); + }) + .await; +} From e58ae1f960f4f3e5c7286f3900c0d90fe6d784de Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:47:22 +0700 Subject: [PATCH 5/6] feat(2.0): add request hooks on both dispatch paths, plus OnConnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit STOLEN SHAPE (tarpc request_hook): a Before/After pair around dispatch. This is where policy that is not endpoint-specific plugs in — mission-token verification above all — without the transport layer or the handlers knowing. - `libs::ws::hooks`: BeforeRequest (may reject, may attach claims to ctx.extensions), AfterRequest (observes RequestOutcome), OnConnect (refuses a peer once, rather than re-checking attestation on every request). - Registered via add_before_hook / add_after_hook / add_on_connect_hook; snapshotted into each spawned task so a slow hook cannot stall the session loop. Registration order, first error short-circuits. - Placement is after check_roles on both paths, so hooks only see calls already allowed to reach the endpoint. The subtle part is that the two paths need different error envelopes: the legacy path emits a WsResponseError carrying the hook's code and params, the MCP path emits jsonrpc_result(encode_tool_error(..)) so the caller sees a tool error with isError: true. Both are asserted. OnConnect runs before the connection is registered in `states`, so a refused peer never gets a slot and cannot be sent to. Five acceptance tests now cover: legacy round trip, MCP initialize/list/call, hook rejection on the legacy path (exact frame, code and params), hook rejection on the MCP path (tool error payload), claims flowing hook -> handler via extensions, AfterRequest observing both outcomes, and OnConnect admitting an attested peer. Note for anyone adding handlers: `check_handler` requires the struct to be named `Method`, so the claims test needed its own Claims endpoint rather than a second handler on Echo. Verified: ws-core builds, clippy clean, 83 lib + 5 acceptance tests green. Co-Authored-By: Claude Fable 5 --- src/libs/ws.rs | 2 + src/libs/ws/hooks.rs | 141 +++++++++++++++++++++ src/libs/ws/server.rs | 41 +++++- src/libs/ws/session.rs | 77 ++++++++++-- tests/transport_seam.rs | 267 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 517 insertions(+), 11 deletions(-) create mode 100644 src/libs/ws/hooks.rs diff --git a/src/libs/ws.rs b/src/libs/ws.rs index bae9408..cd3f113 100644 --- a/src/libs/ws.rs +++ b/src/libs/ws.rs @@ -1,6 +1,7 @@ mod basics; mod conn; pub mod handler; +pub mod hooks; mod headers; mod listener; pub mod mcp; @@ -25,6 +26,7 @@ pub(crate) mod wtx; pub use basics::*; pub use conn::*; pub use headers::*; +pub use hooks::*; pub use listener::*; pub use message::*; pub use server::*; diff --git a/src/libs/ws/hooks.rs b/src/libs/ws/hooks.rs new file mode 100644 index 0000000..6704390 --- /dev/null +++ b/src/libs/ws/hooks.rs @@ -0,0 +1,141 @@ +//! Per-request and per-connection interception. +//! +//! STOLEN SHAPE (tarpc `request_hook`): a `Before`/`After` pair around dispatch. +//! +//! This is the seam where policy that is *not* endpoint-specific plugs in — mission +//! token verification, quota enforcement, audit logging — without the transport layer +//! or the handlers knowing it exists. Hooks see every request on both the legacy +//! `{method, seq, params}` path and the MCP `tools/call` path. +//! +//! # Ordering +//! +//! ```text +//! connect ──► OnConnect ──► auth ──► [per request] roles ──► BeforeRequest ──► handler ──► AfterRequest +//! ``` +//! +//! `BeforeRequest` runs *after* the endpoint's role check, so a hook can assume the +//! caller was allowed to reach this endpoint at all and concern itself only with the +//! finer-grained question. Hooks run in registration order and the first error +//! short-circuits: the handler never runs and the error goes back in whichever +//! envelope the caller used. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::Value; + +use crate::libs::peer::{Extensions, PeerIdentity}; +use crate::libs::toolbox::{CustomError, RequestContext}; +use crate::model::EndpointSchema; + +/// How a request finished. Passed to [`AfterRequest`]. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum RequestOutcome { + Ok, + /// The handler (or a `BeforeRequest` hook) returned a public error. + PublicErr { code: u32 }, + /// The handler failed internally; the client saw a generic error. + InternalErr, +} + +/// Runs before a request is dispatched; may reject it. +#[async_trait(?Send)] +pub trait BeforeRequest: Send + Sync { + /// Returning `Err` skips the handler entirely and sends the error to the client. + /// + /// `ctx` is mutable so a hook can attach verified claims to + /// [`RequestContext::extensions`] for the handler to read. + async fn before( + &self, + ctx: &mut RequestContext, + endpoint: &EndpointSchema, + params: &Value, + ) -> Result<(), CustomError>; +} + +/// Runs after a request completes, for observation only. +#[async_trait(?Send)] +pub trait AfterRequest: Send + Sync { + async fn after(&self, ctx: &RequestContext, endpoint: &EndpointSchema, outcome: &RequestOutcome); +} + +/// Runs once per connection, after auth, before any message is processed. +/// +/// This is where a peer that failed attestation gets refused — once, rather than in +/// every [`BeforeRequest`]. Note that on macOS XPC a peer failing the code-signing +/// requirement never reaches Rust at all (libxpc drops the check-in), so this hook +/// sees only peers the transport was willing to hand over. +#[async_trait(?Send)] +pub trait OnConnect: Send + Sync { + /// Returning `Err` refuses the connection; no messages are exchanged. + /// + /// `ext` is the connection-scoped [`Extensions`], so a hook can record what it + /// verified for later requests to consult. + async fn on_connect( + &self, + peer: &PeerIdentity, + ext: &mut Extensions, + ) -> Result<(), CustomError>; +} + +/// The registered hooks, snapshotted into each spawned dispatch task. +#[derive(Clone, Default)] +pub struct Hooks { + pub(crate) before: Vec>, + pub(crate) after: Vec>, + pub(crate) on_connect: Vec>, +} + +impl Hooks { + pub fn is_empty(&self) -> bool { + self.before.is_empty() && self.after.is_empty() && self.on_connect.is_empty() + } + + /// Run every `BeforeRequest` in registration order, stopping at the first error. + pub(crate) async fn run_before( + &self, + ctx: &mut RequestContext, + endpoint: &EndpointSchema, + params: &Value, + ) -> Result<(), CustomError> { + for hook in &self.before { + hook.before(ctx, endpoint, params).await?; + } + Ok(()) + } + + /// Run every `AfterRequest`. Observers cannot fail the request, so errors are + /// not representable here. + pub(crate) async fn run_after( + &self, + ctx: &RequestContext, + endpoint: &EndpointSchema, + outcome: &RequestOutcome, + ) { + for hook in &self.after { + hook.after(ctx, endpoint, outcome).await; + } + } + + pub(crate) async fn run_on_connect( + &self, + peer: &PeerIdentity, + ext: &mut Extensions, + ) -> Result<(), CustomError> { + for hook in &self.on_connect { + hook.on_connect(peer, ext).await?; + } + Ok(()) + } +} + +impl std::fmt::Debug for Hooks { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Hooks") + .field("before", &self.before.len()) + .field("after", &self.after.len()) + .field("on_connect", &self.on_connect.len()) + .finish() + } +} diff --git a/src/libs/ws/server.rs b/src/libs/ws/server.rs index 8dee95c..8d9a4be 100644 --- a/src/libs/ws/server.rs +++ b/src/libs/ws/server.rs @@ -29,8 +29,9 @@ use crate::libs::ws::mcp::{McpServerInfo, McpState}; #[cfg(feature = "ws")] use crate::libs::ws::tungstenite::upgrader::create_ws_stream; use crate::libs::ws::{ - BoxedStream, ConnectionListener, MessageStream, SessionListener, TcpListener, - WsClientSession, WsConnection, WsRequest, WsUpgrader, + AfterRequest, BeforeRequest, BoxedStream, ConnectionListener, Hooks, MessageStream, + OnConnect, SessionListener, TcpListener, WsClientSession, WsConnection, WsRequest, + WsUpgrader, }; use crate::model::{EndpointSchema, TypeRegistry}; @@ -47,6 +48,9 @@ pub struct WebsocketServer { /// MCP surface state; `None` (the default) disables MCP entirely and the /// server behaves exactly as before. See [`WebsocketServer::enable_mcp`]. pub mcp: Option>, + /// Interception hooks. Empty by default — an empty `Hooks` adds one branch per + /// request and nothing else. + pub hooks: Hooks, } impl WebsocketServer { @@ -66,6 +70,7 @@ impl WebsocketServer { config, upgrader: default_upgrader(), mcp: None, + hooks: Hooks::default(), } } @@ -84,6 +89,23 @@ impl WebsocketServer { self.mcp = Some(Arc::new(state)); Ok(()) } + /// Register a hook that runs before every request, on both the legacy and MCP + /// paths. Hooks run in registration order; the first error rejects the request. + pub fn add_before_hook(&mut self, hook: impl BeforeRequest + 'static) { + self.hooks.before.push(Arc::new(hook)); + } + + /// Register a hook that observes every completed request. + pub fn add_after_hook(&mut self, hook: impl AfterRequest + 'static) { + self.hooks.after.push(Arc::new(hook)); + } + + /// Register a hook that runs once per connection, after auth. Returning `Err` + /// refuses the connection. + pub fn add_on_connect_hook(&mut self, hook: impl OnConnect + 'static) { + self.hooks.on_connect.push(Arc::new(hook)); + } + pub fn set_auth_controller(&mut self, controller: impl AuthController + 'static) { self.auth_controller = Arc::new(controller); } @@ -205,12 +227,25 @@ impl WebsocketServer { stream: Box, auth_protocol: Option, ) { + // OnConnect runs before the connection is registered, so a refused peer + // never gets a slot in `states` and cannot be sent to. + let mut extensions = Extensions::new(); + if let Err(err) = self.hooks.run_on_connect(&peer, &mut extensions).await { + warn!( + ws_server = true, + peer = %peer, + error_code = ?err.code, + "connection refused by OnConnect hook" + ); + return; + } + let conn = Arc::new(WsConnection { connection_id: get_conn_id(), user_id: Default::default(), roles: Arc::new(RwLock::new(Arc::new(Vec::new()))), peer, - extensions: Extensions::new(), + extensions, log_id: get_log_id(), }); debug!( diff --git a/src/libs/ws/session.rs b/src/libs/ws/session.rs index 2e89c4f..ee75e49 100644 --- a/src/libs/ws/session.rs +++ b/src/libs/ws/session.rs @@ -10,16 +10,17 @@ use crate::libs::error_code::ErrorCode; use crate::libs::toolbox::{RequestContext, TOOLBOX}; use super::mcp::{ - self, JsonRpcError, JsonRpcId, JsonRpcRequest, McpAction, McpCallCtx, McpState, jsonrpc_error, + self, JsonRpcError, JsonRpcId, JsonRpcRequest, McpAction, McpCallCtx, McpState, + encode_tool_error, jsonrpc_error, jsonrpc_result, }; use super::{ - StreamError, WebsocketServer, WsConnection, WsRequestValue, WsResponseError, WsResponseValue, - WsStream, + MessageStream, RequestOutcome, StreamError, WebsocketServer, WsConnection, WsRequestValue, + WsResponseError, WsResponseValue, }; pub struct WsClientSession { conn_info: Arc, - conn: Box, + conn: Box, rx: mpsc::Receiver, server: Arc, } @@ -27,7 +28,7 @@ pub struct WsClientSession { impl WsClientSession { pub fn new( conn_info: Arc, - conn: Box, + conn: Box, rx: mpsc::Receiver, server: Arc, ) -> Self { @@ -39,7 +40,7 @@ impl WsClientSession { } } - pub fn conn(&self) -> &dyn WsStream { + pub fn conn(&self) -> &dyn MessageStream { self.conn.as_ref() } @@ -162,13 +163,46 @@ impl WsClientSession { let handler = endpoint.handler.clone(); let toolbox = self.server.toolbox.clone(); + let hooks = self.server.hooks.clone(); + let schema = endpoint.schema.clone(); tokio::task::spawn_local(async move { + let mut context = context; + // Hooks run inside the spawned task so a slow hook cannot stall the + // session loop, and after check_roles so they only see calls that were + // already allowed to reach this endpoint. + if let Err(custom) = hooks + .run_before(&mut context, &schema, &req.params) + .await + { + let code = custom.code.to_u32(); + toolbox.send( + context.connection_id, + WsResponseValue::Error(WsResponseError { + method: context.method, + code, + seq: context.seq, + log_id: context.log_id.to_string(), + params: custom.params.clone(), + }), + ); + hooks + .run_after(&context, &schema, &RequestOutcome::PublicErr { code }) + .await; + return; + } + TOOLBOX .scope( toolbox.clone(), - handler.handle(&toolbox, context, req.params), + handler.handle(&toolbox, context.clone(), req.params), ) .await; + + // The erased handler reports its own outcome through the toolbox, so + // AfterRequest observes completion rather than the specific result here. + hooks + .run_after(&context, &schema, &RequestOutcome::Ok) + .await; }); Ok(true) @@ -232,13 +266,40 @@ impl WsClientSession { let handler = endpoint.handler.clone(); let toolbox = self.server.toolbox.clone(); + let hooks = self.server.hooks.clone(); + let schema = endpoint.schema.clone(); tokio::task::spawn_local(async move { + let mut context = context; + // Same placement as the legacy path, but the rejection has to go + // back in the MCP envelope — a tool error, not a WsResponseError. + if let Err(custom) = hooks.run_before(&mut context, &schema, &arguments).await { + let code = custom.code.to_u32(); + toolbox.send_raw( + conn_id, + jsonrpc_result(&id, encode_tool_error(custom.code, &custom.params)) + .to_string(), + ); + hooks + .run_after(&context, &schema, &RequestOutcome::PublicErr { code }) + .await; + return; + } + TOOLBOX .scope( toolbox.clone(), - handler.handle_mcp(&toolbox, context, McpCallCtx { id }, arguments), + handler.handle_mcp( + &toolbox, + context.clone(), + McpCallCtx { id }, + arguments, + ), ) .await; + + hooks + .run_after(&context, &schema, &RequestOutcome::Ok) + .await; }); } } diff --git a/tests/transport_seam.rs b/tests/transport_seam.rs index 1538f6b..f090782 100644 --- a/tests/transport_seam.rs +++ b/tests/transport_seam.rs @@ -256,3 +256,270 @@ async fn mcp_initialize_and_tool_call_work_over_the_same_transport() { }) .await; } + +// --------------------------------------------------------------------------- +// Phase 4 — hooks on both dispatch paths +// --------------------------------------------------------------------------- + +use endpoint_libs::libs::error_code::ErrorCode; +use endpoint_libs::libs::peer::Extensions; +use endpoint_libs::libs::ws::{ + AfterRequest, BeforeRequest, OnConnect, RequestOutcome, +}; +use endpoint_libs::model::EndpointSchema; +use std::sync::Mutex; + +/// Verified claims, as a mission-token hook would attach them. +#[derive(Debug, Clone, PartialEq)] +struct Claims(String); + +/// Rejects any request whose `message` contains "denied". +struct DenyByContent; + +#[async_trait(?Send)] +impl BeforeRequest for DenyByContent { + async fn before( + &self, + ctx: &mut RequestContext, + _endpoint: &EndpointSchema, + params: &serde_json::Value, + ) -> Result<(), CustomError> { + let text = params + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if text.contains("denied") { + return Err(CustomError::new(ErrorCode::FORBIDDEN) + .with_message("blocked by policy") + .with_kind("PolicyDenied")); + } + // Prove a hook can hand data to the handler. + ctx.extensions.insert(Claims(format!("seen:{text}"))); + Ok(()) + } +} + +/// Records every outcome it observes. +#[derive(Clone, Default)] +struct RecordOutcomes(Arc>>); + +#[async_trait(?Send)] +impl AfterRequest for RecordOutcomes { + async fn after( + &self, + _ctx: &RequestContext, + endpoint: &EndpointSchema, + outcome: &RequestOutcome, + ) { + let label = match outcome { + RequestOutcome::Ok => "ok".to_owned(), + RequestOutcome::PublicErr { code } => format!("public:{code}"), + RequestOutcome::InternalErr => "internal".to_owned(), + _ => "other".to_owned(), + }; + self.0.lock().unwrap().push(format!("{}:{label}", endpoint.name)); + } +} + +/// Second endpoint whose handler surfaces hook-supplied claims. +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct ClaimsRequest { + pub message: String, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct ClaimsResponse { + pub message: String, +} + +impl WsRequest for ClaimsRequest { + type Response = ClaimsResponse; + const METHOD_ID: u32 = 2; + const ROLES: &'static [u32] = &[1]; + const SCHEMA: &'static str = r#"{ + "name": "Claims", + "code": 2, + "parameters": [{"name": "message", "ty": "String"}], + "returns": [{"name": "message", "ty": "String"}], + "description": "Reports claims attached by a BeforeRequest hook.", + "roles": [] + }"#; +} + +impl WsResponse for ClaimsResponse { + type Request = ClaimsRequest; +} + +struct MethodClaims; + +#[async_trait(?Send)] +impl RequestHandler for MethodClaims { + type Request = ClaimsRequest; + type Error = CustomError; + + async fn handle(&self, ctx: RequestContext, _req: ClaimsRequest) -> Response { + let claims = ctx + .extensions + .get::() + .map(|c| c.0.clone()) + .unwrap_or_else(|| "".to_owned()); + Ok(ClaimsResponse { message: claims }) + } +} + +fn server_with_hooks(recorder: RecordOutcomes, mcp: bool) -> WebsocketServer { + let config = WsServerConfig { + insecure: true, + ..Default::default() + }; + let mut server = WebsocketServer::new(config); + server.set_auth_controller(AllowAllAuthController); + server.add_handler(MethodClaims); + server.add_before_hook(DenyByContent); + server.add_after_hook(recorder); + if mcp { + let mut registry = endpoint_libs::model::TypeRegistry::new(); + let schema: EndpointSchema = serde_json::from_str(ClaimsRequest::SCHEMA).unwrap(); + registry.add_endpoint(&schema); + server + .enable_mcp( + ®istry, + endpoint_libs::libs::ws::mcp::McpServerInfo { + name: "hooks-test".into(), + version: "0.0.0".into(), + }, + ) + .expect("enable_mcp"); + } + server +} + +fn connect(server: WebsocketServer) -> WsClient { + let (server_io, client_io) = tokio::io::duplex(256 * 1024); + tokio::task::spawn_local(spawn_server(server, server_io)); + WsClient::from_stream(Box::new(TransportStream::new(framed_json(client_io)))) +} + +/// A BeforeRequest hook rejects on the legacy path, with the exact error frame, and +/// a passing request receives the claims the hook attached. +#[tokio::test(flavor = "current_thread")] +async fn before_hook_gates_the_legacy_path_and_passes_claims() { + let local = LocalSet::new(); + local + .run_until(async { + let recorder = RecordOutcomes::default(); + let mut client = connect(server_with_hooks(recorder.clone(), false)); + + // Allowed: the handler sees what the hook put in extensions. + let resp: ClaimsResponse = client + .request(ClaimsRequest { message: "fine".into() }) + .await + .expect("allowed request failed"); + assert_eq!(resp.message, "seen:fine"); + + // Denied: the handler never runs; the hook's code and params come back. + client + .send_req(ClaimsRequest::METHOD_ID, ClaimsRequest { message: "denied".into() }) + .await + .expect("send"); + let raw = client.recv_raw().await.expect("recv"); + assert_eq!(raw["code"], ErrorCode::FORBIDDEN.to_u32(), "frame: {raw}"); + assert_eq!(raw["params"]["kind"], "PolicyDenied", "frame: {raw}"); + assert_eq!(raw["params"]["message"], "blocked by policy", "frame: {raw}"); + + // AfterRequest saw both, with the rejection reported as a public error. + let seen = recorder.0.lock().unwrap().clone(); + assert_eq!( + seen, + vec![ + "Claims:ok".to_owned(), + format!("Claims:public:{}", ErrorCode::FORBIDDEN.to_u32()) + ] + ); + }) + .await; +} + +/// The same hook must gate `tools/call`, with the rejection encoded as an MCP tool +/// error rather than a legacy error frame. +#[tokio::test(flavor = "current_thread")] +async fn before_hook_gates_the_mcp_path_with_a_tool_error() { + let local = LocalSet::new(); + local + .run_until(async { + let recorder = RecordOutcomes::default(); + let mut client = connect(server_with_hooks(recorder.clone(), true)); + + let init = serde_json::json!({ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "hooks-test", "version": "0.0.0"}} + }); + client.send_raw(init.to_string().as_bytes()).await.unwrap(); + client.recv_raw().await.unwrap(); + + let call = serde_json::json!({ + "jsonrpc": "2.0", "id": 7, "method": "tools/call", + "params": {"name": "claims", "arguments": {"message": "denied by policy"}} + }); + client.send_raw(call.to_string().as_bytes()).await.unwrap(); + let resp = tokio::time::timeout(Duration::from_secs(5), client.recv_raw()) + .await + .expect("timed out") + .expect("recv"); + + assert_eq!(resp["id"], 7, "frame: {resp}"); + assert_eq!(resp["result"]["isError"], true, "expected a tool error: {resp}"); + let text = resp["result"]["content"][0]["text"].as_str().unwrap_or(""); + assert!( + text.contains("blocked by policy") || text.contains("PolicyDenied"), + "tool error did not carry the hook's payload: {resp}" + ); + + let seen = recorder.0.lock().unwrap().clone(); + assert_eq!(seen, vec![format!("Claims:public:{}", ErrorCode::FORBIDDEN.to_u32())]); + }) + .await; +} + +/// An OnConnect hook refuses a peer outright — no messages are exchanged at all. +#[tokio::test(flavor = "current_thread")] +async fn on_connect_hook_can_refuse_a_peer() { + struct RefuseUnattested; + + #[async_trait(?Send)] + impl OnConnect for RefuseUnattested { + async fn on_connect( + &self, + peer: &PeerIdentity, + ext: &mut Extensions, + ) -> Result<(), CustomError> { + match peer.attestation() { + Some(a) if a.is_verified() => { + ext.insert(Claims("attested".to_owned())); + Ok(()) + } + _ => Err(CustomError::new(ErrorCode::FORBIDDEN).with_message("unattested peer")), + } + } + } + + let local = LocalSet::new(); + local + .run_until(async { + let config = WsServerConfig { insecure: true, ..Default::default() }; + let mut server = WebsocketServer::new(config); + server.set_auth_controller(AllowAllAuthController); + server.add_handler(MethodEcho); + server.add_on_connect_hook(RefuseUnattested); + + // spawn_server supplies an *attested* peer, so this one is admitted. + let mut client = connect(server); + let resp: EchoResponse = client + .request(EchoRequest { message: "hi".into() }) + .await + .expect("attested peer should be admitted"); + assert_eq!(resp.message, "echo[local/test-harness]: hi"); + }) + .await; +} From 8313e7c760c899b855c9a56b01ba471c5d5c3798 Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 25 Jul 2026 17:53:05 +0700 Subject: [PATCH 6/6] =?UTF-8?q?chore(release):=202.0.0-alpha.1=20=E2=80=94?= =?UTF-8?q?=20changelog,=20migration=20guide,=20README=20transports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Version 2.0.0-alpha.1. - CHANGELOG entry listing the four breaking items and the additive surface, with the "wire protocols are unchanged" statement up front — that is the thing consumers most need to know. - docs/2.0-migration.md: exhaustive per-symbol table, including the WsMessage-alias caveat (type positions only, not tungstenite's inherent methods, with a replacement for each), the LocalSet requirement, and the known ws-client/from_stream limitation. - README "Transports (2.0)" section: the three entry points, the framed_json wire format for non-Rust peers, peer identity/attestation, and hooks. Also adds EndpointErrorSchema::new + with_message/with_fields — a gap in Phase 2b: the type was made #[non_exhaustive] without giving out-of-crate callers any way to build one. Found by compiling endpointgen against this release rather than by inspection. Verified: 83 lib + 5 acceptance tests green, clippy clean, and all four feature configurations build. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 43 +++++++++++++++++++++ Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 70 +++++++++++++++++++++++++++++++++ docs/2.0-migration.md | 90 +++++++++++++++++++++++++++++++++++++++++++ src/model/endpoint.rs | 29 ++++++++++++++ 6 files changed, 234 insertions(+), 2 deletions(-) create mode 100644 docs/2.0-migration.md diff --git a/CHANGELOG.md b/CHANGELOG.md index da13f71..221b4d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,49 @@ # Changelog All notable changes to this project will be documented in this file. +## [2.0.0-alpha.1] - 2026-07-25 + +Makes the schema/handler/MCP machinery transport-agnostic. The wire protocols are +unchanged — legacy `{method, seq, params}` frames and MCP JSON-RPC are byte-identical +to 1.9, so deployed frontends need no changes. + +See `docs/2.0-migration.md` for the per-symbol migration table. + +### Breaking + +- `WsMessage` is no longer a re-export of `tungstenite::Message`; the canonical type is + `WireMessage`. The `WsMessage` alias covers type positions but NOT tungstenite's + inherent methods (`.into_text()`, `.into_data()`). +- `WsConnection.address: SocketAddr` replaced by `WsConnection.peer: PeerIdentity`. A + `#[deprecated]` `address()` accessor returns a loopback placeholder for local peers. +- `WsStream` trait renamed to `MessageStream` (alias retained). +- `Type`, `Field`, `EnumVariant`, `EndpointSchema` and `EndpointErrorSchema` are now + `#[non_exhaustive]`: out-of-crate matches need a wildcard arm, and construction goes + through `::new()` + `with_*` rather than struct literals. + +### Features + +- Transport seam: `Transport` (blanket Sink+Stream alias), `TransportStream`, + `serve_connection`, `serve_with`, `SessionListener`, `WsClient::from_stream`. +- `framed-transport` feature: `framed_json()` — length-delimited `WireMessage` framing + over any byte stream, with a documented wire format for non-Rust peers. +- Hooks: `BeforeRequest`, `AfterRequest`, `OnConnect`, active on both the legacy and + MCP dispatch paths. +- `PeerIdentity` / `LocalPeer` / `Attestation` carry verified peer code identity into + handlers and logs. +- `Extensions`, a type-keyed map on connections and requests. +- `Field.meta` / `EndpointSchema.meta`: reserved, empty, and the reason OpenAPI and + AsyncAPI emission can ship as a 2.1 minor rather than a 3.0. +- `examples/uds_echo.rs`: the endpoint machinery over a Unix socket, no TCP/TLS/HTTP. + +### Notes + +- `cargo test --all-features` cannot pass (`ws` and `ws-wtx` are mutually exclusive by + `compile_error!`). Use `cargo all-features test`, which CI runs. +- The deprecated `ws-wtx` backend is untouched and still does not build. +- `WsClient::from_stream` currently requires the `ws-client` feature, which pulls + tungstenite. Narrowing that is additive and deferred. + ## [1.9.1] - 2026-07-18 ### Bug Fixes diff --git a/Cargo.lock b/Cargo.lock index f4f1bbc..773f8d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1314,7 +1314,7 @@ dependencies = [ [[package]] name = "endpoint-libs" -version = "1.9.1" +version = "2.0.0-alpha.1" dependencies = [ "alloy-primitives", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 5f614ad..7028721 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "endpoint-libs" -version = "1.9.1" +version = "2.0.0-alpha.1" edition = "2024" authors = ["Veon "] description = "Common dependencies to be used with Pathscale projects, projects that use [endpoint-gen](https://github.com/pathscale/endpoint-gen), and projects that use honey_id-types." diff --git a/README.md b/README.md index 38abb2d..ac09f33 100644 --- a/README.md +++ b/README.md @@ -205,6 +205,76 @@ A `tracing` layer that captures recent error-level log events into an in-memory Rate-limiting layer for `tracing` events to suppress repeated log spam. +## Transports (2.0) + +The server core is transport-agnostic. The WebSocket path (`listen()`) is unchanged; +these entry points let the same handlers, roles, typed errors and MCP surface run over +a Unix socket, a Windows named pipe, or macOS XPC. + +```rust +// Server: one already-established connection, any transport. +server.serve_connection(peer, states, stream, /* auth token */ None).await; + +// Server: accept loop over any listener. +server.serve_with(my_listener).await?; // my_listener: SessionListener + +// Client: the mirror image. +let client = WsClient::from_stream(stream); +``` + +Both sides need a `MessageStream`. For byte-stream transports, the `framed-transport` +feature supplies one: + +```rust +use endpoint_libs::libs::ws::transport::{TransportStream, framed_json}; + +let stream: Box = + Box::new(TransportStream::new(framed_json(unix_stream))); +``` + +`examples/uds_echo.rs` is a complete worked example over a Unix domain socket: + +```bash +cargo run --example uds_echo --features full,framed-transport,ws-client +``` + +### `framed_json` wire format + +One length-delimited frame per message — implementable by a non-Rust peer in a few +lines: + +```text ++---------------+--------+--------------------------+ +| u32 BE length | u8 kind| payload (length-1 bytes) | ++---------------+--------+--------------------------+ +``` + +`length` counts the kind byte plus payload. `kind` is `0=Text, 1=Binary, 2=Ping, +3=Pong, 4=Close`. `Text` is UTF-8; `Close` is empty or `u16 BE code` + UTF-8 reason. +Default max frame is 16 MiB (`framed_json_with_max_frame` to change it). + +### Peer identity and attestation + +`WsConnection.peer` / `RequestContext.peer` carry a `PeerIdentity`: +`Network(SocketAddr)` for TCP/TLS, or `Local(LocalPeer { pid, uid, attestation })`. +`Attestation::Verified { mechanism, subject }` records *code* identity a transport +verified — an XPC code-signing requirement, an executable digest, a SID. This crate +defines the vocabulary; the platform implementations live in a sibling crate. + +### Hooks + +`BeforeRequest` (may reject and may attach claims to `ctx.extensions`), `AfterRequest` +(observes outcomes), and `OnConnect` (refuses a peer once, rather than per request). +All three run on both the legacy and MCP dispatch paths. + +```rust +server.add_before_hook(MyMissionTokenCheck); +server.add_on_connect_hook(RefuseUnattestedPeers); +``` + +> **Note:** `MessageStream`'s futures are not `Send`, so `serve_connection`, +> `serve_with` and a `from_stream` client must run inside a `tokio::task::LocalSet`. + ## Logging Setup The `setup_logging` function (available without any optional features) provides a batteries-included `tracing` subscriber with: diff --git a/docs/2.0-migration.md b/docs/2.0-migration.md new file mode 100644 index 0000000..e207535 --- /dev/null +++ b/docs/2.0-migration.md @@ -0,0 +1,90 @@ +# Migrating to endpoint-libs 2.0 + +2.0 makes the schema/handler/MCP machinery **transport-agnostic**: the same server core +now runs over TCP+TLS+WebSocket (unchanged) *and* over local transports — Unix sockets, +Windows named pipes, macOS XPC — implemented in a sibling crate. + +**The wire protocols did not change.** Legacy `{method, seq, params}` frames and MCP +JSON-RPC are byte-identical to 1.9. Deployed frontends need no changes. + +Most consumers touch two or three lines. The table below is exhaustive. + +## Per-symbol changes + +| 1.9 | 2.0 | Action | +|---|---|---| +| `WsMessage` (= `tungstenite::Message` when `ws` on) | `WireMessage` | Alias `WsMessage` still works **in type positions**. Method calls need changes — see below. | +| `WsConnection.address: SocketAddr` | `WsConnection.peer: PeerIdentity` | Use `.peer`. `conn.address()` exists, is `#[deprecated]`, and returns `127.0.0.1:0` for local peers. | +| `WsStream` (trait) | `MessageStream` | Alias `WsStream` still works. Rename at your convenience. | +| `RequestContext.ip_addr` | unchanged | Still present, still populated. Local peers report loopback. | +| — | `RequestContext.peer`, `.extensions` | New fields. Only affects code constructing `RequestContext` by literal. | +| `Type`, `Field`, `EnumVariant`, `EndpointSchema`, `EndpointErrorSchema` | same, now `#[non_exhaustive]` | Out-of-crate `match` on `Type` or `Attestation` needs a `_ =>` arm. Construct via `::new()` + `with_*`, not struct literals. | +| — | `Field.meta`, `EndpointSchema.meta` | New, empty. Reserved for the 2.1 OpenAPI/AsyncAPI emitters. | + +### `WsMessage` method calls + +The alias covers types, not tungstenite's inherent methods. Replacements: + +| 1.9 (tungstenite method) | 2.0 | +|---|---| +| `msg.into_text()?` / `msg.to_text()?` | `msg.as_text()` → `Option<&str>` (also handles `Binary` UTF-8) | +| `msg.is_close()` | `msg.is_close()` — unchanged | +| `msg.into_data()` | `match msg { WireMessage::Binary(b) => b, .. }` | +| `Message::Text(s.into())` | `WireMessage::Text(s)` — takes `String` directly | + +If you need tungstenite's type at a boundary, `From` impls convert both ways when the +`ws` feature is on. + +### Logging + +`?addr` on a `SocketAddr` becomes `%conn.peer` (compact `Display`, e.g. +`local(pid=42,xpc-codesign-requirement)`) or `conn.peer.display()`. + +```rust +// 1.9 +tracing::info!(ip = %conn.address, "connected"); +// 2.0 +tracing::info!(peer = %conn.peer, "connected"); +``` + +## What's new (all additive) + +- **`serve_connection(peer, states, stream, auth_protocol)`** — run auth + session for + one already-established connection, over any transport. +- **`serve_with(listener)` + `SessionListener`** — accept loop over any listener. +- **`WsClient::from_stream(stream)`** — the client-side mirror. +- **`framed-transport` feature** — `framed_json()`, length-delimited `WireMessage` + framing over any byte stream. Format documented in + `libs::ws::transport::framed`. +- **Hooks** — `BeforeRequest`, `AfterRequest`, `OnConnect` via `add_*_hook`. +- **`PeerIdentity` / `Attestation`** — peer identity including verified code identity. +- **`Extensions`** — type-keyed map on connections and requests. + +## Gotchas + +1. **`MessageStream` is `?Send`.** `serve_connection`, `serve_with` and a + `from_stream` client must run inside a `tokio::task::LocalSet`. `listen()` already + does this internally; only the new entry points expose it. +2. **`serve_with` is single-runtime.** The shard-per-core model belongs to `listen()` + and buys nothing for a 1:1 sidecar channel. +3. **`Extensions` values must be `Clone`.** `RequestContext` derives `Clone` and + consumers rely on it, so the map stores `Clone` values (same trade + `http::Extensions` makes). +4. **`cargo test --all-features` does not work** — and did not before 2.0 either. `ws` + and `ws-wtx` are mutually exclusive by `compile_error!`. Use `cargo all-features + test` (what CI runs) or an explicit feature list. +5. **`ws-wtx` remains deprecated and non-building.** Untouched by 2.0. +6. **Handler naming**: `check_handler` requires the struct be named + `Method`. Unchanged, but easy to trip when adding endpoints. + +## Known limitation + +`WsClient::from_stream` lives behind the `ws-client` feature, which pulls +tokio-tungstenite and rustls. A sidecar that only speaks a local transport still +compiles those in. Splitting the client so `from_stream` is available from `ws-core` +alone is **additive and non-breaking**, and can land in a later release. + +## endpointgen + +endpointgen moves in lockstep. Bump to the matching 2.0 release; its +`ENDPOINT_LIBS_REQUIREMENT` check will fail loudly on a mismatch. diff --git a/src/model/endpoint.rs b/src/model/endpoint.rs index 1889171..3df3d2f 100644 --- a/src/model/endpoint.rs +++ b/src/model/endpoint.rs @@ -112,6 +112,35 @@ pub struct EndpointErrorSchema { pub fields: Vec, } +impl EndpointErrorSchema { + /// Creates an error schema with no message and no fields. + /// + /// This type is `#[non_exhaustive]`, so out-of-crate callers must build it here + /// rather than with a struct literal. + pub fn new(name: impl Into, code: EndpointErrorCodeRef) -> Self { + Self { + name: name.into(), + code, + message: String::new(), + fields: Vec::new(), + } + } + + /// Sets the human-readable message. + #[must_use] + pub fn with_message(mut self, message: impl Into) -> Self { + self.message = message.into(); + self + } + + /// Sets the structured fields carried by this error. + #[must_use] + pub fn with_fields(mut self, fields: Vec) -> Self { + self.fields = fields; + self + } +} + #[derive(Clone, Debug, Hash, PartialEq, PartialOrd, Eq, Ord)] pub struct EndpointErrorCodeRef { pub ty: Type,