From 0a05ea3e86a5cde4ffd56ee08a4f6a95ec40bc59 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Fri, 28 Aug 2026 16:07:53 -0700 Subject: [PATCH 1/2] Verify inbound ACP relay events Signed-off-by: Jordan Mecom --- crates/buzz-acp/src/relay.rs | 261 +++++++++++++++++++++++++++++++++++ 1 file changed, 261 insertions(+) diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 6188e57a11d..81fc84c50fb 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -2155,6 +2155,36 @@ async fn handle_ws_message( subscription_id, event, } => { + // Relay and storage responses are untrusted. Verify before + // any event field can affect routing, replay state, or the + // harness queues. + let event_id = event.id.to_hex(); + let event = match tokio::task::spawn_blocking(move || { + buzz_core::verify_event(&event).map(|()| event) + }) + .await + { + Ok(Ok(event)) => event, + Ok(Err(error)) => { + warn!( + subscription_id, + event_id, + error = %error, + "relay event failed NIP-01 verification — dropping" + ); + return true; + } + Err(error) => { + warn!( + subscription_id, + event_id, + error = %error, + "relay event verification task failed — dropping" + ); + return true; + } + }; + if subscription_id == OBSERVER_CONTROL_SUB_ID { match observer_control_tx.try_send(*event) { Ok(()) => {} @@ -4495,6 +4525,237 @@ mod tests { .expect("parse test websocket frame") } + fn make_signed_channel_event(keys: &Keys, content: &str, created_at_secs: u64) -> Event { + EventBuilder::new(Kind::Custom(9), content) + .tags([]) + .custom_created_at(nostr::Timestamp::from(created_at_secs)) + .sign_with_keys(keys) + .expect("sign channel event") + } + + fn replace_event_field(event: &Event, field: &str, replacement: Value) -> Event { + let mut value = serde_json::to_value(event).expect("serialize event"); + value[field] = replacement; + serde_json::from_value(value).expect("deserialize tampered event") + } + + fn recompute_event_id(event: &Event) -> Event { + let id = nostr::EventId::new( + &event.pubkey, + &event.created_at, + &event.kind, + &event.tags, + &event.content, + ); + replace_event_field(event, "id", json!(id.to_hex())) + } + + async fn handle_test_relay_event( + ws: &mut WsStream, + event_tx: &mpsc::Sender>, + observer_control_tx: &mpsc::Sender, + state: &mut BgState, + subscription_id: &str, + event: &Event, + ) -> bool { + let keys = Keys::generate(); + let agent_pubkey_hex = keys.public_key().to_hex(); + let text = serde_json::to_string(&json!(["EVENT", subscription_id, event])) + .expect("serialize relay frame"); + handle_ws_message( + Message::Text(text.into()), + ws, + event_tx, + observer_control_tx, + state, + &keys, + "wss://relay.example.com", + &agent_pubkey_hex, + None, + ) + .await + } + + #[tokio::test] + async fn verified_channel_event_is_recorded_and_forwarded() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, mut event_rx) = mpsc::channel(4); + let (observer_control_tx, mut observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let channel_id = Uuid::new_v4(); + let event = make_signed_channel_event(&Keys::generate(), "hello", 2_000); + + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + &channel_sub_id(channel_id), + &event, + ) + .await + ); + + let received = event_rx.try_recv().expect("verified event was forwarded"); + let received = received.expect("event channel should not contain shutdown marker"); + assert_eq!(received.channel_id, channel_id); + assert_eq!(received.event.id, event.id); + assert_eq!(state.last_seen.get(&channel_id), Some(&2_000)); + assert!(state.seen_ids.contains(&event.id.to_hex())); + assert!(matches!( + observer_control_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + } + + #[tokio::test] + async fn tampered_channel_events_are_dropped_before_state_or_queue_changes() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, mut event_rx) = mpsc::channel(8); + let (observer_control_tx, _observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let channel_id = Uuid::new_v4(); + let owner_event = make_signed_channel_event(&Keys::generate(), "status", 2_000); + let other_event = make_signed_channel_event(&Keys::generate(), "other", 3_000); + let other = serde_json::to_value(&other_event).expect("serialize other event"); + let owner_command = recompute_event_id(&replace_event_field( + &owner_event, + "content", + json!("!shutdown"), + )); + + let cases = [ + ( + "changed content", + replace_event_field(&owner_event, "content", json!("tampered")), + ), + ("forged owner command with a matching id", owner_command), + ( + "changed event id", + replace_event_field(&owner_event, "id", other["id"].clone()), + ), + ( + "changed signature", + replace_event_field(&owner_event, "sig", other["sig"].clone()), + ), + ( + "changed author pubkey", + replace_event_field(&owner_event, "pubkey", other["pubkey"].clone()), + ), + ( + "changed tags", + replace_event_field( + &owner_event, + "tags", + json!([["h", Uuid::new_v4().to_string()]]), + ), + ), + ( + "changed timestamp", + replace_event_field(&owner_event, "created_at", json!(4_000)), + ), + ]; + + for (case, event) in cases { + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + &channel_sub_id(channel_id), + &event, + ) + .await, + "{case} should not close the connection" + ); + assert!( + matches!(event_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)), + "{case} reached the harness event queue" + ); + } + + assert!(state.last_seen.is_empty()); + assert!(state.seen_ids.current.is_empty()); + assert!(state.seen_ids.previous.is_empty()); + } + + #[tokio::test] + async fn forged_membership_notification_is_dropped_before_state_or_queue_changes() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, mut event_rx) = mpsc::channel(4); + let (observer_control_tx, _observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let attacker_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let channel_id = Uuid::new_v4(); + let event = EventBuilder::new( + Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16), + "membership changed", + ) + .tags([Tag::parse(["h", &channel_id.to_string()]).expect("h tag")]) + .custom_created_at(nostr::Timestamp::from(2_000)) + .sign_with_keys(&attacker_keys) + .expect("sign membership event"); + let forged = recompute_event_id(&replace_event_field( + &event, + "pubkey", + json!(owner_keys.public_key().to_hex()), + )); + + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + MEMBERSHIP_NOTIF_SUB_ID, + &forged, + ) + .await + ); + + assert!(matches!( + event_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + assert_eq!(state.membership_last_seen, None); + assert!(state.seen_ids.current.is_empty()); + assert!(state.seen_ids.previous.is_empty()); + } + + #[tokio::test] + async fn forged_observer_control_is_dropped_before_control_queue() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, _event_rx) = mpsc::channel(4); + let (observer_control_tx, mut observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let event = make_signed_channel_event(&Keys::generate(), "control", 2_000); + let forged = recompute_event_id(&replace_event_field( + &event, + "content", + json!("tampered control"), + )); + + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + OBSERVER_CONTROL_SUB_ID, + &forged, + ) + .await + ); + + assert!(matches!( + observer_control_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + } + fn test_channel_filter() -> ChannelFilter { ChannelFilter { kinds: Some(vec![9]), From 65925224d08936fca7b33451ad95252aa72e9413 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Fri, 28 Aug 2026 17:05:18 -0700 Subject: [PATCH 2/2] Verify ACP query events before use Signed-off-by: Jordan Mecom --- crates/buzz-acp/src/lib.rs | 16 +- crates/buzz-acp/src/pool.rs | 171 ++++++++++++++---- crates/buzz-acp/src/relay.rs | 341 ++++++++++++++++++++++++++++++++++- 3 files changed, 483 insertions(+), 45 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 25c6e549052..af63177091a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -5693,9 +5693,19 @@ mod author_gate_tests { use std::sync::atomic::Ordering; let id = Uuid::new_v4(); - let response = serde_json::json!([{ - "tags": [["d", id.to_string()], ["name", "DM"], ["t", "dm"]] - }]); + let id_string = id.to_string(); + let event = nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core::kind::KIND_NIP29_GROUP_METADATA as u16), + "", + ) + .tags([ + nostr::Tag::parse(["d", id_string.as_str()]).expect("d tag"), + nostr::Tag::parse(["name", "DM"]).expect("name tag"), + nostr::Tag::parse(["t", "dm"]).expect("type tag"), + ]) + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign channel metadata"); + let response = serde_json::json!([event]); let (resolver, requests, server) = lazy_resolver_with_response(response).await; assert!(is_dm_channel(id, &resolver).await); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index f18f7d6fea2..a1d7bd20f4e 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -4917,6 +4917,44 @@ mod tests { use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; + fn request_queries_kind(request: &[u8], kind: u32) -> bool { + let request = String::from_utf8_lossy(request); + let Some((_, body)) = request.split_once("\r\n\r\n") else { + return false; + }; + let Ok(filters) = serde_json::from_str::(body) else { + return false; + }; + filters.as_array().is_some_and(|filters| { + filters.iter().any(|filter| { + filter + .get("kinds") + .and_then(serde_json::Value::as_array) + .is_some_and(|kinds| { + kinds + .iter() + .any(|candidate| candidate.as_u64() == Some(kind as u64)) + }) + }) + }) + } + + fn signed_event_value( + keys: &Keys, + kind: u32, + tags: Vec, + created_at: u64, + ) -> serde_json::Value { + serde_json::to_value( + EventBuilder::new(Kind::Custom(kind as u16), "") + .tags(tags) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("sign test query event"), + ) + .expect("serialize test query event") + } + fn test_mcp_server() -> McpServer { McpServer { name: "dev".into(), @@ -6450,11 +6488,14 @@ done"# let channel_id = Uuid::new_v4(); let keys = Keys::generate(); + let h_tag = Tag::parse(["h", &channel_id.to_string()]).expect("h tag"); let carry_over = EventBuilder::new(Kind::Custom(9), "merged carry-over sentinel") + .tags([h_tag.clone()]) .sign_with_keys(&keys) .unwrap(); let carry_over_id = carry_over.id.to_hex(); let new_event = EventBuilder::new(Kind::Custom(9), "merged new-event sentinel") + .tags([h_tag]) .sign_with_keys(&keys) .unwrap(); let new_event_id = new_event.id.to_hex(); @@ -6496,10 +6537,17 @@ done"# let server = tokio::spawn(async move { while let Ok((mut socket, _)) = listener.accept().await { let mut request = vec![0; 16 * 1024]; - let _ = socket.read(&mut request).await; + let read = socket.read(&mut request).await.unwrap_or(0); + let body = + if request_queries_kind(&request[..read], buzz_core::kind::KIND_STREAM_MESSAGE) + { + response_body.as_str() + } else { + "[]" + }; let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", - response_body.len(), response_body + body.len(), body ); let _ = socket.write_all(response.as_bytes()).await; } @@ -6624,6 +6672,7 @@ done"# let channel_id = Uuid::new_v4(); let keys = Keys::generate(); let steered_event = EventBuilder::new(Kind::Custom(9), "steered context must not replay") + .tags([Tag::parse(["h", &channel_id.to_string()]).expect("h tag")]) .sign_with_keys(&keys) .unwrap(); let steered_event_id = steered_event.id.to_hex(); @@ -6652,10 +6701,17 @@ done"# let server = tokio::spawn(async move { while let Ok((mut socket, _)) = listener.accept().await { let mut request = vec![0; 16 * 1024]; - let _ = socket.read(&mut request).await; + let read = socket.read(&mut request).await.unwrap_or(0); + let body = + if request_queries_kind(&request[..read], buzz_core::kind::KIND_STREAM_MESSAGE) + { + response_body.as_str() + } else { + "[]" + }; let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", - response_body.len(), response_body + body.len(), body ); let _ = socket.write_all(response.as_bytes()).await; } @@ -8629,12 +8685,20 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let base_url = format!("http://{}", listener.local_addr().unwrap()); let requests = std::sync::Arc::new(AtomicUsize::new(0)); let server_requests = requests.clone(); - let body = response.to_string(); + let metadata_body = response.to_string(); let server = tokio::spawn(async move { while let Ok((mut socket, _)) = listener.accept().await { let mut buf = vec![0; 8192]; - let _ = socket.read(&mut buf).await; + let read = socket.read(&mut buf).await.unwrap_or(0); server_requests.fetch_add(1, Ordering::SeqCst); + let body = if request_queries_kind( + &buf[..read], + buzz_core::kind::KIND_NIP29_GROUP_METADATA, + ) { + metadata_body.as_str() + } else { + "[]" + }; let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), @@ -8657,9 +8721,18 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" } fn channel_metadata_response(id: Uuid, tags: &[[&str; 2]]) -> serde_json::Value { - let mut event_tags = vec![json!(["d", id.to_string()])]; - event_tags.extend(tags.iter().map(|[k, v]| json!([k, v]))); - json!([{ "tags": event_tags }]) + let id = id.to_string(); + let mut event_tags = vec![Tag::parse(["d", id.as_str()]).expect("d tag")]; + event_tags.extend( + tags.iter() + .map(|[key, value]| Tag::parse([*key, *value]).expect("metadata tag")), + ); + json!([signed_event_value( + &Keys::generate(), + buzz_core::kind::KIND_NIP29_GROUP_METADATA, + event_tags, + 2_000, + )]) } #[tokio::test] @@ -8668,19 +8741,29 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" let id = Uuid::new_v4(); let channel = id.to_string(); - let owner = "a".repeat(64); + let owner_keys = Keys::generate(); + let owner = owner_keys.public_key().to_hex(); let coordinate = format!("30617:{owner}:app"); let responses = [ - json!([{ - "kind": 30621, - "pubkey": owner, - "tags": [["d", "app"], ["buzz-channel", channel], ["a", coordinate]] - }]), - json!([{ - "kind": 30617, - "pubkey": "a".repeat(64), - "tags": [["d", "app"], ["buzz-channel", id.to_string()]] - }]), + json!([signed_event_value( + &owner_keys, + buzz_core::kind::KIND_PROJECT, + vec![ + Tag::parse(["d", "app"]).expect("project d tag"), + Tag::parse(["buzz-channel", channel.as_str()]).expect("project channel tag"), + Tag::parse(["a", coordinate.as_str()]).expect("project repo tag"), + ], + 2_000, + )]), + json!([signed_event_value( + &owner_keys, + buzz_core::kind::KIND_GIT_REPO_ANNOUNCEMENT, + vec![ + Tag::parse(["d", "app"]).expect("repo d tag"), + Tag::parse(["buzz-channel", channel.as_str()]).expect("repo channel tag"), + ], + 1_999, + )]), ]; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let base_url = format!("http://{}", listener.local_addr().unwrap()); @@ -8947,30 +9030,46 @@ done"# let id = Uuid::new_v4(); let channel = id.to_string(); - let owner = "a".repeat(64); + let owner_keys = Keys::generate(); + let owner = owner_keys.public_key().to_hex(); let coordinate = format!("30617:{owner}:app"); + let decoy_keys = Keys::generate(); let first_page: Vec<_> = (0..500) .map(|index| { - json!({ - "id": format!("{index:064x}"), - "created_at": 1_000 - index, - "kind": 30621, - "pubkey": "b".repeat(64), - "tags": [["d", format!("decoy-{index}")], ["buzz-channel", channel]] - }) + let identifier = format!("decoy-{index}"); + signed_event_value( + &decoy_keys, + buzz_core::kind::KIND_PROJECT, + vec![ + Tag::parse(["d", identifier.as_str()]).expect("decoy d tag"), + Tag::parse(["buzz-channel", channel.as_str()]).expect("decoy channel tag"), + ], + 1_000 - index, + ) }) .collect(); let responses = [ channel_metadata_response(id, &[["name", "project-home"], ["t", "stream"]]), serde_json::Value::Array(first_page), - json!([{ - "id": "f".repeat(64), "created_at": 1, "kind": 30621, "pubkey": owner, - "tags": [["d", "app"], ["buzz-channel", channel], ["a", coordinate]] - }]), - json!([{ - "id": "e".repeat(64), "created_at": 1, "kind": 30617, "pubkey": "a".repeat(64), - "tags": [["d", "app"], ["buzz-channel", id.to_string()]] - }]), + json!([signed_event_value( + &owner_keys, + buzz_core::kind::KIND_PROJECT, + vec![ + Tag::parse(["d", "app"]).expect("project d tag"), + Tag::parse(["buzz-channel", channel.as_str()]).expect("project channel tag"), + Tag::parse(["a", coordinate.as_str()]).expect("project repo tag"), + ], + 1, + )]), + json!([signed_event_value( + &owner_keys, + buzz_core::kind::KIND_GIT_REPO_ANNOUNCEMENT, + vec![ + Tag::parse(["d", "app"]).expect("repo d tag"), + Tag::parse(["buzz-channel", channel.as_str()]).expect("repo channel tag"), + ], + 1, + )]), ]; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let base_url = format!("http://{}", listener.local_addr().unwrap()); diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 81fc84c50fb..4ce8986ad23 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -276,6 +276,106 @@ fn unix_now_secs() -> u64 { .as_secs() } +/// Verify an HTTP `/query` response before any caller can inspect event fields. +/// +/// A relay controls both the response bytes and which events it returns. NIP-01 +/// verification authenticates each event, while local filter matching prevents +/// a valid event from being substituted into a query for a different author, +/// kind, channel, or thread. +async fn verify_query_response(response: Value, filters: Vec) -> Result { + tokio::task::spawn_blocking(move || verify_query_response_blocking(response, &filters)) + .await + .map_err(|error| RelayError::Http(format!("query verification task failed: {error}")))? +} + +fn verify_query_response_blocking(response: Value, filters: &[Value]) -> Result { + if filters.is_empty() { + return Err(RelayError::Http( + "query verification requires at least one filter".into(), + )); + } + + let events = response + .as_array() + .ok_or_else(|| RelayError::Http("query response is not an array".into()))?; + + for (index, raw) in events.iter().enumerate() { + let event: Event = serde_json::from_value(raw.clone()).map_err(|error| { + RelayError::Http(format!( + "query response event {index} is malformed: {error}" + )) + })?; + buzz_core::verify_event(&event).map_err(|error| { + RelayError::Http(format!( + "query response event {index} failed NIP-01 verification: {error}" + )) + })?; + + let mut matched = false; + for filter in filters { + if query_filter_matches_event(filter, &event)? { + matched = true; + break; + } + } + if !matched { + return Err(RelayError::Http(format!( + "query response event {index} does not match any requested filter" + ))); + } + } + + Ok(response) +} + +fn query_filter_matches_event(filter: &Value, event: &Event) -> Result { + let object = filter + .as_object() + .ok_or_else(|| RelayError::Http("query filter is not an object".into()))?; + let nostr_filter: nostr::Filter = serde_json::from_value(filter.clone()) + .map_err(|error| RelayError::Http(format!("invalid query filter: {error}")))?; + + if !nostr_filter.match_event(event, nostr::filter::MatchEventOptions::default()) { + return Ok(false); + } + + // `nostr::Filter` supports NIP-01's single-letter generic tags. Buzz also + // queries multi-character tags such as `#buzz-channel`, so enforce every + // raw `#...` constraint explicitly as well. + for (key, value) in object { + let Some(tag_name) = key.strip_prefix('#') else { + continue; + }; + if tag_name.is_empty() { + return Err(RelayError::Http( + "query filter has an empty tag name".into(), + )); + } + let allowed = value + .as_array() + .ok_or_else(|| RelayError::Http(format!("query filter {key} value is not an array")))?; + let allowed: Vec<&str> = allowed + .iter() + .map(|value| { + value.as_str().ok_or_else(|| { + RelayError::Http(format!("query filter {key} contains a non-string value")) + }) + }) + .collect::>()?; + let tag_matches = event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() >= 2 + && parts[0] == tag_name + && allowed.iter().any(|value| parts[1] == *value) + }); + if !tag_matches { + return Ok(false); + } + } + + Ok(true) +} + impl RestClient { /// Sign a NIP-98 HTTP Auth event (kind:27235) for the given method/URL/body. /// @@ -415,12 +515,19 @@ impl RestClient { /// Accepts a slice of `nostr::Filter` (serialized as JSON array). /// Returns the events as a `serde_json::Value` (JSON array of event objects). pub async fn query(&self, filters: &[nostr::Filter]) -> Result { - let body_bytes = serde_json::to_vec(filters) + let filter_values = filters + .iter() + .map(serde_json::to_value) + .collect::, _>>() + .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; + let body_bytes = serde_json::to_vec(&filter_values) .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; let resp = self.bridge_post("/query", &body_bytes).await?; - resp.json() + let response = resp + .json() .await - .map_err(|e| RelayError::Http(e.to_string())) + .map_err(|e| RelayError::Http(e.to_string()))?; + verify_query_response(response, filter_values).await } /// Query events via `POST /query` with a raw NIP-01 filter document. @@ -431,9 +538,11 @@ impl RestClient { let body_bytes = serde_json::to_vec(filters) .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; let resp = self.bridge_post("/query", &body_bytes).await?; - resp.json() + let response = resp + .json() .await - .map_err(|e| RelayError::Http(e.to_string())) + .map_err(|e| RelayError::Http(e.to_string()))?; + verify_query_response(response, filters.to_vec()).await } /// Query every historical event matching one raw filter across bounded pages. @@ -2256,6 +2365,26 @@ async fn handle_ws_message( Err(mpsc::error::TrySendError::Closed(_)) => return false, } } else if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { + let active_subscription = state + .active_subscriptions + .get(&channel_id) + .map(String::as_str); + if active_subscription != Some(subscription_id.as_str()) { + warn!( + channel_id = %channel_id, + subscription_id, + "received EVENT for inactive channel subscription — dropping" + ); + return true; + } + if !event_has_h_tag(&event, channel_id) { + warn!( + channel_id = %channel_id, + event_id = %event.id.to_hex(), + "channel EVENT does not carry the subscribed h tag — dropping" + ); + return true; + } let ts = event.created_at.as_secs(); let event_id_hex = event.id.to_hex(); if state.record_event(channel_id, &event) { @@ -3545,6 +3674,14 @@ fn extract_h_tag_uuid(event: &nostr::Event) -> Option { }) } +fn event_has_h_tag(event: &nostr::Event, channel_id: Uuid) -> bool { + let expected = channel_id.to_string(); + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() >= 2 && parts[0] == "h" && parts[1] == expected + }) +} + /// Build and send a NIP-42 AUTH response event. /// /// If `auth_tag` is provided (NIP-OA owner attestation), it is included in the @@ -4533,6 +4670,19 @@ mod tests { .expect("sign channel event") } + fn make_signed_channel_scoped_event( + keys: &Keys, + channel_id: Uuid, + content: &str, + created_at_secs: u64, + ) -> Event { + EventBuilder::new(Kind::Custom(9), content) + .tags([Tag::parse(["h", &channel_id.to_string()]).expect("h tag")]) + .custom_created_at(nostr::Timestamp::from(created_at_secs)) + .sign_with_keys(keys) + .expect("sign channel event") + } + fn replace_event_field(event: &Event, field: &str, replacement: Value) -> Event { let mut value = serde_json::to_value(event).expect("serialize event"); value[field] = replacement; @@ -4550,6 +4700,115 @@ mod tests { replace_event_field(event, "id", json!(id.to_hex())) } + #[test] + fn query_response_accepts_verified_event_matching_any_requested_filter() { + let keys = Keys::generate(); + let channel_id = Uuid::new_v4(); + let event = make_signed_channel_scoped_event(&keys, channel_id, "hello", 2_000); + let filters = vec![ + serde_json::to_value(nostr::Filter::new().kind(Kind::Metadata)) + .expect("serialize first filter"), + serde_json::to_value( + nostr::Filter::new() + .kind(Kind::Custom(9)) + .author(keys.public_key()) + .custom_tags( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + [channel_id.to_string()], + ), + ) + .expect("serialize matching filter"), + ]; + let response = json!([event]); + + let verified = verify_query_response_blocking(response.clone(), &filters) + .expect("valid matching query response"); + + assert_eq!(verified, response); + } + + #[test] + fn query_response_rejects_event_with_recomputed_id_and_stale_signature() { + let channel_id = Uuid::new_v4(); + let event = make_signed_channel_scoped_event(&Keys::generate(), channel_id, "safe", 2_000); + let forged = recompute_event_id(&replace_event_field( + &event, + "content", + json!("forged prompt context"), + )); + let filter = serde_json::to_value(nostr::Filter::new().kind(Kind::Custom(9)).custom_tags( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + [channel_id.to_string()], + )) + .expect("serialize filter"); + + let error = verify_query_response_blocking(json!([forged]), &[filter]) + .expect_err("forged query event must be rejected"); + + assert!(error.to_string().contains("failed NIP-01 verification")); + } + + #[test] + fn query_response_rejects_valid_event_outside_requested_filter() { + let requested_channel = Uuid::new_v4(); + let other_channel = Uuid::new_v4(); + let event = make_signed_channel_scoped_event( + &Keys::generate(), + other_channel, + "wrong channel", + 2_000, + ); + let filter = serde_json::to_value(nostr::Filter::new().kind(Kind::Custom(9)).custom_tags( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + [requested_channel.to_string()], + )) + .expect("serialize filter"); + + let error = verify_query_response_blocking(json!([event]), &[filter]) + .expect_err("relay must not substitute a valid event from another channel"); + + assert!(error + .to_string() + .contains("does not match any requested filter")); + } + + #[test] + fn query_response_enforces_multi_character_tag_filters() { + let channel_id = Uuid::new_v4(); + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), + "project", + ) + .tags([Tag::parse(["buzz-channel", &channel_id.to_string()]).expect("buzz-channel tag")]) + .custom_created_at(nostr::Timestamp::from(2_000)) + .sign_with_keys(&Keys::generate()) + .expect("sign project event"); + let matching_filter = json!({ + "kinds": [buzz_core::kind::KIND_PROJECT], + "#buzz-channel": [channel_id.to_string()], + }); + let wrong_filter = json!({ + "kinds": [buzz_core::kind::KIND_PROJECT], + "#buzz-channel": [Uuid::new_v4().to_string()], + }); + + assert!(verify_query_response_blocking(json!([event.clone()]), &[matching_filter]).is_ok()); + assert!(verify_query_response_blocking(json!([event]), &[wrong_filter]).is_err()); + } + + #[test] + fn query_response_rejects_malformed_or_non_array_payloads() { + let filter = serde_json::to_value(nostr::Filter::new().kind(Kind::Custom(9))) + .expect("serialize filter"); + + assert!(verify_query_response_blocking(json!({}), std::slice::from_ref(&filter)).is_err()); + assert!(verify_query_response_blocking( + json!([{"id": "not-an-event"}]), + std::slice::from_ref(&filter), + ) + .is_err()); + } + async fn handle_test_relay_event( ws: &mut WsStream, event_tx: &mpsc::Sender>, @@ -4583,7 +4842,10 @@ mod tests { let (observer_control_tx, mut observer_control_rx) = mpsc::channel(4); let mut state = BgState::new(); let channel_id = Uuid::new_v4(); - let event = make_signed_channel_event(&Keys::generate(), "hello", 2_000); + state + .active_subscriptions + .insert(channel_id, channel_sub_id(channel_id)); + let event = make_signed_channel_scoped_event(&Keys::generate(), channel_id, "hello", 2_000); assert!( handle_test_relay_event( @@ -4609,6 +4871,73 @@ mod tests { )); } + #[tokio::test] + async fn verified_event_for_inactive_channel_is_dropped_before_state_or_queue_changes() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, mut event_rx) = mpsc::channel(4); + let (observer_control_tx, _observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let channel_id = Uuid::new_v4(); + let event = make_signed_channel_scoped_event(&Keys::generate(), channel_id, "hello", 2_000); + + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + &channel_sub_id(channel_id), + &event, + ) + .await + ); + + assert!(matches!( + event_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + assert!(state.last_seen.is_empty()); + assert!(!state.seen_ids.contains(&event.id.to_hex())); + } + + #[tokio::test] + async fn verified_event_with_wrong_channel_tag_is_dropped_before_state_or_queue_changes() { + let (mut client, _server) = test_ws_pair().await; + let (event_tx, mut event_rx) = mpsc::channel(4); + let (observer_control_tx, _observer_control_rx) = mpsc::channel(4); + let mut state = BgState::new(); + let subscribed_channel = Uuid::new_v4(); + let signed_channel = Uuid::new_v4(); + state + .active_subscriptions + .insert(subscribed_channel, channel_sub_id(subscribed_channel)); + let event = make_signed_channel_scoped_event( + &Keys::generate(), + signed_channel, + "wrong channel", + 2_000, + ); + + assert!( + handle_test_relay_event( + &mut client, + &event_tx, + &observer_control_tx, + &mut state, + &channel_sub_id(subscribed_channel), + &event, + ) + .await + ); + + assert!(matches!( + event_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + assert!(state.last_seen.is_empty()); + assert!(!state.seen_ids.contains(&event.id.to_hex())); + } + #[tokio::test] async fn tampered_channel_events_are_dropped_before_state_or_queue_changes() { let (mut client, _server) = test_ws_pair().await;