From 00a76985c253bb97a22636d1def7149787e48e4f Mon Sep 17 00:00:00 2001 From: jm Date: Wed, 8 Jul 2026 16:51:23 -0400 Subject: [PATCH] Scope relay runtime state by community --- crates/buzz-relay/src/audio/handler.rs | 20 +++-- crates/buzz-relay/src/audio/room.rs | 64 +++++++++++---- crates/buzz-relay/src/handlers/auth.rs | 6 +- crates/buzz-relay/src/handlers/event.rs | 100 +++++++++++++++++++++++- crates/buzz-relay/src/state.rs | 7 +- 5 files changed, 173 insertions(+), 24 deletions(-) diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 9e930ccc40..63cc2f7a76 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -253,7 +253,9 @@ async fn handle_audio_connection( return; } - let room = state.audio_rooms.get_or_create(channel_id); + let room = state + .audio_rooms + .get_or_create(tenant.community(), channel_id); // Re-check archived status after obtaining the room. This closes the // cross-boundary race: a joiner that passed ensure_membership before @@ -271,12 +273,16 @@ async fn handle_audio_connection( .into(), )) .await; - state.audio_rooms.cleanup_if_empty(channel_id); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); return; } Err(e) => { warn!(channel_id = %channel_id, "pre-join channel check failed (fail-closed): {e}"); - state.audio_rooms.cleanup_if_empty(channel_id); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); return; } Ok(_) => {} // Channel exists and is not archived — proceed. @@ -475,7 +481,9 @@ async fn handle_audio_connection( room.clear_ended(); } Ok(()) => { - state.audio_rooms.cleanup_if_empty(channel_id); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); emit_participant_event( &state, @@ -489,7 +497,9 @@ async fn handle_audio_connection( } } } else { - state.audio_rooms.cleanup_if_empty(channel_id); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); } info!( diff --git a/crates/buzz-relay/src/audio/room.rs b/crates/buzz-relay/src/audio/room.rs index d0a70fbe7c..5639e45fa9 100644 --- a/crates/buzz-relay/src/audio/room.rs +++ b/crates/buzz-relay/src/audio/room.rs @@ -8,6 +8,7 @@ //! Frames are opaque Opus bytes — the relay never decodes audio. //! `try_send` is used throughout: real-time audio tolerates drops, never queues. +use buzz_core::CommunityId; use bytes::Bytes; use dashmap::DashMap; use std::sync::Arc; @@ -87,8 +88,8 @@ struct AdmissionGuard { /// /// Pin is per-`Room`-instance and clears when the manager evicts the /// Room via [`AudioRoomManager::cleanup_if_empty`] — the next - /// `get_or_create` for the same channel id then constructs a fresh - /// `Room` with a fresh `AdmissionGuard` (and therefore `None` pin), + /// `get_or_create` for the same community-local channel then constructs a + /// fresh `Room` with a fresh `AdmissionGuard` (and therefore `None` pin), /// so a new generation of joiners can negotiate a new version. /// A momentarily-empty-but-not-yet-cleaned-up Room keeps its pin so /// reconnecting peers don't accidentally renegotiate mid-call. See @@ -126,6 +127,8 @@ impl AdmissionGuard { /// A single audio room for one channel. pub struct Room { + /// Community this room belongs to. + pub community_id: CommunityId, /// Channel UUID this room belongs to. pub channel_id: Uuid, /// Connected peers keyed by peer UUID. @@ -135,9 +138,10 @@ pub struct Room { } impl Room { - /// Create an empty room for the given channel. - pub fn new(channel_id: Uuid) -> Self { + /// Create an empty room for the given community-local channel. + pub fn new(community_id: CommunityId, channel_id: Uuid) -> Self { Self { + community_id, channel_id, peers: DashMap::new(), guard: std::sync::Mutex::new(AdmissionGuard::new()), @@ -321,7 +325,7 @@ impl Room { /// Global registry of active audio rooms. pub struct AudioRoomManager { - rooms: DashMap>, + rooms: DashMap<(CommunityId, Uuid), Arc>, } impl AudioRoomManager { @@ -333,17 +337,21 @@ impl AudioRoomManager { } /// Get an existing room or create a new one. - pub fn get_or_create(&self, channel_id: Uuid) -> Arc { + /// + /// Channel UUIDs are only unique inside a community. The room key must + /// carry both labels so two tenants that legitimately reuse the same UUID + /// never share peer lists, protocol pins, or audio frames. + pub fn get_or_create(&self, community_id: CommunityId, channel_id: Uuid) -> Arc { self.rooms - .entry(channel_id) - .or_insert_with(|| Arc::new(Room::new(channel_id))) + .entry((community_id, channel_id)) + .or_insert_with(|| Arc::new(Room::new(community_id, channel_id))) .clone() } /// Remove the room if it has no peers. Returns `true` if the room was removed. - pub fn cleanup_if_empty(&self, channel_id: Uuid) -> bool { + pub fn cleanup_if_empty(&self, community_id: CommunityId, channel_id: Uuid) -> bool { self.rooms - .remove_if(&channel_id, |_, room| room.is_empty()) + .remove_if(&(community_id, channel_id), |_, room| room.is_empty()) .is_some() } } @@ -359,7 +367,7 @@ mod tests { use super::*; fn fresh_room() -> Room { - Room::new(Uuid::new_v4()) + Room::new(CommunityId::from_uuid(Uuid::new_v4()), Uuid::new_v4()) } /// First peer's `requested_version` becomes the room's pin; later peers @@ -424,9 +432,10 @@ mod tests { #[test] fn manager_cleanup_resets_version_pin() { let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); let channel_id = Uuid::new_v4(); - let room1 = manager.get_or_create(channel_id); + let room1 = manager.get_or_create(community_id, channel_id); let (peer_id, _, _, _) = room1 .add_peer("alice".to_string(), 2) .expect("first peer admits"); @@ -435,16 +444,43 @@ mod tests { .remove_peer_and_check_ended(peer_id) .expect("peer existed"); assert!(ended, "single-peer room should end on its last departure"); - assert!(manager.cleanup_if_empty(channel_id)); + assert!(manager.cleanup_if_empty(community_id, channel_id)); // Next joiner with a different version on the same channel id gets a // brand-new room (no v=2 pin carried over from the prior generation). - let room2 = manager.get_or_create(channel_id); + let room2 = manager.get_or_create(community_id, channel_id); let _ = room2 .add_peer("bob".to_string(), 1) .expect("fresh room must accept any version"); } + #[test] + fn manager_isolates_same_channel_uuid_across_communities() { + let manager = AudioRoomManager::new(); + let channel_id = Uuid::new_v4(); + let community_a = CommunityId::from_uuid(Uuid::new_v4()); + let community_b = CommunityId::from_uuid(Uuid::new_v4()); + + let room_a = manager.get_or_create(community_a, channel_id); + let room_b = manager.get_or_create(community_b, channel_id); + + assert!( + !Arc::ptr_eq(&room_a, &room_b), + "same channel UUID in two communities must create distinct rooms" + ); + assert_eq!(room_a.community_id, community_a); + assert_eq!(room_b.community_id, community_b); + + room_a + .add_peer("alice".to_string(), 1) + .expect("A peer admits"); + assert_eq!(room_a.peer_pubkeys(), vec![("alice".to_string(), 0)]); + assert!( + room_b.peer_pubkeys().is_empty(), + "A room peers must not appear in B's same-UUID room" + ); + } + /// Peer-index reuse: after a peer leaves, their index is released; a new /// peer joining the same (still-pinned) room reuses the freed index. /// Version pin must persist across this reuse — the room generation diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index a5d6be50c7..a050a29366 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -284,7 +284,11 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: // Successfully materialized — this owner is authoritative. auth_ctx.agent_owner_pubkey = Some(owner); // Pre-warm the observer cache to avoid stale negatives. - let cache_key = (pubkey.to_bytes().to_vec(), owner.to_bytes().to_vec()); + let cache_key = ( + conn.tenant.community(), + pubkey.to_bytes().to_vec(), + owner.to_bytes().to_vec(), + ); state.observer_owner_cache.insert(cache_key, true); } Ok(false) => { diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 9ab57bab86..5f3393b99c 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -997,7 +997,11 @@ async fn handle_agent_observer_event( let agent_bytes = route.agent.to_bytes().to_vec(); let owner_bytes = route.owner.to_bytes().to_vec(); - let cache_key = (agent_bytes.clone(), owner_bytes.clone()); + let cache_key = ( + conn.tenant.community(), + agent_bytes.clone(), + owner_bytes.clone(), + ); let is_owner = if session_owner_match { true } else { @@ -1153,6 +1157,8 @@ fn single_tag_content<'a>(event: &'a Event, tag_name: &str) -> Result<&'a str, S #[cfg(test)] mod tests { + use std::collections::HashMap; + use std::sync::atomic::AtomicU8; use std::sync::Arc; use buzz_core::kind::{ @@ -1164,6 +1170,9 @@ mod tests { OBSERVER_FRAME_TELEMETRY, }; use nostr::{EventBuilder, Keys, Kind, Tag}; + use tokio::sync::{mpsc, Mutex, RwLock}; + use tokio_util::sync::CancellationToken; + use uuid::Uuid; #[test] fn fanout_event_frame_matches_legacy_format_byte_for_byte() { @@ -1304,6 +1313,95 @@ mod tests { assert!(err.contains("NIP-44")); } + #[tokio::test] + async fn observer_owner_cache_is_scoped_to_community() { + let state = fanout_access::test_state().await; + let agent = Keys::generate(); + let owner = Keys::generate(); + let agent_bytes = agent.public_key().to_bytes().to_vec(); + let owner_bytes = owner.public_key().to_bytes().to_vec(); + let community_a = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + let community_b = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + + state.observer_owner_cache.insert( + (community_a, agent_bytes.clone(), owner_bytes.clone()), + true, + ); + assert_eq!( + state.observer_owner_cache.get(&( + community_b, + agent_bytes.clone(), + owner_bytes.clone() + )), + None, + "A cached allow must not populate B's observer authorization key" + ); + state.observer_owner_cache.insert( + (community_b, agent_bytes.clone(), owner_bytes.clone()), + false, + ); + + let encrypted = encrypt_observer_payload( + &agent, + &owner.public_key(), + &serde_json::json!({"type": "acp_read"}), + ) + .expect("encrypt observer payload"); + let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted) + .tags([ + Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"), + Tag::parse([OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"), + Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"), + ]) + .sign_with_keys(&agent) + .expect("sign event"); + + let (send_tx, mut send_rx) = mpsc::channel(1); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(1); + let conn = Arc::new(crate::connection::ConnectionState { + conn_id: Uuid::new_v4(), + tenant: buzz_core::TenantContext::resolved(community_b, "b.example"), + remote_addr: "127.0.0.1:1234".parse().expect("socket addr"), + auth_state: RwLock::new(crate::connection::AuthState::Authenticated( + buzz_auth::AuthContext { + pubkey: agent.public_key(), + scopes: vec![], + channel_ids: None, + auth_method: buzz_auth::AuthMethod::Nip42, + agent_owner_pubkey: None, + }, + )), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + send_tx, + ctrl_tx, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + + super::handle_agent_observer_event( + event.clone(), + conn.conn_id, + &event.id.to_hex(), + conn, + state, + ) + .await; + + let axum::extract::ws::Message::Text(text) = + send_rx.try_recv().expect("observer rejection sent") + else { + panic!("expected text relay message"); + }; + let frame: serde_json::Value = serde_json::from_str(&text).expect("relay frame JSON"); + assert_eq!(frame[0], "OK"); + assert_eq!(frame[2], false); + assert_eq!( + frame[3], + "restricted: observer frame is not authorized for this agent owner" + ); + } + mod pubsub_fanout { use std::collections::HashMap; use std::sync::atomic::AtomicU8; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index ce955524fc..a3041df82c 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -357,11 +357,12 @@ pub struct AppState { /// Current in-flight media uploads per uploader pubkey. pub media_uploads_in_flight: Arc>, /// Cache for observer agent-owner authorization (kind 24200). - /// Key: (agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner. - /// agent_owner_pubkey is immutable so a long TTL (5 min) is safe. + /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner. + /// `agent_owner_pubkey` is immutable inside one community, so a long TTL + /// (5 min) is safe once the community label is part of the key. /// Prevents repeated DB lookups from bursty observer traffic. #[allow(clippy::type_complexity)] - pub observer_owner_cache: Arc, Vec), bool>>, + pub observer_owner_cache: Arc, Vec), bool>>, /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`] /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to