From 35a7547b78d50771bc19de3f88b6c541923c15db Mon Sep 17 00:00:00 2001 From: ContextVM Date: Wed, 8 Jul 2026 17:50:27 +0200 Subject: [PATCH 1/4] feat(server): inject caller's Nostr pubkey into request extensions Add `ClientPubkey` type and inject the caller's Nostr public key (hex) into every inbound request's `extensions` typemap. This allows tool/resource/prompt handlers to identify the caller via `ctx.extensions.get::()`. The feature is always on, unlike the TS adapter's opt-in `_meta` field, and uses rmcp's typed extensions (local-only). The inbound event id remains accessible as `ctx.id`. --- CHANGELOG.md | 12 ++++++++++++ docs/server-transport.md | 30 ++++++++++++++++++++++++++++++ src/lib.rs | 2 +- src/rmcp_transport/worker.rs | 24 +++++++++++++++++------- src/transport/server/mod.rs | 23 +++++++++++++++++++++++ tests/e2e_happy_path.rs | 33 ++++++++++++++++++++++++++++++--- 6 files changed, 113 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e01b8b2..ebd22e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## [Unreleased] + +### Added + +- `ClientPubkey`: the rmcp server worker now injects the caller's Nostr public + key (hex) into every inbound request's `extensions` typemap, so tool/resource/ + prompt handlers can identify their caller via `ctx.extensions.get::()`. + This closes the parity gap with the TS adapter's `extra._meta.clientPubkey`, but + uses rmcp's typed extensions (local-only, never on the wire) instead of the + `_meta` field, so it is always on rather than opt-in. The inbound event id is + already reachable as the rmcp request id (`ctx.id`). + ## [0.2.0] - 2026-06-24 ### Added diff --git a/docs/server-transport.md b/docs/server-transport.md index c54ed9a..64c1221 100644 --- a/docs/server-transport.md +++ b/docs/server-transport.md @@ -166,6 +166,36 @@ after the stream closes. Open-stream is disabled by default; enable it with `with_open_stream(OpenStreamConfig::enabled())`. See [open-stream.md](open-stream.md) for a full example. +## Identifying the calling client + +Every inbound request carries its caller's Nostr public key. The rmcp worker +injects it into the request `extensions`, so a tool, resource, or prompt handler +reads it from `ctx.extensions`: + +```rust +use contextvm_sdk::transport::server::ClientPubkey; +use rmcp::handler::server::wrapper::Parameters; +use rmcp::model::{CallToolResult, Content, ErrorData}; +use rmcp::service::RequestContext; +use rmcp::{tool, RoleServer}; + +#[tool(description = "Echo the caller's pubkey")] +async fn whoami( + &self, + ctx: RequestContext, +) -> Result { + let pk = ctx.extensions.get::().map(|c| c.0.clone()); + Ok(CallToolResult::success(vec![Content::text(pk.unwrap_or_default())])) +} +``` + +`ClientPubkey` lives in `contextvm_sdk::transport::server` (also re-exported at +the crate root). It is present on every real inbound request, so retrieving it +never returns `None` for a genuine client call. This mirrors the TS adapter's +`extra._meta.clientPubkey`, but via rmcp's typed extensions instead of the +on-wire `_meta` field. The inbound Nostr event id is available separately as +`ctx.id` (the worker rewrites the request id to the event id). + ## When to use this instead of the gateway Use this page's approach when you are writing a new Rust MCP server. diff --git a/src/lib.rs b/src/lib.rs index 0cad02b..20cffbe 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -83,7 +83,7 @@ pub use transport::discovery_tags::{DiscoveredPeerCapabilities, PeerCapabilities // ── Transport (server) ────────────────────────────────────────────── pub use transport::server::{ - IncomingRequest, NostrServerTransport, NostrServerTransportConfig, RouteEntry, + ClientPubkey, IncomingRequest, NostrServerTransport, NostrServerTransportConfig, RouteEntry, ServerEventRouteStore, SessionSnapshot, SessionStore, }; diff --git a/src/rmcp_transport/worker.rs b/src/rmcp_transport/worker.rs index 27d7dab..ea3ba08 100644 --- a/src/rmcp_transport/worker.rs +++ b/src/rmcp_transport/worker.rs @@ -7,7 +7,7 @@ use crate::core::constants::ANNOUNCEMENT_REQUEST_ID; use crate::core::error::Result; use crate::core::types::{JsonRpcMessage, JsonRpcNotification, JsonRpcRequest}; use crate::transport::client::{NostrClientTransport, NostrClientTransportConfig}; -use crate::transport::server::{NostrServerTransport, NostrServerTransportConfig}; +use crate::transport::server::{ClientPubkey, NostrServerTransport, NostrServerTransportConfig}; use rmcp::model::GetExtensions; use rmcp::transport::worker::{Worker, WorkerContext, WorkerQuitReason}; use std::collections::HashSet; @@ -206,13 +206,23 @@ impl Worker for NostrServerWorker { } if let Some(mut rmcp_msg) = internal_to_rmcp_server_rx(&message) { - // CEP-41: inject the open-stream writer into the - // request's `extensions` typemap so the tool handler can - // reach it via `ctx.extensions.get::()`. - // The rmcp service loop swaps these extensions straight into - // the handler's `RequestContext` before dispatch. No-op when - // open-stream is disabled or the request has no writer. + // Inject caller identity + (CEP-41) the open-stream writer + // into the request's `extensions` typemap so the handler can + // reach them via `ctx.extensions.get::()`. The rmcp service + // loop swaps these extensions straight into the handler's + // `RequestContext` before dispatch. if let rmcp::model::JsonRpcMessage::Request(ref mut jr) = rmcp_msg { + // Caller pubkey is relevant to every request (auth, + // payments, logging) and extensions are local-only, so + // inject it unconditionally. This mirrors the TS adapter's + // `extra._meta.clientPubkey`, but via rmcp's typed + // extensions rather than the on-wire `_meta` field (which + // is why the TS knob is opt-in but this is not). + jr.request + .extensions_mut() + .insert(ClientPubkey(client_pubkey.clone())); + // CEP-41: no-op when open-stream is disabled or the + // request has no writer. if let Some(writer) = self.transport.get_open_stream_writer(&event_id) { diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs index 4841be2..c1c9045 100644 --- a/src/transport/server/mod.rs +++ b/src/transport/server/mod.rs @@ -439,6 +439,29 @@ pub struct IncomingRequest { pub is_encrypted: bool, } +/// The Nostr public key (hex) of the client that issued the current request. +/// +/// [`NostrServerWorker`](crate::rmcp_transport::NostrServerWorker) injects this +/// into every inbound request's rmcp `extensions` typemap, so a tool, resource, +/// or prompt handler can identify its caller: +/// +/// ```ignore +/// use contextvm_sdk::transport::server::ClientPubkey; +/// use rmcp::service::RequestContext; +/// use rmcp::RoleServer; +/// +/// let caller = ctx.extensions.get::().map(|c| c.0.clone()); +/// ``` +/// +/// The value is the raw hex pubkey (the same string as +/// [`IncomingRequest::client_pubkey`]). It is present on every real inbound +/// request, unlike `OpenStreamWriter`, which is injected only for open-stream +/// `tools/call` calls. This is purely a local affordance — it is never +/// serialized onto the wire. The inbound Nostr event id is available separately +/// as the rmcp request id (`ctx.id`), which the worker rewrites to the event id. +#[derive(Debug, Clone)] +pub struct ClientPubkey(pub String); + impl NostrServerTransport { /// Create a new server transport. pub async fn new(signer: T, config: NostrServerTransportConfig) -> Result diff --git a/tests/e2e_happy_path.rs b/tests/e2e_happy_path.rs index 529dea8..fcd5a23 100644 --- a/tests/e2e_happy_path.rs +++ b/tests/e2e_happy_path.rs @@ -12,7 +12,9 @@ use std::time::Duration; use contextvm_sdk::core::types::EncryptionMode; use contextvm_sdk::relay::mock::MockRelayPool; use contextvm_sdk::transport::client::{NostrClientTransport, NostrClientTransportConfig}; -use contextvm_sdk::transport::server::{NostrServerTransport, NostrServerTransportConfig}; +use contextvm_sdk::transport::server::{ + ClientPubkey, NostrServerTransport, NostrServerTransportConfig, +}; use contextvm_sdk::RelayPoolTrait; use rmcp::{ @@ -79,6 +81,18 @@ impl DemoServer { "Total echo calls: {n}" ))])) } + + /// Self-check for caller-identity injection: returns the `ClientPubkey` the + /// rmcp worker places in request extensions, or `"none"` if absent. + #[tool(description = "Return the caller's Nostr pubkey from request extensions")] + async fn whoami(&self, ctx: RequestContext) -> Result { + let pk = ctx + .extensions + .get::() + .map(|c| c.0.clone()) + .unwrap_or_else(|| "none".to_string()); + Ok(CallToolResult::success(vec![Content::text(pk)])) + } } #[tool_handler] @@ -159,8 +173,9 @@ fn call_params(name: &'static str, args: Option) -> CallToolR async fn run_e2e_scenario(mode: EncryptionMode) { let (client_pool, server_pool) = MockRelayPool::create_pair(); - // Extract server pubkey before as_pool() takes ownership + // Extract pubkeys before as_pool() takes ownership let server_pubkey_hex = server_pool.mock_public_key().to_hex(); + let client_pubkey_hex = client_pool.mock_public_key().to_hex(); let server_transport = NostrServerTransport::with_relay_pool( NostrServerTransportConfig::default().with_encryption_mode(mode), @@ -207,7 +222,7 @@ async fn run_e2e_scenario(mode: EncryptionMode) { ); let tools = client.list_all_tools().await.expect("list_all_tools"); - assert_eq!(tools.len(), 3, "expected 3 tools"); + assert_eq!(tools.len(), 4, "expected 4 tools"); let tool_names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect(); assert!(tool_names.contains(&"echo"), "missing echo tool"); assert!(tool_names.contains(&"add"), "missing add tool"); @@ -215,6 +230,18 @@ async fn run_e2e_scenario(mode: EncryptionMode) { tool_names.contains(&"get_echo_count"), "missing get_echo_count tool" ); + assert!(tool_names.contains(&"whoami"), "missing whoami tool"); + + // ClientPubkey injection: the worker puts the caller's pubkey in extensions. + let whoami = client + .call_tool(call_params("whoami", None)) + .await + .expect("whoami call"); + assert_eq!( + first_text(&whoami), + client_pubkey_hex, + "whoami should return the caller's pubkey from request extensions" + ); let echo1 = client .call_tool(call_params( From d911c7ee855814ca26ad5ee1b2f3edcbdd73686b Mon Sep 17 00:00:00 2001 From: ContextVM Date: Thu, 9 Jul 2026 10:57:51 +0200 Subject: [PATCH 2/4] feat(server): inject full client-signed Nostr event into extensions Expose the complete verified Nostr event via `InboundEvent` in `extensions` so handlers can access `sig`, `id`, `pubkey`, `tags`, etc. This eliminates the need for handlers to fabricate synthetic events when binding tool calls to the original publishing event (e.g., an MLS key-package coordinator returning the publication event). Also add an `event` field to `IncomingRequest` for the channel seam. FFI types (`FfiIncomingRequest`, the UniFFI `IncomingRequest`) intentionally do not mirror the new field yet because no FFI consumer requires the raw Nostr event, and exposing `nostr_sdk::Event` across the C ABI + UniFFI is non-trivial. Omission is documented in place. --- CHANGELOG.md | 16 ++ contextvm-ffi/src/types.rs | 4 + contextvm-ffi/src/uniffi_types.rs | 9 + docs/server-transport.md | 41 ++++ src/lib.rs | 4 +- src/rmcp_transport/worker.rs | 13 +- src/transport/server/announcement_manager.rs | 3 + src/transport/server/mod.rs | 217 +++++++++++-------- tests/e2e_happy_path.rs | 44 +++- 9 files changed, 261 insertions(+), 90 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ebd22e5..e085004 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,22 @@ uses rmcp's typed extensions (local-only, never on the wire) instead of the `_meta` field, so it is always on rather than opt-in. The inbound event id is already reachable as the rmcp request id (`ctx.id`). +- `InboundEvent`: the rmcp server worker now also injects the **full** + client-signed Nostr request event into `extensions`, reachable via + `ctx.extensions.get::()`. For gift-wrapped requests this is the + inner, signature-verified event (its `pubkey` matches `ClientPubkey` by + construction); for plaintext requests it is the outer event. This exposes + `id`, `pubkey`, `sig`, `tags`, … — notably `sig`, which the server cannot + reconstruct without the client's private key. Handlers that must bind a tool + call to / store / audit the publishing event (e.g. an MLS key-package + coordinator returning the publication event) no longer have to fabricate a + synthetic event. Injected only for real client requests; synthetic + transport-internal requests carry none (`get` returns `None`). +- `IncomingRequest` gained an `event: Option` field carrying + the same event through the channel seam. The FFI mirrors (`FfiIncomingRequest`, + the UniFFI `IncomingRequest`) intentionally do not surface it yet (no FFI + consumer needs the raw event; mirroring `nostr_sdk::Event` + `Tags` is + non-trivial) — the omission is documented in place. ## [0.2.0] - 2026-06-24 diff --git a/contextvm-ffi/src/types.rs b/contextvm-ffi/src/types.rs index e82c9c4..3053245 100644 --- a/contextvm-ffi/src/types.rs +++ b/contextvm-ffi/src/types.rs @@ -45,6 +45,10 @@ pub struct FfiJsonRpcMessage { } /// An FFI-safe incoming request (server-side). +/// +/// NOTE: the SDK's `IncomingRequest` also carries an `event: +/// Option` (the full client-signed event), intentionally not +/// mirrored here — see `uniffi_types::IncomingRequest` for the rationale. #[repr(C)] #[derive(Debug)] pub struct FfiIncomingRequest { diff --git a/contextvm-ffi/src/uniffi_types.rs b/contextvm-ffi/src/uniffi_types.rs index ce23fd8..f7ecf5d 100644 --- a/contextvm-ffi/src/uniffi_types.rs +++ b/contextvm-ffi/src/uniffi_types.rs @@ -53,6 +53,13 @@ pub struct JsonRpcMessage { } /// An incoming MCP request (server-side). +/// +/// NOTE: the SDK's `contextvm_sdk::IncomingRequest` also carries an +/// `event: Option` field (the full client-signed event). It +/// is intentionally NOT mirrored here: no FFI consumer needs the raw Nostr +/// event today, and `nostr_sdk::Event` (with its `Tags` collection) is +/// non-trivial to expose across the C ABI + UniFFI. Add an `Event` mirror and a +/// field here when a foreign consumer needs `sig` / event binding. #[derive(Debug, Clone, uniffi::Record)] pub struct IncomingRequest { pub message: JsonRpcMessage, @@ -231,6 +238,8 @@ fn incoming_to_uniffi(req: &contextvm_sdk::IncomingRequest) -> IncomingRequest { client_pubkey: req.client_pubkey.clone(), event_id: req.event_id.clone(), is_encrypted: req.is_encrypted, + // req.event (the full client-signed Nostr event) is deliberately not + // forwarded — see the note on `IncomingRequest` above. } } diff --git a/docs/server-transport.md b/docs/server-transport.md index 64c1221..6a8c636 100644 --- a/docs/server-transport.md +++ b/docs/server-transport.md @@ -196,6 +196,47 @@ never returns `None` for a genuine client call. This mirrors the TS adapter's on-wire `_meta` field. The inbound Nostr event id is available separately as `ctx.id` (the worker rewrites the request id to the event id). +## The inbound Nostr event + +When a handler needs the **full** client-signed request event — not just the +pubkey — the worker also injects an `InboundEvent` into `extensions`. This +exposes the event's `id`, `pubkey`, `sig`, `tags`, …, which matters when a +handler must bind a tool call to the publishing event, store it for later +return, or audit it. `sig` in particular is the client's Schnorr signature and +cannot be reconstructed by the server (it does not hold the client's private +key), so it has to be threaded through from ingest. + +```rust +use contextvm_sdk::transport::server::InboundEvent; +use rmcp::handler::server::wrapper::Parameters; +use rmcp::model::{CallToolResult, Content, ErrorData}; +use rmcp::service::RequestContext; +use rmcp::{tool, RoleServer}; + +#[tool(description = "Return the inbound event's id + sig")] +async fn audit( + &self, + ctx: RequestContext, +) -> Result { + match ctx.extensions.get::() { + Some(ev) => Ok(CallToolResult::success(vec![Content::text(format!( + "id={} sig={}", + ev.0.id.to_hex(), + ev.0.sig // Display renders hex + ))])), + None => Err(ErrorData::invalid_params("no inbound event", None)), + } +} +``` + +For gift-wrapped requests this is the **inner**, signature-verified event (the +same one whose `pubkey` is surfaced as `ClientPubkey`, so `ev.0.pubkey` and +`ClientPubkey` agree by construction); for plaintext requests it is the outer +request event. It is injected only for real client requests — synthetic +transport-internal requests (announcement / initialization drives, CEP-22 +oversized reassembly) carry no event, so `get::()` returns `None` +for them. Like `ClientPubkey`, this is local-only and never touches the wire. + ## When to use this instead of the gateway Use this page's approach when you are writing a new Rust MCP server. diff --git a/src/lib.rs b/src/lib.rs index 20cffbe..58aa375 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -83,8 +83,8 @@ pub use transport::discovery_tags::{DiscoveredPeerCapabilities, PeerCapabilities // ── Transport (server) ────────────────────────────────────────────── pub use transport::server::{ - ClientPubkey, IncomingRequest, NostrServerTransport, NostrServerTransportConfig, RouteEntry, - ServerEventRouteStore, SessionSnapshot, SessionStore, + ClientPubkey, InboundEvent, IncomingRequest, NostrServerTransport, NostrServerTransportConfig, + RouteEntry, ServerEventRouteStore, SessionSnapshot, SessionStore, }; // ── rmcp re-export ────────────────────────────────────────────────── diff --git a/src/rmcp_transport/worker.rs b/src/rmcp_transport/worker.rs index ea3ba08..549e27b 100644 --- a/src/rmcp_transport/worker.rs +++ b/src/rmcp_transport/worker.rs @@ -7,7 +7,9 @@ use crate::core::constants::ANNOUNCEMENT_REQUEST_ID; use crate::core::error::Result; use crate::core::types::{JsonRpcMessage, JsonRpcNotification, JsonRpcRequest}; use crate::transport::client::{NostrClientTransport, NostrClientTransportConfig}; -use crate::transport::server::{ClientPubkey, NostrServerTransport, NostrServerTransportConfig}; +use crate::transport::server::{ + ClientPubkey, InboundEvent, NostrServerTransport, NostrServerTransportConfig, +}; use rmcp::model::GetExtensions; use rmcp::transport::worker::{Worker, WorkerContext, WorkerQuitReason}; use std::collections::HashSet; @@ -148,6 +150,7 @@ impl Worker for NostrServerWorker { mut message, event_id, client_pubkey, + event, .. } = incoming; @@ -221,6 +224,14 @@ impl Worker for NostrServerWorker { jr.request .extensions_mut() .insert(ClientPubkey(client_pubkey.clone())); + // Full inbound event (id + sig + pubkey + …) for + // handlers that must bind to / store / audit the + // publishing event. Only real client requests carry + // one; synthetic transport requests have none, so + // handlers see `None` via get() for those. + if let Some(ev) = event { + jr.request.extensions_mut().insert(InboundEvent(ev)); + } // CEP-41: no-op when open-stream is disabled or the // request has no writer. if let Some(writer) = diff --git a/src/transport/server/announcement_manager.rs b/src/transport/server/announcement_manager.rs index c680899..fe45292 100644 --- a/src/transport/server/announcement_manager.rs +++ b/src/transport/server/announcement_manager.rs @@ -744,6 +744,7 @@ impl AnnouncementManager { client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(), event_id: ANNOUNCEMENT_REQUEST_ID.to_string(), is_encrypted: false, + event: None, }); } @@ -815,6 +816,7 @@ async fn publish_public_announcements( client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(), event_id: ANNOUNCEMENT_REQUEST_ID.to_string(), is_encrypted: false, + event: None, }) .is_err() { @@ -865,6 +867,7 @@ async fn publish_public_announcements( client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(), event_id: ANNOUNCEMENT_REQUEST_ID.to_string(), is_encrypted: false, + event: None, }); } diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs index c1c9045..80e5bde 100644 --- a/src/transport/server/mod.rs +++ b/src/transport/server/mod.rs @@ -437,6 +437,15 @@ pub struct IncomingRequest { pub event_id: String, /// Whether the original message was encrypted. pub is_encrypted: bool, + /// The inbound (client-signed) Nostr event, if this request carried one. + /// + /// `Some` for real client requests — the inner, signature-verified gift-wrap + /// event for encrypted requests, or the outer event for plaintext. `None` + /// for requests the transport synthesizes itself (announcement / + /// initialization drives, CEP-22 oversized reassembly), which carry no real + /// Nostr event. Handlers reach it via [`InboundEvent`] in request + /// `extensions`; see that type's docs. + pub event: Option, } /// The Nostr public key (hex) of the client that issued the current request. @@ -462,6 +471,37 @@ pub struct IncomingRequest { #[derive(Debug, Clone)] pub struct ClientPubkey(pub String); +/// The inbound (client-signed) Nostr event for the current request. +/// +/// Injected into a request's rmcp `extensions` typemap by +/// [`NostrServerWorker`](crate::rmcp_transport::NostrServerWorker) alongside +/// [`ClientPubkey`], so a tool, resource, or prompt handler can read the full +/// signed event — its `id`, `pubkey`, `sig`, … — to bind a tool call to the +/// publishing event, store it for later return, or audit it: +/// +/// ```ignore +/// use contextvm_sdk::transport::server::InboundEvent; +/// use rmcp::service::RequestContext; +/// use rmcp::RoleServer; +/// +/// if let Some(ev) = ctx.extensions.get::() { +/// let sig_hex = ev.0.sig.to_string(); // Schnorr signature, hex +/// // … +/// } +/// ``` +/// +/// For gift-wrapped requests this is the **inner**, signature-verified event +/// (the same one whose `pubkey` is surfaced as [`ClientPubkey`], so `ev.0.pubkey` +/// and `ClientPubkey` agree by construction). For plaintext requests it is the +/// outer request event. It is injected only for real client requests; synthetic +/// transport-internal requests carry no event, so `extensions.get::()` +/// returns `None` for them. This is purely a local affordance — it is never +/// serialized onto the wire. The event id is also reachable as the rmcp request +/// id (`ctx.id`); this type additionally exposes `sig`, which the server cannot +/// reconstruct without the client's private key. +#[derive(Debug, Clone)] +pub struct InboundEvent(pub Event); + impl NostrServerTransport { /// Create a new server transport. pub async fn new(signer: T, config: NostrServerTransportConfig) -> Result @@ -1608,107 +1648,112 @@ impl NostrServerTransport { continue; } - let (content, sender_pubkey, event_id, is_encrypted, inner_tags) = if is_gift_wrap { - if encryption_mode == EncryptionMode::Disabled { - tracing::warn!( - target: LOG_TARGET, - event_id = %event.id.to_hex(), - sender_pubkey = %event.pubkey.to_hex(), - "Received encrypted message but encryption is disabled" - ); - continue; - } - { - let guard = match seen_gift_wrap_ids.lock() { - Ok(g) => g, - Err(poisoned) => poisoned.into_inner(), - }; - if guard.contains(&event.id) { - tracing::debug!( + let (content, sender_pubkey, event_id, is_encrypted, inner_tags, inbound_event) = + if is_gift_wrap { + if encryption_mode == EncryptionMode::Disabled { + tracing::warn!( target: LOG_TARGET, event_id = %event.id.to_hex(), - "Skipping duplicate gift-wrap (outer id)" + sender_pubkey = %event.pubkey.to_hex(), + "Received encrypted message but encryption is disabled" ); continue; } - } - // Single-layer NIP-44 decrypt (matches JS/TS SDK) - let signer = match relay_pool.signer().await { - Ok(s) => s, - Err(error) => { - tracing::error!( - target: LOG_TARGET, - error = %error, - "Failed to get signer" - ); - continue; + { + let guard = match seen_gift_wrap_ids.lock() { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + }; + if guard.contains(&event.id) { + tracing::debug!( + target: LOG_TARGET, + event_id = %event.id.to_hex(), + "Skipping duplicate gift-wrap (outer id)" + ); + continue; + } } - }; - match encryption::decrypt_gift_wrap_single_layer(&signer, &event).await { - Ok(decrypted_json) => { - // The decrypted content is JSON of the inner signed event. - // Use the INNER event's ID for correlation — the client - // registers the inner event ID in its correlation store. - match serde_json::from_str::(&decrypted_json) { - Ok(inner) => { - if let Err(e) = inner.verify() { - tracing::warn!( - "Inner event signature verification failed: {e}" + // Single-layer NIP-44 decrypt (matches JS/TS SDK) + let signer = match relay_pool.signer().await { + Ok(s) => s, + Err(error) => { + tracing::error!( + target: LOG_TARGET, + error = %error, + "Failed to get signer" + ); + continue; + } + }; + match encryption::decrypt_gift_wrap_single_layer(&signer, &event).await { + Ok(decrypted_json) => { + // The decrypted content is JSON of the inner signed event. + // Use the INNER event's ID for correlation — the client + // registers the inner event ID in its correlation store. + match serde_json::from_str::(&decrypted_json) { + Ok(inner) => { + if let Err(e) = inner.verify() { + tracing::warn!( + "Inner event signature verification failed: {e}" + ); + continue; + } + { + let mut guard = match seen_gift_wrap_ids.lock() { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + }; + guard.put(event.id, ()); + } + let inbound = inner.clone(); + let inner_tags: Vec = inner.tags.to_vec(); + ( + inner.content, + inner.pubkey.to_hex(), + inner.id.to_hex(), + true, + inner_tags, + Some(inbound), + ) + } + Err(error) => { + tracing::error!( + target: LOG_TARGET, + error = %error, + "Failed to parse inner event" ); continue; } - { - let mut guard = match seen_gift_wrap_ids.lock() { - Ok(g) => g, - Err(poisoned) => poisoned.into_inner(), - }; - guard.put(event.id, ()); - } - let inner_tags: Vec = inner.tags.to_vec(); - ( - inner.content, - inner.pubkey.to_hex(), - inner.id.to_hex(), - true, - inner_tags, - ) - } - Err(error) => { - tracing::error!( - target: LOG_TARGET, - error = %error, - "Failed to parse inner event" - ); - continue; } } + Err(error) => { + tracing::error!( + target: LOG_TARGET, + error = %error, + "Failed to decrypt" + ); + continue; + } } - Err(error) => { - tracing::error!( + } else { + if encryption_mode == EncryptionMode::Required { + tracing::warn!( target: LOG_TARGET, - error = %error, - "Failed to decrypt" + sender_pubkey = %event.pubkey.to_hex(), + "Received unencrypted message but encryption is required" ); continue; } - } - } else { - if encryption_mode == EncryptionMode::Required { - tracing::warn!( - target: LOG_TARGET, - sender_pubkey = %event.pubkey.to_hex(), - "Received unencrypted message but encryption is required" - ); - continue; - } - ( - event.content.clone(), - event.pubkey.to_hex(), - event.id.to_hex(), - false, - event.tags.to_vec(), - ) - }; + let inbound = (*event).clone(); + ( + event.content.clone(), + event.pubkey.to_hex(), + event.id.to_hex(), + false, + event.tags.to_vec(), + Some(inbound), + ) + }; // Parse MCP message let mcp_msg = match validation::validate_and_parse(&content) { @@ -2009,6 +2054,7 @@ impl NostrServerTransport { client_pubkey: sender_pubkey, event_id, is_encrypted, + event: inbound_event, }); } } @@ -2162,6 +2208,7 @@ impl NostrServerTransport { client_pubkey: sender_pubkey.to_string(), event_id: event_id.to_string(), is_encrypted, + event: None, }); } // Clean up locally, let the peer's own timeout fire. diff --git a/tests/e2e_happy_path.rs b/tests/e2e_happy_path.rs index fcd5a23..88f031d 100644 --- a/tests/e2e_happy_path.rs +++ b/tests/e2e_happy_path.rs @@ -13,7 +13,7 @@ use contextvm_sdk::core::types::EncryptionMode; use contextvm_sdk::relay::mock::MockRelayPool; use contextvm_sdk::transport::client::{NostrClientTransport, NostrClientTransportConfig}; use contextvm_sdk::transport::server::{ - ClientPubkey, NostrServerTransport, NostrServerTransportConfig, + ClientPubkey, InboundEvent, NostrServerTransport, NostrServerTransportConfig, }; use contextvm_sdk::RelayPoolTrait; @@ -93,6 +93,21 @@ impl DemoServer { .unwrap_or_else(|| "none".to_string()); Ok(CallToolResult::success(vec![Content::text(pk)])) } + + /// Self-check for inbound-event injection: returns `||` + /// from the `InboundEvent` the worker places in request extensions, or + /// `none` if no event was carried. + #[tool(description = "Return the inbound Nostr event id/sig/pubkey from extensions")] + async fn whoami_event( + &self, + ctx: RequestContext, + ) -> Result { + let text = match ctx.extensions.get::() { + Some(ev) => format!("{}|{}|{}", ev.0.id.to_hex(), ev.0.sig, ev.0.pubkey.to_hex()), + None => "none".to_string(), + }; + Ok(CallToolResult::success(vec![Content::text(text)])) + } } #[tool_handler] @@ -222,7 +237,7 @@ async fn run_e2e_scenario(mode: EncryptionMode) { ); let tools = client.list_all_tools().await.expect("list_all_tools"); - assert_eq!(tools.len(), 4, "expected 4 tools"); + assert_eq!(tools.len(), 5, "expected 5 tools"); let tool_names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect(); assert!(tool_names.contains(&"echo"), "missing echo tool"); assert!(tool_names.contains(&"add"), "missing add tool"); @@ -231,6 +246,10 @@ async fn run_e2e_scenario(mode: EncryptionMode) { "missing get_echo_count tool" ); assert!(tool_names.contains(&"whoami"), "missing whoami tool"); + assert!( + tool_names.contains(&"whoami_event"), + "missing whoami_event tool" + ); // ClientPubkey injection: the worker puts the caller's pubkey in extensions. let whoami = client @@ -243,6 +262,27 @@ async fn run_e2e_scenario(mode: EncryptionMode) { "whoami should return the caller's pubkey from request extensions" ); + // InboundEvent injection: the worker threads the full client-signed event + // (gift-wrap inner event when encrypted, outer event when plaintext), so + // its sig/id are reachable — the server cannot fabricate these on its own. + let ev = client + .call_tool(call_params("whoami_event", None)) + .await + .expect("whoami_event call"); + let ev_text = first_text(&ev); + assert_ne!( + ev_text, "none", + "InboundEvent must be present for real requests" + ); + let parts: Vec<&str> = ev_text.split('|').collect(); + assert_eq!(parts.len(), 3, "whoami_event format is id|sig|pubkey"); + assert!(!parts[0].is_empty(), "event id must be non-empty"); + assert!(!parts[1].is_empty(), "event sig must be non-empty"); + assert_eq!( + parts[2], client_pubkey_hex, + "inbound event pubkey must match ClientPubkey" + ); + let echo1 = client .call_tool(call_params( "echo", From d44c6bd337f84fa8ff901885f36963a894a52316 Mon Sep 17 00:00:00 2001 From: ContextVM Date: Thu, 9 Jul 2026 14:25:32 +0200 Subject: [PATCH 3/4] fix(open-stream): abort server writers on silent client disconnect (CEP-41) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement sender-side keepalive for OpenStreamWriter to prevent resource leaks when a client silently disconnects (crash, sleep, network drop) without sending an abort frame. Previously only reader sessions ran keepalive timers, so pure producer streams (e.g. subscription tools) were never probed, leaving writers and downstream producers alive indefinitely. - Add idle_timeout / probe_timeout to OpenStreamWriterOptions to enable keepalive (disabled by default, e.g. for tests). - Arm an idle window when the writer starts streaming; the server keepalive sweep now probes writers via the new `tick` method. - Route inbound pong frames for the stream to the writer to clear probes. - On probe timeout, abort with "Probe timeout", flush a terminal response via the existing on_abort hook, and evict the dead client’s session (mirroring the TS SDK 0.13.8 fix, per CEP-41’s “release local state”). - Reuse the reader idle/probe timeout knobs (one idle/probe pair per stream). - Per CEP-41, only inbound frames reset the idle window; relay writes do not. Closes the silent-disconnect leak and aligns with CEP-41 requirements. --- CHANGELOG.md | 21 ++ sdk | 2 +- src/transport/open_stream/writer.rs | 355 +++++++++++++++++++++++++++- src/transport/server/mod.rs | 213 ++++++++++++++++- 4 files changed, 578 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e085004..f8326c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,27 @@ consumer needs the raw event; mirroring `nostr_sdk::Event` + `Tags` is non-trivial) — the omission is documented in place. +### Fixed + +- fix(open-stream): abort server writers on silent client disconnect (CEP-41). + Server→client `OpenStreamWriter`s leaked when a client silently disappeared + (crash/sleep/network drop) without sending `abort`. CEP-41 mandates each peer + maintain an idle timeout and probe the other with `ping`/`pong`, but only the + reader session ran keepalive timers; a pure producer stream (e.g. a + subscription-style tool streaming to a client) was never probed, so a dead + client left the writer — and any upstream producer keyed on `is_active()` — + alive indefinitely. The writer now arms an idle window once it starts + streaming, the server keepalive sweep probes it (mirroring the existing reader + sweep, driven by `OpenStreamWriter::tick`), an inbound `pong` for the stream is + routed to the writer to clear the probe (`ack_probe`), and a missing `pong` + aborts with `"Probe timeout"` (flushing any deferred final response via the + existing `on_abort` hook) **and evicts the dead client's session** (CEP-41 + "release local state", mirroring the TS `handleProbeTimeout`, firing the + `SessionStore` eviction callback). Per CEP-41 only inbound frames reset the idle + window — a successful `write()` against the relay is not liveness. Reuses the + reader `idle_timeout_ms` / `probe_timeout_ms` knobs (one idle/probe pair per + stream). This is the rs-sdk port of the TS SDK 0.13.8 fix. + ## [0.2.0] - 2026-06-24 ### Added diff --git a/sdk b/sdk index 572926d..2aaa38a 160000 --- a/sdk +++ b/sdk @@ -1 +1 @@ -Subproject commit 572926d78adb11ec488aaedd4a377ea2ea4070bd +Subproject commit 2aaa38afac378aa69bb0c02dc15811dcc2af9ea9 diff --git a/src/transport/open_stream/writer.rs b/src/transport/open_stream/writer.rs index 0c1b160..4c2f244 100644 --- a/src/transport/open_stream/writer.rs +++ b/src/transport/open_stream/writer.rs @@ -16,13 +16,14 @@ //! server transport wiring lands. use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; +use std::time::{Duration, Instant}; use futures::future::BoxFuture; use tokio::sync::Mutex; use super::frame::OpenStreamFrame; -use super::session::PublishFrame; +use super::session::{KeepaliveAction, PublishFrame}; /// Lifecycle hook fired after a terminal `close` frame is published. /// @@ -47,14 +48,47 @@ pub struct OpenStreamWriterOptions { pub on_close: Option, /// Fired after a terminal `abort` frame is published. pub on_abort: Option, + /// Sender-side keepalive idle window (CEP-41 §Timeout and Keepalive). `None` + /// disables keepalive (e.g. unit tests). When set, the writer arms the window + /// once it starts streaming and probes the peer with `ping` if no inbound + /// frame arrives within this duration; a missing `pong` aborts with + /// `"Probe timeout"`, closing a leak where a silently-disconnected client + /// left the writer (and its upstream producer) alive indefinitely. + pub idle_timeout: Option, + /// Time the writer waits for a matching `pong` after probing before aborting. + /// Ignored when [`idle_timeout`](OpenStreamWriterOptions::idle_timeout) is `None`. + pub probe_timeout: Duration, +} + +impl OpenStreamWriterOptions { + /// Enable sender-side keepalive with the given idle/probe windows. + pub fn with_keepalive(mut self, idle: Duration, probe: Duration) -> Self { + self.idle_timeout = Some(idle); + self.probe_timeout = probe; + self + } +} + +/// Sender-side keepalive state (CEP-41), guarded by a `std::sync::Mutex` so the +/// pure sync [`tick`](OpenStreamWriter::tick) can mutate it without awaiting. +/// +/// Per CEP-41 only **inbound** frames reset +/// [`last_activity`](Self::last_activity) — our own `chunk` publishes do not (a +/// vanished client keeps accepting relay events, so a successful `write` is not +/// liveness). +struct WriterKeepalive { + /// Last time an inbound frame was observed. Armed at `start`. + last_activity: Instant, + /// Nonce of an in-flight probe awaiting a matching `pong`. + pending_probe_nonce: Option, + /// When the in-flight probe times out → abort. + probe_deadline: Option, } /// State mutated only under the op `Mutex` (serialized publishes). struct WriterOpState { /// Next `chunkIndex` to assign (touched only by `write`). chunk_index: u64, - /// Control-frame nonce counter (touched only by `ping`). - control_nonce: u64, } struct WriterInner { @@ -68,11 +102,18 @@ struct WriterInner { /// Monotonic outer `progress`, shared so `abort` can mint one without the op /// lock. Frames use `fetch_add(1) + 1` → 1, 2, 3, … progress: AtomicU64, + /// Control-frame nonce counter shared by manual `ping` and keepalive probes + /// so nonces stay unique within the stream (`{token}:{n}`). + control_nonce: AtomicU64, /// Liveness flag, **outside** the op lock so `abort` never blocks on a stuck /// write. `false` once closed or aborted. active: AtomicBool, /// Whether the lazy `start` frame has been published. started: AtomicBool, + /// Sender-side keepalive config + state. Inert when `idle_timeout` is `None`. + idle_timeout: Option, + probe_timeout: Duration, + keepalive: StdMutex, } /// Minimal CEP-41 producer/writer. @@ -91,13 +132,18 @@ impl OpenStreamWriter { publish_frame: options.publish_frame, on_close: options.on_close, on_abort: options.on_abort, - op: Mutex::new(WriterOpState { - chunk_index: 0, - control_nonce: 0, - }), + op: Mutex::new(WriterOpState { chunk_index: 0 }), progress: AtomicU64::new(0), + control_nonce: AtomicU64::new(0), active: AtomicBool::new(true), started: AtomicBool::new(false), + idle_timeout: options.idle_timeout, + probe_timeout: options.probe_timeout, + keepalive: StdMutex::new(WriterKeepalive { + last_activity: Instant::now(), + pending_probe_nonce: None, + probe_deadline: None, + }), }), } } @@ -144,6 +190,12 @@ impl OpenStreamWriter { )?; (self.inner.publish_frame)(notification).await?; self.inner.started.store(true, Ordering::SeqCst); + // Arm the sender-side keepalive idle window (CEP-41): the stream is active + // from `start`, so the probe clock starts here. Only inbound frames reset + // it afterwards — our own chunk publishes are not liveness. + if self.inner.idle_timeout.is_some() { + self.inner.keepalive.lock().unwrap().last_activity = Instant::now(); + } Ok(()) } @@ -171,12 +223,12 @@ impl OpenStreamWriter { /// Publish a keepalive `ping` carrying a fresh `{token}:{n}` nonce. pub async fn ping(&self) -> crate::Result<()> { - let mut op = self.inner.op.lock().await; + let _op = self.inner.op.lock().await; if !self.inner.active.load(Ordering::SeqCst) { return Ok(()); } - op.control_nonce += 1; - let nonce = format!("{}:{}", self.inner.progress_token, op.control_nonce); + let n = self.inner.control_nonce.fetch_add(1, Ordering::SeqCst) + 1; + let nonce = format!("{}:{}", self.inner.progress_token, n); let progress = self.next_progress(); let notification = OpenStreamFrame::Ping { nonce }.into_progress_notification( &self.inner.progress_token, @@ -188,11 +240,18 @@ impl OpenStreamWriter { } /// Publish a `pong` echoing the peer's `ping` nonce. + /// + /// An inbound `ping` is proof of peer liveness (CEP-41: receipt of any valid + /// frame resets the idle window), so this refreshes the keepalive clock + /// before responding. pub async fn pong(&self, nonce: String) -> crate::Result<()> { let _op = self.inner.op.lock().await; if !self.inner.active.load(Ordering::SeqCst) { return Ok(()); } + if self.inner.idle_timeout.is_some() { + self.inner.keepalive.lock().unwrap().last_activity = Instant::now(); + } let progress = self.next_progress(); let notification = OpenStreamFrame::Pong { nonce }.into_progress_notification( &self.inner.progress_token, @@ -203,6 +262,99 @@ impl OpenStreamWriter { Ok(()) } + // ── sender-side keepalive (CEP-41), driven by the transport sweep ─────── + + /// Pure keepalive transition (idle → `ping`, probe deadline → abort), mirroring + /// the reader-side + /// [`OpenStreamSession::tick`](super::session::OpenStreamSession::tick). + /// Driven by the owning transport's periodic sweep with `Instant::now()`. + /// + /// Returns [`KeepaliveAction::None`] when keepalive is disabled, the writer is + /// inactive, or it has not yet started (the idle window arms at `start`). + pub fn tick(&self, now: Instant) -> KeepaliveAction { + let Some(idle_timeout) = self.inner.idle_timeout else { + return KeepaliveAction::None; + }; + if !self.inner.active.load(Ordering::SeqCst) || !self.inner.started.load(Ordering::SeqCst) { + return KeepaliveAction::None; + } + let mut ka = self.inner.keepalive.lock().unwrap(); + + // 1. A pending probe whose deadline passed → abort. + if let Some(deadline) = ka.probe_deadline { + if ka.pending_probe_nonce.is_some() && now >= deadline { + return KeepaliveAction::Abort("Probe timeout".to_string()); + } + } + + // 2. Idle past the threshold (and not already probing) → ping. + if ka.pending_probe_nonce.is_none() + && now.saturating_duration_since(ka.last_activity) >= idle_timeout + { + let n = self.inner.control_nonce.fetch_add(1, Ordering::SeqCst) + 1; + let nonce = format!("{}:{}", self.inner.progress_token, n); + ka.pending_probe_nonce = Some(nonce.clone()); + ka.probe_deadline = Some(now + self.inner.probe_timeout); + return KeepaliveAction::SendPing(nonce); + } + + KeepaliveAction::None + } + + /// Publish a keepalive `ping` carrying a `nonce` already minted by [`tick`](Self::tick). + /// + /// Used by the transport sweep; unlike [`ping`](Self::ping) it does not mint a + /// new nonce (the probe recorded by `tick` must match the frame on the wire so + /// a matching `pong` clears it). Does not take the op lock, so a stuck app + /// `write` cannot block keepalive (mirroring the TS writer's queue bypass). + /// + /// Ceiling: by skipping the op lock, this probe's `progress` (shared atomic) + /// can interleave with an in-flight `write` — a probe may carry a lower + /// `progress` than a chunk published just before it. Harmless: relays reorder + /// regardless and control frames aren't `chunkIndex`-gap-checked. Re-serialize + /// under the op lock only if strict monotonic publish-order is ever required + /// (which would re-block keepalive behind a stuck `write`). + pub async fn send_probe(&self, nonce: String) -> crate::Result<()> { + if !self.inner.active.load(Ordering::SeqCst) { + return Ok(()); + } + let progress = self.next_progress(); + let notification = OpenStreamFrame::Ping { nonce }.into_progress_notification( + &self.inner.progress_token, + progress, + None, + )?; + (self.inner.publish_frame)(notification).await.map(|_| ()) + } + + /// Acknowledge an inbound `pong` matching the pending keepalive probe (CEP-41). + /// + /// A matching `pong` clears the probe and refreshes the idle window. A `pong` + /// whose nonce does not match the pending probe is **not** evidence of + /// liveness (CEP-41: "A `pong` with an unknown … nonce MUST NOT be treated as + /// evidence of liveness") and is ignored. + pub fn ack_probe(&self, nonce: &str) { + let mut ka = self.inner.keepalive.lock().unwrap(); + if self.inner.active.load(Ordering::SeqCst) + && ka.pending_probe_nonce.as_deref() == Some(nonce) + { + ka.pending_probe_nonce = None; + ka.probe_deadline = None; + ka.last_activity = Instant::now(); + } + } + + /// Release writer resources without publishing a terminal frame — used on + /// transport teardown so the writer no longer drives keepalive. Idempotent. + /// + /// Only flips `active` to `false`: [`tick`](Self::tick) and + /// [`ack_probe`](Self::ack_probe) early-return on that flag, so an armed probe + /// becomes inert without needing to be cleared (matching `close`/`abort`, + /// which also leave keepalive state untouched once terminal). + pub fn dispose(&self) { + let _ = self.inner.active.swap(false, Ordering::SeqCst); + } + /// Close the stream gracefully. Declares `lastChunkIndex` iff any chunks were /// written. Runs [`on_close`](OpenStreamWriterOptions::on_close) **after** the /// frame is published (even on publish failure); propagates the publish error. @@ -309,6 +461,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }) } @@ -321,6 +475,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); assert!(writer.is_active()); @@ -356,6 +512,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); writer.start().await.unwrap(); @@ -419,6 +577,8 @@ mod tests { content_type: None, on_close: Some(on_close), on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); writer.close().await.unwrap(); assert_eq!(frame_type(log.lock().unwrap().last().unwrap()), "close"); @@ -439,6 +599,8 @@ mod tests { content_type: None, on_close: None, on_abort: Some(on_abort), + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); abort_writer.abort(Some("done".to_string())).await.unwrap(); let frames = log.lock().unwrap(); @@ -476,6 +638,8 @@ mod tests { content_type: None, on_close: None, on_abort: Some(on_abort), + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); writer.abort(Some("ordered".to_string())).await.unwrap(); @@ -509,6 +673,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); assert!(writer.start().await.is_err()); @@ -540,6 +706,8 @@ mod tests { content_type: None, on_close: Some(on_close), on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); assert!(writer.close().await.is_err()); @@ -573,6 +741,8 @@ mod tests { content_type: None, on_close: None, on_abort: Some(on_abort), + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); assert!(writer.abort(Some("cleanup".to_string())).await.is_err()); @@ -616,6 +786,8 @@ mod tests { content_type: None, on_close: None, on_abort: Some(on_abort), + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); let writer2 = writer.clone(); @@ -659,6 +831,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }); let (a, b, c) = tokio::join!( @@ -682,4 +856,163 @@ mod tests { assert_eq!(progress_of(&frames[3]), 4); assert_eq!(cvm(&frames[3])["lastChunkIndex"], 1); } + + // ── sender-side keepalive (CEP-41) ────────────────────────────────────── + + /// A writer with sender-side keepalive armed (`idle_timeout = Some(..)`). + /// Returns the frame log so the probe publish can be inspected. + fn keepalive_writer(idle_ms: u64, probe_ms: u64) -> (OpenStreamWriter, FrameLog) { + let log = frame_log(); + let writer = OpenStreamWriter::new(OpenStreamWriterOptions { + progress_token: "tok-ka".to_string(), + publish_frame: recording_publisher(log.clone()), + content_type: None, + on_close: None, + on_abort: None, + idle_timeout: Some(Duration::from_millis(idle_ms)), + probe_timeout: Duration::from_millis(probe_ms), + }); + (writer, log) + } + + #[tokio::test] + async fn keepalive_disabled_writer_tick_is_always_none() { + // `writer_with` builds a keepalive-disabled writer (idle_timeout: None). + let log = frame_log(); + let writer = writer_with(log); + writer.start().await.unwrap(); + + let t = Instant::now(); + assert_eq!(writer.tick(t), KeepaliveAction::None); + assert_eq!( + writer.tick(t + Duration::from_secs(60)), + KeepaliveAction::None + ); + } + + #[tokio::test] + async fn tick_returns_none_before_start() { + let (writer, _) = keepalive_writer(10, 20); + // Keepalive arms at `start`; a never-started writer must not probe. + assert_eq!(writer.tick(Instant::now()), KeepaliveAction::None); + } + + #[tokio::test] + async fn tick_idle_probes_then_probe_deadline_aborts() { + let (writer, _) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + let t0 = Instant::now(); + + // Before the idle threshold: nothing. + assert_eq!(writer.tick(t0), KeepaliveAction::None); + // At the idle threshold: probe with a `{token}:{n}` nonce. + let nonce = match writer.tick(t0 + Duration::from_millis(11)) { + KeepaliveAction::SendPing(n) => n, + other => panic!("expected SendPing, got {other:?}"), + }; + assert!(nonce.starts_with("tok-ka:")); + // Probe in flight, deadline not yet reached: nothing. + assert_eq!( + writer.tick(t0 + Duration::from_millis(15)), + KeepaliveAction::None + ); + // Probe deadline reached with no pong: abort. + assert_eq!( + writer.tick(t0 + Duration::from_millis(31)), + KeepaliveAction::Abort("Probe timeout".to_string()) + ); + } + + #[tokio::test] + async fn matching_pong_clears_probe_and_rearms_idle() { + let (writer, _) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + let t0 = Instant::now(); + let nonce = match writer.tick(t0 + Duration::from_millis(11)) { + KeepaliveAction::SendPing(n) => n, + other => panic!("expected SendPing, got {other:?}"), + }; + + // A matching pong clears the probe and refreshes liveness. + writer.ack_probe(&nonce); + + // At the *old* probe deadline the stream is NOT aborted: the probe was + // cleared and the idle window restarted from ack time. + let after_ack = Instant::now(); + assert!(!matches!( + writer.tick(after_ack + Duration::from_millis(5)), + KeepaliveAction::Abort(_) + )); + // A fresh probe fires only once the new idle window elapses. + assert!(matches!( + writer.tick(after_ack + Duration::from_millis(12)), + KeepaliveAction::SendPing(_) + )); + } + + #[tokio::test] + async fn unmatched_pong_does_not_clear_probe() { + let (writer, _) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + let t0 = Instant::now(); + let _ = writer.tick(t0 + Duration::from_millis(11)); + + // A pong with a non-matching nonce must not clear the pending probe + // (CEP-41: an unknown nonce is not evidence of liveness). + writer.ack_probe("tok-ka:999"); + + assert_eq!( + writer.tick(t0 + Duration::from_millis(31)), + KeepaliveAction::Abort("Probe timeout".to_string()) + ); + } + + #[tokio::test] + async fn inbound_ping_refreshes_idle_window() { + let (writer, _) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + + // Handling an inbound `ping` (pong) is liveness — refreshes the idle + // window so a tick shortly after does not probe. + writer.pong("peer-ping".to_string()).await.unwrap(); + let after_pong = Instant::now(); + assert_eq!( + writer.tick(after_pong + Duration::from_millis(5)), + KeepaliveAction::None + ); + // Once the idle window elapses again, it probes. + assert!(matches!( + writer.tick(after_pong + Duration::from_millis(12)), + KeepaliveAction::SendPing(_) + )); + } + + #[tokio::test] + async fn send_probe_publishes_a_ping_with_the_given_nonce() { + let (writer, log) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + writer.send_probe("custom-nonce".to_string()).await.unwrap(); + + // `frame_types` takes the `log` lock itself; call it before acquiring + // `frames` so the two lock acquisitions don't overlap (std::sync::Mutex is + // not re-entrant). + assert_eq!(frame_types(&log), vec!["start", "ping"]); + let frames = log.lock().unwrap(); + assert_eq!(cvm(&frames[1])["nonce"], "custom-nonce"); + } + + #[tokio::test] + async fn dispose_makes_writer_inactive_and_tick_inert() { + let (writer, _) = keepalive_writer(10, 20); + writer.start().await.unwrap(); + assert!(writer.is_active()); + + writer.dispose(); + + assert!(!writer.is_active()); + assert_eq!( + writer.tick(Instant::now() + Duration::from_secs(60)), + KeepaliveAction::None + ); + } } diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs index 80e5bde..3ece626 100644 --- a/src/transport/server/mod.rs +++ b/src/transport/server/mod.rs @@ -798,7 +798,9 @@ impl NostrServerTransport { } receivers.clear(); } - self.open_stream.lock_slots().clear(); + for (_, slot) in self.open_stream.lock_slots().drain() { + slot.writer.dispose(); + } self.open_stream.lock_token_index().clear(); Ok(()) } @@ -1612,6 +1614,7 @@ impl NostrServerTransport { &relay_pool, encryption_mode, gift_wrap_mode, + &sessions, ) .await; continue; @@ -2391,6 +2394,13 @@ impl NostrServerTransport { content_type: None, on_close: Some(on_close), on_abort: Some(on_abort), + // CEP-41 sender-side keepalive: reuse the reader idle/probe knobs (one + // idle/probe pair per stream, per spec). `None` when idle is disabled, + // keeping the writer keepalive in lockstep with the sweep gate + // (`open_stream_sweep_enabled` is false when `idle_timeout_ms == 0`). + idle_timeout: (state.policy.idle_timeout_ms != 0) + .then(|| Duration::from_millis(state.policy.idle_timeout_ms)), + probe_timeout: Duration::from_millis(state.policy.probe_timeout_ms), }); let snapshot = RouteSnapshot { client_pubkey, @@ -2479,6 +2489,29 @@ impl NostrServerTransport { .await; } } + Some(OpenStreamFrame::Pong { nonce }) => { + // CEP-41: a `pong` for a stream the server is *writing* (server→client) + // acknowledges our keepalive probe (completing the round trip the reader + // sweep opened). Only intercept when a writer owns the token; + // client→server streams fall through to the reader engine below. + if let Some(writer) = writer { + writer.ack_probe(&nonce); + } else { + Self::feed_open_stream_reader( + state, + relay_pool, + encryption_mode, + gift_wrap_mode, + notification, + sender_pubkey, + event_id, + is_encrypted, + is_gift_wrap, + outer_kind, + ) + .await; + } + } Some(OpenStreamFrame::Start { .. }) => { Self::feed_open_stream_reader( state, @@ -2674,6 +2707,7 @@ impl NostrServerTransport { relay_pool: &Arc, encryption_mode: EncryptionMode, gift_wrap_mode: GiftWrapMode, + sessions: &SessionStore, ) { let now = Instant::now(); let mut actions: Vec<(String, String, KeepaliveAction)> = Vec::new(); @@ -2727,6 +2761,68 @@ impl NostrServerTransport { KeepaliveAction::None => {} } } + + // CEP-41: writer (server→client) keepalive — each producer stream MUST + // maintain an idle timeout (CEP-41 §Timeout and Keepalive). A client that + // vanishes without `abort` is probed here and aborted on a missing `pong`; + // the writer's `on_abort` hook then flushes any deferred final response. + // This is the mirror of the reader sweep above for sender-side streams. + // The client pubkey is captured up front: the slot may be removed by the + // time we act on an `Abort` (the writer's on_abort flush drops it), and we + // need it to release the dead client's session. + let writer_actions: Vec<(String, PublicKey, KeepaliveAction)> = state + .lock_slots() + .iter() + .map(|(event_id, slot)| { + ( + event_id.clone(), + slot.snapshot.client_pubkey, + slot.writer.tick(now), + ) + }) + .filter(|(.., action)| !matches!(action, KeepaliveAction::None)) + .collect(); + for (event_id, client_pubkey, action) in writer_actions { + let Some(writer) = state.writer_for(&event_id) else { + continue; + }; + match action { + KeepaliveAction::SendPing(nonce) => { + if let Err(error) = writer.send_probe(nonce).await { + tracing::warn!( + target: LOG_TARGET, + error = %error, + event_id = %event_id, + "Failed to publish open-stream keepalive ping" + ); + } + } + KeepaliveAction::Abort(reason) => { + if let Err(error) = writer.abort(Some(reason)).await { + tracing::warn!( + target: LOG_TARGET, + error = %error, + event_id = %event_id, + "Failed to abort open-stream writer on probe timeout" + ); + } + // CEP-41: a probe timeout means the client is gone — release + // its session too ("release local state"), mirroring the TS + // `handleProbeTimeout`. Done after the abort so the abort frame + // is published first; eviction is independent of the + // snapshot-backed deferred-response flush. Only the sweep + // produces a writer `Abort`, so this path is probe-timeout only + // (a tool-initiated `close`/`abort` does not evict the session). + let pubkey_hex = client_pubkey.to_hex(); + if sessions.remove_session(&pubkey_hex).await { + if let Some(cb) = sessions.eviction_callback() { + cb(pubkey_hex); + } + } + } + KeepaliveAction::None => {} + } + } } async fn cleanup_sessions( @@ -3551,6 +3647,8 @@ mod tests { content_type: None, on_close: None, on_abort: None, + idle_timeout: None, + probe_timeout: Duration::from_millis(20_000), }) } @@ -3694,4 +3792,117 @@ mod tests { "a disabled server must not expose writers (deferral never attempted)" ); } + + // ── CEP-41 writer keepalive sweep (server→client, silent-client abort) ── + + /// A keepalive-armed writer whose publish closure is a no-op. The sweep + /// drives `tick`; a missing `pong` aborts it on the probe deadline. + fn keepalive_test_writer(token: &str, idle_ms: u64, probe_ms: u64) -> OpenStreamWriter { + let publish_frame: PublishFrame = Arc::new(|_frame: JsonRpcNotification| { + Box::pin(async move { Ok(EventId::all_zeros()) }) + }); + OpenStreamWriter::new(OpenStreamWriterOptions { + progress_token: token.to_string(), + publish_frame, + content_type: None, + on_close: None, + on_abort: None, + idle_timeout: Some(Duration::from_millis(idle_ms)), + probe_timeout: Duration::from_millis(probe_ms), + }) + } + + /// Regression: a client that silently disappears (sends no `pong`, no + /// `abort`) is detected by the server-writer keepalive sweep — the writer is + /// probed after the idle window and aborted once the probe times out, and the + /// dead client's session is evicted (CEP-41 "release local state", mirroring + /// the TS `handleProbeTimeout`). This is the rs-sdk port of the TS 0.13.8 fix; + /// without it the writer (and any upstream producer keyed on `is_active`) + /// leaks indefinitely. `tokio::time::sleep` guarantees *at least* the requested + /// duration, so the idle/probe margins are deterministic, not flaky. + #[tokio::test] + async fn sweep_aborts_writer_and_evicts_session_when_client_goes_silent() { + let config = NostrServerTransportConfig::default() + .with_open_stream(OpenStreamConfig::default().with_enabled(true)); + let pool: Arc = Arc::new(MockRelayPool::new()); + let mut transport = NostrServerTransport::with_relay_pool(config, pool) + .await + .expect("server transport"); + + let writer = keepalive_test_writer("tok-silent", 40, 60); + writer.start().await.expect("start the stream"); + install_slot(&transport.open_stream, "evt-silent", writer.clone(), false); + assert!(writer.is_active(), "writer starts active"); + + // Install a session for the slot's client and arm the eviction callback. + let pubkey_hex = transport + .open_stream + .lock_slots() + .get("evt-silent") + .expect("slot") + .snapshot + .client_pubkey + .to_hex(); + let evicted = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let evicted_cb = evicted.clone(); + transport.sessions.set_eviction_callback(Arc::new(move |_| { + evicted_cb.store(true, std::sync::atomic::Ordering::SeqCst); + })); + transport + .sessions + .get_or_create_session(&pubkey_hex, false, &transport.event_routes) + .await; + assert!( + transport.sessions.get_session(&pubkey_hex).await.is_some(), + "session present before the sweep" + ); + + let relay_pool = Arc::clone(&transport.base.relay_pool); + let encryption_mode = transport.config.encryption_mode; + let gift_wrap_mode = transport.config.gift_wrap_mode; + + // Past the idle window: the sweep probes (idle → SendPing). The writer is + // NOT aborted yet — it is waiting for a `pong` the silent client never sends. + tokio::time::sleep(Duration::from_millis(60)).await; + NostrServerTransport::sweep_open_stream_sessions( + &transport.open_stream, + &relay_pool, + encryption_mode, + gift_wrap_mode, + &transport.sessions, + ) + .await; + assert!( + writer.is_active(), + "writer must remain active while a probe is in flight" + ); + assert!( + transport.sessions.get_session(&pubkey_hex).await.is_some(), + "session must survive a mere probe (client not yet declared dead)" + ); + + // Past the probe timeout: the sweep aborts (probe deadline → Abort) and + // evicts the dead client's session. + tokio::time::sleep(Duration::from_millis(80)).await; + NostrServerTransport::sweep_open_stream_sessions( + &transport.open_stream, + &relay_pool, + encryption_mode, + gift_wrap_mode, + &transport.sessions, + ) + .await; + assert!( + !writer.is_active(), + "writer must abort after a silent client misses the probe deadline" + ); + assert!( + transport.sessions.get_session(&pubkey_hex).await.is_none(), + "dead client's session must be evicted on probe timeout" + ); + assert!( + evicted.load(std::sync::atomic::Ordering::SeqCst), + "eviction callback must fire on probe-timeout session release" + ); + } } From 609239c473ee6e5a25dbdbf204fe0bc23ff53b64 Mon Sep 17 00:00:00 2001 From: ContextVM Date: Fri, 10 Jul 2026 12:50:58 +0200 Subject: [PATCH 4/4] fix(server): skip ClientPubkey injection for synthetic announcement/init drives The transport's own announcement and initialization drives use the `ANNOUNCEMENT_REQUEST_ID` sentinel as their pubkey. Previously, this sentinel was unconditionally injected into request extensions, handing handlers a bogus caller identity. Now guard the injection so that only real client pubkeys (including those from oversized CEP-22 requests) are exposed via `ClientPubkey`. --- CHANGELOG.md | 19 ++++++-- src/rmcp_transport/worker.rs | 24 ++++++---- src/transport/server/mod.rs | 37 ++++++++++++---- tests/transport_integration.rs | 81 ++++++++++++++++++++++++++++++++++ 4 files changed, 140 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f8326c2..5e55af4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,13 @@ ### Added - `ClientPubkey`: the rmcp server worker now injects the caller's Nostr public - key (hex) into every inbound request's `extensions` typemap, so tool/resource/ - prompt handlers can identify their caller via `ctx.extensions.get::()`. + key (hex) into every **real** inbound request's `extensions` typemap, so + tool/resource/prompt handlers can identify their caller via + `ctx.extensions.get::()`. It is not injected for the transport's + own synthetic announcement/initialization drives (which carry the + `ANNOUNCEMENT_REQUEST_ID` sentinel as their pubkey), so handlers never observe + a bogus caller; oversized (CEP-22) requests carry a real pubkey and are + injected as usual. This closes the parity gap with the TS adapter's `extra._meta.clientPubkey`, but uses rmcp's typed extensions (local-only, never on the wire) instead of the `_meta` field, so it is always on rather than opt-in. The inbound event id is @@ -15,7 +20,8 @@ client-signed Nostr request event into `extensions`, reachable via `ctx.extensions.get::()`. For gift-wrapped requests this is the inner, signature-verified event (its `pubkey` matches `ClientPubkey` by - construction); for plaintext requests it is the outer event. This exposes + construction); for plaintext requests it is the outer event; for CEP-22 + oversized requests it is the carrying `end` frame's event. This exposes `id`, `pubkey`, `sig`, `tags`, … — notably `sig`, which the server cannot reconstruct without the client's private key. Handlers that must bind a tool call to / store / audit the publishing event (e.g. an MLS key-package @@ -48,6 +54,13 @@ window — a successful `write()` against the relay is not liveness. Reuses the reader `idle_timeout_ms` / `probe_timeout_ms` knobs (one idle/probe pair per stream). This is the rs-sdk port of the TS SDK 0.13.8 fix. +- fix(server): verify plaintext event signatures before trusting `event.pubkey` + for handler identity, the auth allowlist, and request correlation, mirroring + the gift-wrap arm. The default `RelayPool` verifies inbound signatures itself, + but `RelayPoolTrait` is public, so a custom pool that skips verification + (e.g. `MockRelayPool`) left caller identity dependent on an undocumented pool + assumption — and a forged pubkey could bypass the auth allowlist. This is the + rs-sdk half of the TS identity-forgery fix (ContextVM/sdk#64, #69). ## [0.2.0] - 2026-06-24 diff --git a/src/rmcp_transport/worker.rs b/src/rmcp_transport/worker.rs index 549e27b..bd939e0 100644 --- a/src/rmcp_transport/worker.rs +++ b/src/rmcp_transport/worker.rs @@ -215,15 +215,21 @@ impl Worker for NostrServerWorker { // loop swaps these extensions straight into the handler's // `RequestContext` before dispatch. if let rmcp::model::JsonRpcMessage::Request(ref mut jr) = rmcp_msg { - // Caller pubkey is relevant to every request (auth, - // payments, logging) and extensions are local-only, so - // inject it unconditionally. This mirrors the TS adapter's - // `extra._meta.clientPubkey`, but via rmcp's typed - // extensions rather than the on-wire `_meta` field (which - // is why the TS knob is opt-in but this is not). - jr.request - .extensions_mut() - .insert(ClientPubkey(client_pubkey.clone())); + // Caller pubkey is relevant to every real request (auth, + // payments, logging) and extensions are local-only. Skip + // the transport's own announcement/init drives, whose + // "pubkey" is the `ANNOUNCEMENT_REQUEST_ID` sentinel — + // injecting it would hand handlers a bogus caller. This + // mirrors the TS adapter's `extra._meta.clientPubkey`, + // but via rmcp's typed extensions rather than the on-wire + // `_meta` field (which is why the TS knob is opt-in but + // this is not). Gate on the sentinel, not on `event`: + // oversized requests carry a real pubkey with no event. + if client_pubkey != ANNOUNCEMENT_REQUEST_ID { + jr.request + .extensions_mut() + .insert(ClientPubkey(client_pubkey.clone())); + } // Full inbound event (id + sig + pubkey + …) for // handlers that must bind to / store / audit the // publishing event. Only real client requests carry diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs index 3ece626..aa1c4cc 100644 --- a/src/transport/server/mod.rs +++ b/src/transport/server/mod.rs @@ -439,11 +439,12 @@ pub struct IncomingRequest { pub is_encrypted: bool, /// The inbound (client-signed) Nostr event, if this request carried one. /// - /// `Some` for real client requests — the inner, signature-verified gift-wrap - /// event for encrypted requests, or the outer event for plaintext. `None` - /// for requests the transport synthesizes itself (announcement / - /// initialization drives, CEP-22 oversized reassembly), which carry no real - /// Nostr event. Handlers reach it via [`InboundEvent`] in request + /// `Some` for real client requests: the inner, signature-verified gift-wrap + /// event for encrypted requests, the outer event for plaintext, or the + /// carrying frame event for CEP-22 oversized reassembly. `None` only for + /// requests the transport synthesizes itself (announcement / + /// initialization drives), which carry no real Nostr event. Handlers reach + /// it via [`InboundEvent`] in request /// `extensions`; see that type's docs. pub event: Option, } @@ -493,9 +494,13 @@ pub struct ClientPubkey(pub String); /// For gift-wrapped requests this is the **inner**, signature-verified event /// (the same one whose `pubkey` is surfaced as [`ClientPubkey`], so `ev.0.pubkey` /// and `ClientPubkey` agree by construction). For plaintext requests it is the -/// outer request event. It is injected only for real client requests; synthetic -/// transport-internal requests carry no event, so `extensions.get::()` -/// returns `None` for them. This is purely a local affordance — it is never +/// outer request event. For CEP-22 oversized requests it is the carrying `end` +/// frame's event (a `notifications/progress` event signed by the client whose +/// `id` is the correlation id) — the request was chunked, so no single dedicated +/// request event exists. It is injected only for real client requests; synthetic +/// transport-internal requests (announcement / initialization drives) carry no +/// event, so `extensions.get::()` returns `None` for them. This is +/// purely a local affordance — it is never /// serialized onto the wire. The event id is also reachable as the rmcp request /// id (`ctx.id`); this type additionally exposes `sig`, which the server cannot /// reconstruct without the client's private key. @@ -1747,6 +1752,18 @@ impl NostrServerTransport { ); continue; } + // Verify the signature (and that `id` matches content) before + // trusting `pubkey`/`id` for handler identity + correlation. + // Redundant against the default RelayPool (it verifies inbound + // signatures), but keeps caller identity independent of the + // pool impl — a custom RelayPoolTrait may skip verification. + if let Err(e) = event.verify() { + tracing::warn!( + target: LOG_TARGET, + "Plaintext event signature verification failed: {e}" + ); + continue; + } let inbound = (*event).clone(); ( event.content.clone(), @@ -1928,6 +1945,7 @@ impl NostrServerTransport { &request_wrap_kinds, &tx, &open_stream, + inbound_event, ) .await; continue; @@ -2088,6 +2106,7 @@ impl NostrServerTransport { request_wrap_kinds: &Arc>>>, tx: &tokio::sync::mpsc::UnboundedSender, open_stream: &ServerOpenStreamState, + inbound_event: Option, ) { // The outer progressToken keys the transfer (needed for accept + route). // String or number — defensive only: every known sender stringifies @@ -2211,7 +2230,7 @@ impl NostrServerTransport { client_pubkey: sender_pubkey.to_string(), event_id: event_id.to_string(), is_encrypted, - event: None, + event: inbound_event, }); } // Clean up locally, let the peer's own timeout fire. diff --git a/tests/transport_integration.rs b/tests/transport_integration.rs index 4868c0e..25c52c7 100644 --- a/tests/transport_integration.rs +++ b/tests/transport_integration.rs @@ -389,6 +389,74 @@ async fn auth_allowlist_blocks_disallowed_pubkey() { ); } +// ── 4b. Plaintext signature verify blocks a forged (allowlisted) pubkey ───── +// +// A custom RelayPoolTrait that does NOT signature-verify inbound events (our +// own MockRelayPool, or any third-party pool) lets a client put any pubkey in a +// plaintext event. Without an in-app `verify()` that forgery bypasses the auth +// allowlist — the gap TS fixed in ContextVM/sdk#64/#69. The plaintext arm must +// verify the signature (and that id matches content) before trusting +// `event.pubkey` for identity or correlation. + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn plaintext_forged_pubkey_is_rejected_by_signature_verify() { + let (_client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let server_pool = Arc::new(server_pool); + + // The "victim" pubkey the attacker forges — it IS on the allowlist. + let victim_keys = Keys::generate(); + // The attacker's real key, which actually signs the event. + let attacker_keys = Keys::generate(); + + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default() + .with_encryption_mode(EncryptionMode::Disabled) + .with_allowed_public_keys(vec![victim_keys.public_key().to_hex()]), + Arc::clone(&server_pool) as Arc, + ) + .await + .expect("create server transport"); + + let mut server_rx = server + .take_message_receiver() + .expect("server message receiver"); + server.start().await.expect("server start"); + let_event_loops_start().await; + + // Build a valid plaintext tools/list request addressed to the server, + // signed by the attacker, then forge the author pubkey to the victim's. + let request = JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!("forge-1"), + method: "tools/list".to_string(), + params: None, + }); + let mut event = contextvm_sdk::core::serializers::mcp_to_nostr_event( + &request, + CTXVM_MESSAGES_KIND, + BaseTransport::create_recipient_tags(&server_pubkey), + ) + .expect("serialize request") + .sign_with_keys(&attacker_keys) + .expect("sign event"); + // The forgery: claim to be the allowlisted victim. The signature was made by + // the attacker, so an in-app `verify()` must reject this before the allowlist + // ever sees the (allowlisted) victim pubkey. + event.pubkey = victim_keys.public_key(); + server_pool + .publish_event(&event) + .await + .expect("deliver forged event"); + + // The forged request must NOT reach the handler. + let result = tokio::time::timeout(Duration::from_millis(500), server_rx.recv()).await; + assert!( + result.is_err(), + "forged-pubkey plaintext event must be rejected by signature verification" + ); +} + // ── 5. Encryption mode Required drops plaintext ───────────────────────────── #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -3660,6 +3728,7 @@ async fn server_close_stops_event_loop() { async fn oversized_request_roundtrip_client_to_server() { let (client_pool, server_pool) = MockRelayPool::create_pair(); let server_pubkey = server_pool.mock_public_key(); + let client_pubkey_hex = client_pool.mock_public_key().to_hex(); let server_pool = Arc::new(server_pool); let oversized = OversizedTransferConfig::enabled() @@ -3723,6 +3792,18 @@ async fn oversized_request_roundtrip_client_to_server() { assert_eq!(incoming.message.method(), Some("tools/call")); assert_eq!(incoming.message.id(), Some(&serde_json::json!("big-1"))); + // CEP-22 reassembly surfaces the carrying event (parity with TS), so a + // handler can bind to / audit the chunked request rather than seeing `None`. + assert!( + incoming.event.is_some(), + "oversized request must carry its event" + ); + assert_eq!( + incoming.event.as_ref().unwrap().pubkey.to_hex(), + client_pubkey_hex, + "carrying event pubkey must be the client's" + ); + // No second message should surface. let second = tokio::time::timeout(Duration::from_millis(100), server_rx.recv()).await; assert!(second.is_err(), "only one reassembled request expected");