diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 127f1fc40e0..d03494ab94c 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -275,10 +275,17 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } info!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "NIP-42 auth successful"); + let membership_pubkey = auth_ctx + .agent_owner_pubkey + .unwrap_or(pubkey) + .to_bytes() + .to_vec(); *conn.auth_state.write().await = AuthState::Authenticated(auth_ctx); - state - .conn_manager - .set_authenticated_pubkey(conn_id, pubkey.to_bytes().to_vec()); + state.conn_manager.set_authenticated_membership( + conn_id, + pubkey.to_bytes().to_vec(), + membership_pubkey, + ); conn.send(RelayMessage::ok(&event_id_hex, true, "")); } Err(e) => { diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ecb3e41fcc7..91c36a325e2 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -608,13 +608,14 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc ( conn.conn_id, ctx.pubkey.to_bytes().to_vec(), ctx.pubkey, + ctx.agent_owner_pubkey.unwrap_or(ctx.pubkey), ctx.scopes.clone(), ctx.channel_ids.clone(), ), @@ -630,6 +631,40 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc {} + Ok(false) => { + reject("membership"); + state.conn_manager.disconnect_membership_principal( + conn.tenant.community(), + membership_pubkey.as_bytes(), + &event_id_hex, + "blocked: relay membership removed", + ); + return; + } + Err(error) => { + warn!(%error, "relay membership recheck failed"); + reject("membership"); + conn.send(RelayMessage::ok( + &event_id_hex, + false, + "restricted: relay membership could not be verified", + )); + return; + } + } + } + // Must run before both ephemeral and persistent branches. Persistent // events get a second check inside ingest_event() (step 3), but // ephemeral events bypass the pipeline entirely. diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 840d9dcfe58..20c89eca14f 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -29,6 +29,38 @@ use crate::handlers::side_effects::{ }; use crate::state::AppState; +const RELAY_MEMBERSHIP_REMOVED_REASON: &str = "blocked: relay membership removed"; + +#[derive(Debug, Eq, PartialEq)] +enum RemovalDisposition { + Removed, + AlreadyAbsent, +} + +fn removal_disposition( + result: RemoveResult, + target_is_virtual_agent: bool, + require_relay_membership: bool, +) -> Result { + match result { + RemoveResult::Removed => Ok(RemovalDisposition::Removed), + // A repeated command must be safe and must re-fan-out the disconnect + // in case the first Redis publish was lost. Durable absence is still + // the authorization backstop. A NIP-OA virtual agent is deliberately + // absent from relay_members, so it is not a valid idempotent retry. + RemoveResult::NotFound if require_relay_membership && !target_is_virtual_agent => { + Ok(RemovalDisposition::AlreadyAbsent) + } + RemoveResult::NotFound => { + Err("member not found: absence does not prove a removable member".to_string()) + } + RemoveResult::IsOwner => Err("cannot remove the relay owner".to_string()), + RemoveResult::RoleMismatch => { + Err("actor not authorized: admins can only remove members".to_string()) + } + } +} + /// Extract the hex pubkey from the first `p` tag, returning it as a `String`. fn extract_p_tag_hex(event: &Event) -> Option { for tag in event.tags.iter() { @@ -252,30 +284,50 @@ pub async fn handle_relay_admin_event( .map_err(|e| format!("database error: {e}"))? }; - match remove_result { - RemoveResult::Removed => {} - RemoveResult::IsOwner => { - return Err("cannot remove the relay owner".to_string()); - } - RemoveResult::NotFound => { - return Err(format!("member not found: {target_hex}")); - } - RemoveResult::RoleMismatch => { - return Err("actor not authorized: admins can only remove members".to_string()); - } - } + let target_pubkey = hex::decode(&target_hex) + .map_err(|e| format!("invalid target pubkey encoding: {e}"))?; + let target_is_virtual_agent = if remove_result == RemoveResult::NotFound { + state + .db + .get_agent_channel_policy(tenant.community(), &target_pubkey) + .await + .map_err(|e| format!("database error: {e}"))? + .is_some_and(|(_, owner)| owner.is_some()) + } else { + false + }; + let disposition = removal_disposition( + remove_result, + target_is_virtual_agent, + state.config.require_relay_membership, + )?; + + // The DB mutation/absence check above is the durable authorization + // backstop. Close matching sockets on this pod and fan the same + // community-scoped command to every other pod. Repeating an + // already-completed removal intentionally replays this fan-out. + let closed_locally = state.disconnect_pubkey_clusterwide( + tenant, + &target_pubkey, + &event.id.to_hex(), + RELAY_MEMBERSHIP_REMOVED_REASON, + ); info!( sender = %sender_hex, target = %target_hex, - "relay member removed" + ?disposition, + closed_locally, + "relay member removal enforced" ); - if let Err(e) = publish_nip43_member_removed(tenant, state, &target_hex).await { - warn!(error = %e, "failed to publish NIP-43 member removed event"); - } - if let Err(e) = publish_nip43_membership_list(tenant, state).await { - warn!(error = %e, "failed to publish NIP-43 membership list"); + if disposition == RemovalDisposition::Removed { + if let Err(e) = publish_nip43_member_removed(tenant, state, &target_hex).await { + warn!(error = %e, "failed to publish NIP-43 member removed event"); + } + if let Err(e) = publish_nip43_membership_list(tenant, state).await { + warn!(error = %e, "failed to publish NIP-43 membership list"); + } } } @@ -431,6 +483,46 @@ mod tests { assert_eq!(extract_tag_value(&event, "p"), None); } + #[test] + fn removal_disposition_accepts_idempotent_retry() { + assert_eq!( + removal_disposition(RemoveResult::Removed, false, true), + Ok(RemovalDisposition::Removed) + ); + assert_eq!( + removal_disposition(RemoveResult::NotFound, false, true), + Ok(RemovalDisposition::AlreadyAbsent) + ); + } + + #[test] + fn removal_disposition_rejects_virtual_agent_as_retry() { + assert_eq!( + removal_disposition(RemoveResult::NotFound, true, true), + Err("member not found: absence does not prove a removable member".to_string()) + ); + } + + #[test] + fn removal_disposition_rejects_absent_target_on_open_relay() { + assert_eq!( + removal_disposition(RemoveResult::NotFound, false, false), + Err("member not found: absence does not prove a removable member".to_string()) + ); + } + + #[test] + fn removal_disposition_preserves_protected_role_errors() { + assert_eq!( + removal_disposition(RemoveResult::IsOwner, false, true), + Err("cannot remove the relay owner".to_string()) + ); + assert_eq!( + removal_disposition(RemoveResult::RoleMismatch, false, true), + Err("actor not authorized: admins can only remove members".to_string()) + ); + } + #[test] fn workspace_icon_empty_ok() { assert!(validate_workspace_icon("").is_ok()); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index be9794922bb..75c37b3e60f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -892,6 +892,21 @@ async fn main() -> anyhow::Result<()> { )); } + // Durable relay-membership backstop: a pod can miss a Redis disconnect + // while offline or disconnected. Recheck only principals that currently + // authorize local sockets and converge within 15 seconds. + if state.config.require_relay_membership { + let membership_state = Arc::clone(&state); + let interval_value = std::env::var("BUZZ_RELAY_MEMBERSHIP_REVALIDATE_INTERVAL_SECS").ok(); + let interval_secs = membership_revalidate_interval_secs(interval_value.as_deref()); + let cancel = membership_state.community_revalidator_cancel.clone(); + tokio::spawn(run_membership_revalidator( + membership_state, + std::time::Duration::from_secs(interval_secs), + cancel, + )); + } + // Cross-pod connection-control consumer: receive disconnect commands from // Redis pub/sub (published by the pod that recorded a ban) and close any // matching local sockets. A member's live connections may land on any pod, @@ -1077,6 +1092,30 @@ async fn run_community_revalidator( .await; } +async fn run_membership_revalidator( + state: Arc, + period: std::time::Duration, + cancel: CancellationToken, +) { + run_periodic_until_cancelled(period, cancel, || async { + let closed = state.revalidate_live_memberships().await; + if closed > 0 { + tracing::info!( + closed, + "closed sockets for removed relay members during revalidation" + ); + } + }) + .await; +} + +fn membership_revalidate_interval_secs(value: Option<&str>) -> u64 { + value + .and_then(|value| value.parse::().ok()) + .unwrap_or(10) + .clamp(1, 15) +} + async fn run_periodic_until_cancelled( period: std::time::Duration, cancel: CancellationToken, @@ -1805,8 +1844,8 @@ mod tests { use super::{ buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, - refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope, - InMemoryMetricKey, + membership_revalidate_interval_secs, refresh_legacy_active_gauge_recency, + run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, }; use metrics::GaugeFn; use metrics_util::{ @@ -1854,6 +1893,14 @@ mod tests { assert!(buzz_auto_migrate_enabled(Some("on"))); } + #[test] + fn membership_revalidation_interval_never_exceeds_fifteen_seconds() { + assert_eq!(membership_revalidate_interval_secs(None), 10); + assert_eq!(membership_revalidate_interval_secs(Some("0")), 1); + assert_eq!(membership_revalidate_interval_secs(Some("15")), 15); + assert_eq!(membership_revalidate_interval_secs(Some("300")), 15); + } + #[test] fn test_emission_scope_off_disallows_every_community() { assert!(EmissionScope::All.allows(&Uuid::new_v4())); diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 3a6ca49282b..cc5eef733e2 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -54,6 +54,10 @@ struct ConnEntry { backpressure_count: Arc, subscriptions: ConnectionSubscriptions, authenticated_pubkey: Arc>>>, + /// Relay-member key that authorizes this connection. This is the + /// authenticated key for direct members and the verified owner key for + /// NIP-OA virtual agents. + membership_pubkey: Arc>>>, grace_limit: u8, } @@ -178,6 +182,33 @@ where (closed, failures) } +async fn revalidate_membership_principals( + manager: &ConnectionManager, + mut check_member: Check, +) -> (usize, Vec<(CommunityId, Vec, buzz_db::DbError)>) +where + Check: FnMut(CommunityId, Vec) -> CheckFuture, + CheckFuture: Future>, +{ + let mut closed = 0; + let mut failures = Vec::new(); + for (community_id, membership_pubkey) in manager.live_membership_principals() { + match check_member(community_id, membership_pubkey.clone()).await { + Ok(false) => { + closed += manager.disconnect_membership_principal( + community_id, + &membership_pubkey, + "", + "blocked: relay membership removed", + ); + } + Ok(true) => {} + Err(error) => failures.push((community_id, membership_pubkey, error)), + } + } + (closed, failures) +} + /// Tracks active Nostr WebSocket connections and provides message routing by connection ID. pub struct ConnectionManager { connections: DashMap, @@ -218,6 +249,7 @@ impl ConnectionManager { backpressure_count, subscriptions, authenticated_pubkey: Arc::new(std::sync::RwLock::new(None)), + membership_pubkey: Arc::new(std::sync::RwLock::new(None)), grace_limit, }, ); @@ -230,11 +262,80 @@ impl ConnectionManager { /// Record the authenticated pubkey for a connection after NIP-42 succeeds. pub fn set_authenticated_pubkey(&self, conn_id: Uuid, pubkey_bytes: Vec) { + self.set_authenticated_membership(conn_id, pubkey_bytes.clone(), pubkey_bytes); + } + + /// Record the authenticated key and the relay-member principal that + /// authorizes it. For NIP-OA agents these keys intentionally differ. + pub fn set_authenticated_membership( + &self, + conn_id: Uuid, + pubkey_bytes: Vec, + membership_pubkey_bytes: Vec, + ) { if let Some(entry) = self.connections.get(&conn_id) { if let Ok(mut slot) = entry.authenticated_pubkey.write() { *slot = Some(pubkey_bytes); } + if let Ok(mut slot) = entry.membership_pubkey.write() { + *slot = Some(membership_pubkey_bytes); + } + } + } + + /// Return the distinct live `(community, relay-member principal)` pairs. + pub fn live_membership_principals(&self) -> HashSet<(CommunityId, Vec)> { + self.connections + .iter() + .filter_map(|entry| { + entry + .membership_pubkey + .read() + .ok()? + .clone() + .map(|pubkey| (entry.community_id, pubkey)) + }) + .collect() + } + + /// Disconnect every connection whose authorization derives from the given + /// relay-member principal in one community. + pub fn disconnect_membership_principal( + &self, + community: CommunityId, + membership_pubkey: &[u8], + event_id: &str, + reason: &str, + ) -> usize { + let frame = crate::protocol::RelayMessage::ok(event_id, false, reason); + let conn_ids: Vec = self + .connections + .iter() + .filter_map(|entry| { + let matches = entry.community_id == community + && entry + .membership_pubkey + .read() + .ok() + .and_then(|stored| { + stored + .as_ref() + .map(|stored| stored.as_slice() == membership_pubkey) + }) + .unwrap_or(false); + matches.then_some(*entry.key()) + }) + .collect(); + + for conn_id in &conn_ids { + if let Some(entry) = self.connections.get(conn_id) { + let _ = entry + .ctrl_tx + .try_send(WsMessage::Text(frame.clone().into())); + entry.cancel.cancel(); + } } + conn_ids.len() } /// Return live connection IDs authenticated as `pubkey_bytes` in one community. @@ -1031,6 +1132,31 @@ impl AppState { closed } + /// Revalidate the durable relay-member principals behind live sockets. + /// + /// Redis disconnect delivery is best-effort. This bounded local scan is + /// the durable backstop that closes idle sessions even when a pod missed + /// the removal command. + pub async fn revalidate_live_memberships(&self) -> usize { + if !self.config.require_relay_membership { + return 0; + } + + let (closed, failures) = + revalidate_membership_principals(&self.conn_manager, |community_id, pubkey| { + let db = self.db.clone(); + async move { + let membership_hex = hex::encode(pubkey); + db.is_relay_member(community_id, &membership_hex).await + } + }) + .await; + for (community_id, _, error) in failures { + tracing::warn!(%community_id, %error, "relay membership revalidation failed; retaining its sockets until next tick"); + } + closed + } + /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss. pub async fn get_accessible_channel_ids_cached( &self, @@ -1250,6 +1376,106 @@ mod tests { ); } + #[test] + fn membership_principal_disconnect_covers_delegated_agents_and_preserves_tenant_fence() { + let mgr = ConnectionManager::new(); + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb)); + let owner = vec![1u8; 32]; + let agent = vec![2u8; 32]; + let other = vec![3u8; 32]; + + let register = |community, authenticated, principal| { + let conn_id = Uuid::new_v4(); + let (tx, _rx) = mpsc::channel(8); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); + let cancel = CancellationToken::new(); + mgr.register( + conn_id, + tx, + ctrl_tx, + cancel.clone(), + community, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + mgr.set_authenticated_membership(conn_id, authenticated, principal); + cancel + }; + + let owner_a = register(community_a, owner.clone(), owner.clone()); + let agent_a = register(community_a, agent, owner.clone()); + let other_a = register(community_a, other, vec![3u8; 32]); + let owner_b = register(community_b, owner.clone(), owner.clone()); + + assert_eq!( + mgr.disconnect_membership_principal( + community_a, + &owner, + "", + "blocked: relay membership removed" + ), + 2 + ); + assert!(owner_a.is_cancelled()); + assert!(agent_a.is_cancelled()); + assert!(!other_a.is_cancelled()); + assert!(!owner_b.is_cancelled()); + } + + #[tokio::test] + async fn durable_membership_revalidation_closes_missed_disconnects() { + let mgr = ConnectionManager::new(); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let removed = vec![1u8; 32]; + let retained = vec![2u8; 32]; + let failed = vec![3u8; 32]; + + let register = |authenticated, principal| { + let conn_id = Uuid::new_v4(); + let (tx, _rx) = mpsc::channel(8); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); + let cancel = CancellationToken::new(); + mgr.register( + conn_id, + tx, + ctrl_tx, + cancel.clone(), + community, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + mgr.set_authenticated_membership(conn_id, authenticated, principal); + cancel + }; + + let removed_cancel = register(vec![11u8; 32], removed.clone()); + let retained_cancel = register(retained.clone(), retained.clone()); + let failed_cancel = register(failed.clone(), failed.clone()); + + let (closed, failures) = revalidate_membership_principals(&mgr, |_, principal| { + let removed = removed.clone(); + let failed = failed.clone(); + async move { + if principal == failed { + Err(buzz_db::DbError::InvalidData("injected failure".into())) + } else { + Ok(principal != removed) + } + } + }) + .await; + + assert_eq!(closed, 1); + assert!(removed_cancel.is_cancelled()); + assert!(!retained_cancel.is_cancelled()); + assert!(!failed_cancel.is_cancelled()); + assert_eq!(failures.len(), 1); + assert_eq!(failures[0].1, failed); + } + #[test] fn send_to_increments_grace_counter_on_full() { // Buffer size 1 — fill it, then the next send is Full.