diff --git a/Justfile b/Justfile index 32d83355e1c..cc28fa5aaef 100644 --- a/Justfile +++ b/Justfile @@ -325,6 +325,9 @@ test-unit: cargo test -p buzz-auth --doc cargo nextest run -p buzz-voice --lib cargo nextest run -p buzz-cli + # buzz-acp owns the relay-to-agent trust boundary. Run its tests here so + # forged relay events cannot regain a path into agent routing unnoticed. + cargo nextest run -p buzz-acp # buzz-db migrator/lint tests: pure SQL-parsing unit tests (no infra). # They guard the embedded-migrator invariant (the complete checked-in # additive migration set; legacy cutover/backfill remains an operator 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]), diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 9dca8c82c37..cc987c53b35 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -90,6 +90,11 @@ run_unit_tests() { run_test_step "buzz-cli tests" \ cargo test -p buzz-cli -- --nocapture + # Keep the relay-to-agent trust-boundary regressions in the fallback path + # when cargo-nextest is unavailable. + run_test_step "buzz-acp tests" \ + cargo test -p buzz-acp -- --nocapture + # buzz-db migrator/lint unit tests (no infra): guard the embedded-migrator # invariant (exactly the consolidated 0001; cutover/backfill stays an operator # script, not startup state) and the tenant-scoping lints. The Postgres-backed