diff --git a/AGENTS.md b/AGENTS.md index 5932bcd9e1..bd721075d8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -313,7 +313,9 @@ Filters and queries must scope to `h` tags when operating within a channel. This applies to events *inside* a channel. Addressable events that describe a channel carry its id in their `d` tag instead: kind:39000 (metadata), kind:39001, kind:39002 (membership). `get_channels` resolves a user's channels -from the `d` tag of their kind:39002 events, not from `h`. +from the `d` tag of their kind:39002 events, not from `h`. The relay leaves +system channels out of these reads unless the filter asks for them (see +`docs/nips/NIP-CL.md`). **Agent-facing operations go in `buzz-cli`**: New agent-facing features belong in `buzz-cli` — add a subcommand there first, then wire the REST/WebSocket call in `client.rs`. `buzz-dev-mcp` (shell + file tools for `buzz-agent`) is separate. diff --git a/Justfile b/Justfile index a41427d46c..fdc48f7261 100644 --- a/Justfile +++ b/Justfile @@ -531,7 +531,7 @@ test-unit: # are infra-free; config parses env into a struct and never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index f1b0af4362..90aaf69272 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -255,6 +255,48 @@ async fn lifecycle_cas_queries_move_redaction_and_retention() { f.db.validate_deletion_serving_catalog().await.unwrap(); } +/// An artifact in a system channel is read only by a query whose `#h` names +/// that channel (NIP-CL), even when the channel is open. +/// Mutation: drop the `channel_type` condition from `query_artifacts` → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn system_channel_artifacts_are_read_only_by_name() { + let f = Fixture::new().await; + let system = Uuid::new_v4(); + sqlx::query("INSERT INTO channels(community_id,id,name,visibility,channel_type,created_by) VALUES($1,$2,$3,'open','system',$4)") + .bind(f.community.as_uuid()) + .bind(system) + .bind(system.to_string()) + .bind(f.owner.public_key().to_bytes().as_slice()) + .execute(&f.db.pool) + .await + .unwrap(); + for home in [f.a, system] { + let create = f.revision(Uuid::new_v4(), "create", home, None, &f.owner, vec![]); + assert!(matches!( + f.accept(&create, None).await, + ArtifactOutcome::Accepted(_) + )); + } + for reader in [&f.owner, &f.peer] { + let (events, _) = f + .query(reader, serde_json::json!({"artifact":"current"}), false) + .await; + let homes: Vec<_> = events.iter().map(|e| e.channel_id).collect(); + assert_eq!(homes, [Some(f.a)], "no #h"); + assert_eq!( + f.count(reader, serde_json::json!({"artifact":"current"})) + .await, + 1 + ); + let named = serde_json::json!({"artifact":"current","#h":[system.to_string()]}); + let (events, _) = f.query(reader, named.clone(), false).await; + let homes: Vec<_> = events.iter().map(|e| e.channel_id).collect(); + assert_eq!(homes, [Some(system)], "#h names the system channel"); + assert_eq!(f.count(reader, named).await, 1); + } +} + #[tokio::test] #[ignore = "requires Postgres"] async fn repeated_moves_get_distinct_source_safe_markers() { diff --git a/crates/buzz-db/src/store/artifact_query.rs b/crates/buzz-db/src/store/artifact_query.rs index 5d2390f76f..fcc4669575 100644 --- a/crates/buzz-db/src/store/artifact_query.rs +++ b/crates/buzz-db/src/store/artifact_query.rs @@ -36,9 +36,19 @@ impl Db { if query.view == ArtifactView::Current { qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); } + // A system channel is read only by a query whose `#h` names it + // (NIP-CL); malformed `#h` values name no channel. + let named_channels: Vec = query + .tags + .iter() + .filter(|(name, _)| name == "h") + .flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok())) + .collect(); // The explicit kind lets the community/kind index exclude chat before // the join, even for an unfiltered current-state page. - qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (c.visibility='open' OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") + qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (c.channel_type<>'system' OR c.id=ANY(") + .push_bind(named_channels) + .push(")) AND (c.visibility='open' OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") .push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))"); for (name, values) in &query.tags { // GIN-usable necessary prefilter; retain the positional recheck diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index fd3041f969..f178ab5814 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -1117,6 +1117,10 @@ pub async fn get_members_bulk( /// /// Includes channels where the pubkey is an active member AND all open channels. /// Open channels must be included in REQ filter resolution. +/// +/// System channels are left out of both halves: this is the implicit scope of +/// queries that do not name a channel. A query that names a system channel, +/// or opts in to them, adds them from [`get_readable_system_channel_ids`]. pub async fn get_accessible_channel_ids( pool: &PgPool, community_id: CommunityId, @@ -1133,10 +1137,12 @@ pub async fn get_accessible_channel_ids( FROM channel_members cm JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL WHERE cm.community_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL + AND c.channel_type <> 'system' UNION SELECT id AS channel_id FROM channels WHERE community_id = $1 AND visibility = 'open' AND deleted_at IS NULL + AND channel_type <> 'system' "#, ) .bind(community_id.as_uuid()) @@ -1152,6 +1158,72 @@ pub async fn get_accessible_channel_ids( .collect() } +/// Get the system channels a pubkey can read: those where it is an active +/// member, and every open system channel. Uncached, so a caller can trust it +/// for a channel the request names. +pub async fn get_readable_system_channel_ids( + pool: &PgPool, + community_id: CommunityId, + pubkey: &[u8], +) -> Result> { + let mut connection = crate::observability::acquire_writer( + pool, + crate::observability::WriterOperation::Authorization, + ) + .await?; + let rows = sqlx::query( + r#" + SELECT c.id AS channel_id + FROM channels c + WHERE c.community_id = $1 AND c.deleted_at IS NULL AND c.channel_type = 'system' + AND ( + c.visibility = 'open' + OR EXISTS ( + SELECT 1 FROM channel_members cm + WHERE cm.community_id = c.community_id AND cm.channel_id = c.id + AND cm.pubkey = $2 AND cm.removed_at IS NULL + ) + ) + "#, + ) + .bind(community_id.as_uuid()) + .bind(pubkey) + .fetch_all(&mut *connection) + .await?; + + rows.into_iter() + .map(|r| { + let id: Uuid = r.try_get("channel_id")?; + Ok(id) + }) + .collect() +} + +/// The IDs in `channel_ids` that name system channels, deleted or not. +/// Push matching and membership notifications use it to skip them. +pub async fn system_channel_ids_among( + pool: &PgPool, + community_id: CommunityId, + channel_ids: &[Uuid], +) -> Result> { + if channel_ids.is_empty() { + return Ok(Vec::new()); + } + let mut connection = crate::observability::acquire_writer( + pool, + crate::observability::WriterOperation::Authorization, + ) + .await?; + Ok(sqlx::query_scalar( + "SELECT id FROM channels \ + WHERE community_id = $1 AND id = ANY($2) AND channel_type = 'system'", + ) + .bind(community_id.as_uuid()) + .bind(channel_ids) + .fetch_all(&mut *connection) + .await?) +} + /// A large channel whose canonical active-member count may need its legacy /// discovery snapshot repaired. #[derive(Debug, Clone, PartialEq, Eq)] @@ -1377,6 +1449,7 @@ pub async fn get_accessible_channels( WHERE c.community_id = $1 AND c.deleted_at IS NULL {membership_clause} AND (c.channel_type != 'dm' OR cm.hidden_at IS NULL) + AND c.channel_type <> 'system' "# ); @@ -1782,6 +1855,26 @@ impl Db { get_accessible_channel_ids(&self.pool, community_id, pubkey).await } + /// Get the system channels a pubkey can read (member, or open). + #[datastore_span(name = "get_readable_system_channel_ids", system = "postgresql")] + pub async fn get_readable_system_channel_ids( + &self, + community_id: CommunityId, + pubkey: &[u8], + ) -> Result> { + get_readable_system_channel_ids(&self.pool, community_id, pubkey).await + } + + /// The IDs in `channel_ids` that name system channels, deleted or not. + #[datastore_span(name = "system_channel_ids_among", system = "postgresql")] + pub async fn system_channel_ids_among( + &self, + community_id: CommunityId, + channel_ids: &[Uuid], + ) -> Result> { + system_channel_ids_among(&self.pool, community_id, channel_ids).await + } + /// Returns large active-channel rosters whose relay-authored snapshots differ. #[datastore_span( name = "list_large_channel_rosters_needing_reconciliation", diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index 38e6e02600..873b41a71a 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -99,6 +99,8 @@ impl Db { WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL AND c.deleted_at IS NULL AND ($3::uuid IS NULL OR c.id>$3) AND ($9::uuid[] IS NULL OR c.id=ANY($9)) + -- System channels only when the request names them. + AND ($9::uuid[] IS NOT NULL OR c.channel_type<>'system') ORDER BY c.id LIMIT $4 ) SELECT r.*, COALESCE(e.latest_id, latest.latest_id) AS latest_id, diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 64efc336de..685d275c42 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1438,12 +1438,26 @@ async fn query_events_authed( .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) .await .map_err(|e| internal_error(&format!("channel access lookup: {e}")))?; + // System channels are outside `accessible_channels`; a filter that names + // one reads it in its own scope only (see `handlers::system_channels`). + let mut readable_system = + crate::handlers::system_channels::readable_system_channels_for_filters( + state, + tenant.community(), + &pubkey_bytes, + &filters, + &accessible_channels, + None, + ) + .await + .map_err(|e| internal_error(&format!("system channel access lookup: {e}")))?; repair_requested_channel_access( state, tenant, &filters, &pubkey_bytes, &mut accessible_channels, + &mut readable_system, ) .await?; @@ -1459,6 +1473,7 @@ async fn query_events_authed( &raw_filters, &filters, &accessible_channels, + &readable_system, tenant, &authed_pubkey_hex, &pubkey_bytes, @@ -1490,7 +1505,12 @@ async fn query_events_authed( // Resolve reply owners from retained thread metadata, including tombstones. for (idx, (raw, filter)) in raw_filters.iter().zip(filters.iter()).enumerate() { if extension_flag(raw, "resolve_thread_roots") { - events.extend(thread_roots::query(state, tenant, filter, &accessible_channels).await?); + let scope = crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); + events.extend(thread_roots::query(state, tenant, filter, &scope).await?); handled.insert(idx); } } @@ -1506,7 +1526,11 @@ async fn query_events_authed( tenant, raw, filter, - &accessible_channels, + &crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ), &pubkey_bytes, &mut events, ) @@ -1628,6 +1652,11 @@ async fn query_events_authed( Ok(b) if b.len() == 32 => b, _ => continue, }; + let accessible_channels = crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); if let Some(ch_id) = extract_channel_from_filter(filter) { if !accessible_channels.contains(&ch_id) { @@ -1717,6 +1746,11 @@ async fn query_events_authed( if handled.contains(&idx) { continue; } + let accessible_channels = crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); if let Some(ch_id) = extract_channel_from_filter(filter) { if !accessible_channels.contains(&ch_id) { @@ -1818,6 +1852,11 @@ async fn query_events_authed( // Phase 3 — post-processing, strictly in filter order. while let Some((idx, filter_events)) = catchall_results.next().await { let filter = &filters[idx]; + let accessible_channels = crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); match filter_events { Ok(stored_events) => { for se in stored_events { @@ -1849,13 +1888,15 @@ async fn query_events_authed( Ok(Json(Value::Array(events))) } -async fn repair_requested_channel_access( +pub(crate) async fn repair_requested_channel_access( state: &AppState, tenant: &TenantContext, filters: &[nostr::Filter], pubkey_bytes: &[u8], accessible_channels: &mut Vec, + readable_system: &mut Vec, ) -> Result<(), (StatusCode, Json)> { + let repaired_from = accessible_channels.len(); for filter in filters { let Some(requested) = crate::handlers::req::extract_channel_ids_from_filters(std::slice::from_ref(filter)) @@ -1863,7 +1904,8 @@ async fn repair_requested_channel_access( continue; }; for channel_id in requested { - if accessible_channels.contains(&channel_id) { + // A readable system channel stays out of the request-wide set. + if accessible_channels.contains(&channel_id) || readable_system.contains(&channel_id) { continue; } let is_member = state @@ -1879,7 +1921,15 @@ async fn repair_requested_channel_access( ); } } - Ok(()) + crate::handlers::system_channels::keep_repaired_system_channels_named( + state, + tenant.community(), + accessible_channels, + repaired_from, + readable_system, + ) + .await + .map_err(|e| internal_error(&format!("system channel access lookup: {e}"))) } /// Count events via HTTP bridge (NIP-98 auth). Returns `{"count": N}`. @@ -2035,17 +2085,36 @@ async fn count_events_authed( .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) .await .map_err(|e| internal_error(&format!("channel access lookup: {e}")))?; + // System channels are outside `accessible_channels`; a filter that names + // one reads it in its own scope only (see `handlers::system_channels`). + let mut readable_system = + crate::handlers::system_channels::readable_system_channels_for_filters( + state, + tenant.community(), + &pubkey_bytes, + &filters, + &accessible_channels, + None, + ) + .await + .map_err(|e| internal_error(&format!("system channel access lookup: {e}")))?; repair_requested_channel_access( state, tenant, &filters, &pubkey_bytes, &mut accessible_channels, + &mut readable_system, ) .await?; let mut total: u64 = 0; for (raw, filter) in raw.iter().zip(&filters) { + let accessible_channels = crate::handlers::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); let needs_author_only_filtering = crate::handlers::req::filter_can_match_author_only_kinds(filter); // Same result-gated guard as the WS COUNT handler: force the per-event @@ -2270,26 +2339,17 @@ fn search_hit_accepted( /// Handle search filters by routing to Postgres FTS, then fetching full events /// from DB. Supports a bridge-only `page` extension over the FTS result set. +#[allow(clippy::too_many_arguments)] async fn handle_bridge_search( state: &AppState, raw_filters: &[Value], filters: &[nostr::Filter], - accessible_channels: &[uuid::Uuid], + implicit_channels: &[uuid::Uuid], + readable_system: &[uuid::Uuid], tenant: &buzz_core::tenant::TenantContext, reader_pubkey_hex: &str, pubkey_bytes: &[u8], ) -> Result, (StatusCode, Json)> { - // Bridge always includes global (channel-less) events — same as WS with - // full scopes. `None` means no accessible channels and no global access → - // empty result set (the caller short-circuits exactly as the WS door EOSEs). - let channel_scope = match crate::handlers::req::build_search_channel_scope_filter( - accessible_channels, - true, // include_global - ) { - Some(scope) => scope, - None => return Ok(Json(Value::Array(Vec::new()))), - }; - let mut events: Vec = Vec::new(); let mut seen_ids: std::collections::HashSet<[u8; 32]> = std::collections::HashSet::new(); @@ -2305,6 +2365,13 @@ async fn handle_bridge_search( if limit == 0 { continue; } + // Implicit channels plus the readable system channels this filter + // names; never another filter's. + let accessible_channels = &*crate::handlers::system_channels::filter_channel_scope( + filter, + implicit_channels, + readable_system, + ); // Scope by channel — push the #h tag (intersected with accessible // channels) if present, else the community-wide scope. @@ -2321,7 +2388,15 @@ async fn handle_bridge_search( } buzz_search::ChannelScope::Channels(valid) } else { - channel_scope.clone() + // Bridge always includes global (channel-less) events — same + // as WS with full scopes. + match crate::handlers::req::build_search_channel_scope_filter( + accessible_channels, + true, // include_global + ) { + Some(scope) => scope, + None => continue, + } }; let kinds = filter.kinds.as_ref().and_then(|ks| { diff --git a/crates/buzz-relay/src/api/bridge/thread_window.rs b/crates/buzz-relay/src/api/bridge/thread_window.rs index 793a541c0d..c1a84d6002 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window.rs +++ b/crates/buzz-relay/src/api/bridge/thread_window.rs @@ -84,6 +84,40 @@ fn append(events: &mut Vec, budget: &mut Budget, event: &nostr::Event) -> Ok(()) } +/// The reader's implicit channel set and, when a request names a channel +/// outside it, the system channels it can read. A thread window names its +/// channel, so a readable system channel joins that request's scope only. +async fn access_sets( + state: &AppState, + tenant: &TenantContext, + reader: &nostr::PublicKey, + requests: &[&Request], + context: &str, +) -> Result<(Vec, Vec), Error> { + let implicit = state + .db + .get_accessible_channel_ids(tenant.community(), &reader.to_bytes()) + .await + .map_err(|e| database_error(context, e))?; + let system = if requests + .iter() + .any(|request| !implicit.contains(&request.channel)) + { + state + .db + .get_readable_system_channel_ids(tenant.community(), &reader.to_bytes()) + .await + .map_err(|e| database_error(context, e))? + } else { + Vec::new() + }; + Ok((implicit, system)) +} + +fn same_set(a: &[uuid::Uuid], b: &[uuid::Uuid]) -> bool { + a.iter().collect::>() == b.iter().collect::>() +} + /// Authorize the entire batch against one writer access set, then refresh it /// once before releasing any output. A later window must never suppress only /// its own rows while releasing an earlier window built before revocation. @@ -93,26 +127,28 @@ pub(super) async fn query_batch<'a>( reader: &nostr::PublicKey, requests: impl IntoIterator, ) -> Result, Error> { - let accessible = state - .db - .get_accessible_channel_ids(tenant.community(), &reader.to_bytes()) - .await - .map_err(|e| database_error("access", e))?; + let requests: Vec<&Request> = requests.into_iter().collect(); + let (implicit, system) = access_sets(state, tenant, reader, &requests, "access").await?; let mut budget = Budget::default(); let mut events = Vec::new(); - for request in requests { - if accessible.contains(&request.channel) { - events.extend(query(state, tenant, reader, request, &accessible, &mut budget).await?); - } + for request in &requests { + let accessible: std::borrow::Cow<'_, [uuid::Uuid]> = if implicit.contains(&request.channel) + { + std::borrow::Cow::Borrowed(&implicit) + } else if system.contains(&request.channel) { + let mut scope = implicit.clone(); + scope.push(request.channel); + std::borrow::Cow::Owned(scope) + } else { + continue; + }; + events.extend(query(state, tenant, reader, request, &accessible, &mut budget).await?); } - let current = state - .db - .get_accessible_channel_ids(tenant.community(), &reader.to_bytes()) - .await - .map_err(|e| database_error("final access", e))?; + let (current_implicit, current_system) = + access_sets(state, tenant, reader, &requests, "final access").await?; // Grants can expose auxiliary events omitted from the original closure; // revocations can invalidate earlier windows or their cross-channel aux. - if accessible.iter().collect::>() != current.iter().collect::>() { + if !same_set(&implicit, ¤t_implicit) || !same_set(&system, ¤t_system) { return Err(unavailable("thread authorization changed; retry query")); } Ok(events) diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index b6a1f98e00..16685f3041 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -121,7 +121,18 @@ async fn authorize_workflow_read( .map_err(|error| { internal_error(&format!("workflow channel access lookup: {error}")).into_response() })?; - if !accessible.contains(&channel_id) { + // A workflow names its channel, so a readable system channel counts. + let readable = accessible.contains(&channel_id) + || state + .db + .get_readable_system_channel_ids(tenant.community(), &pubkey_bytes) + .await + .map_err(|error| { + internal_error(&format!("workflow system channel access lookup: {error}")) + .into_response() + })? + .contains(&channel_id); + if !readable { return Err(not_found()); } diff --git a/crates/buzz-relay/src/handlers/count.rs b/crates/buzz-relay/src/handlers/count.rs index 4dca0beb7f..5d2ba44e95 100644 --- a/crates/buzz-relay/src/handlers/count.rs +++ b/crates/buzz-relay/src/handlers/count.rs @@ -99,6 +99,25 @@ pub async fn handle_count( if let Some(allowed) = token_channel_ids.as_deref() { accessible_channels.retain(|channel_id| allowed.contains(channel_id)); } + // System channels are outside `accessible_channels`; a filter that names + // one counts it in its own scope only, as in the WS REQ handler. + let mut readable_system = match super::system_channels::readable_system_channels_for_filters( + &state, + conn.tenant.community(), + &pubkey_bytes, + &filters, + &accessible_channels, + token_channel_ids.as_deref(), + ) + .await + { + Ok(ids) => ids, + Err(e) => { + warn!(sub_id = %sub_id, "Failed to get readable system channels: {e}"); + conn.send(RelayMessage::closed(&sub_id, "error: database error")); + return; + } + }; // B2: acquire effect permit immediately before the first DB count query. // The permit is held through all count queries and the COUNT response. @@ -144,8 +163,9 @@ pub async fn handle_count( && !result_gated_count_safe_for_pushdown(filter, &authed_pubkey_hex); if let Some(requested_channels) = requested_channels { + let repaired_from = accessible_channels.len(); for &ch_id in &requested_channels { - if accessible_channels.contains(&ch_id) { + if accessible_channels.contains(&ch_id) || readable_system.contains(&ch_id) { continue; } let token_allows = token_channel_ids @@ -174,10 +194,28 @@ pub async fn handle_count( db_is_member, ); } + if let Err(e) = super::system_channels::keep_repaired_system_channels_named( + &state, + conn.tenant.community(), + &mut accessible_channels, + repaired_from, + &mut readable_system, + ) + .await + { + warn!(sub_id = %sub_id, "Failed to classify repaired channels: {e}"); + conn.send(RelayMessage::closed(&sub_id, "error: database error")); + return; + } + let filter_scope = super::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); let authorized_requested: Vec<_> = requested_channels .iter() .copied() - .filter(|channel_id| accessible_channels.contains(channel_id)) + .filter(|channel_id| filter_scope.contains(channel_id)) .collect(); if authorized_requested.is_empty() { continue; @@ -195,12 +233,7 @@ pub async fn handle_count( conn.tenant.community(), ) .await; - super::req::apply_channel_scope_to_query( - &mut query, - filter, - ch_id, - &accessible_channels, - ); + super::req::apply_channel_scope_to_query(&mut query, filter, ch_id, &filter_scope); // Shared-gated visibility pushdown: pre-filter the fallback // query_events candidate page before ORDER/LIMIT. if needs_shared_gate_filtering { @@ -279,7 +312,14 @@ pub async fn handle_count( conn.tenant.community(), ) .await; - query.channel_ids = Some(accessible_channels.to_vec()); + query.channel_ids = Some( + super::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ) + .into_owned(), + ); // Shared-gated visibility pushdown for the fallback query_events path. if needs_shared_gate_filtering { query.shared_gated_reader = Some(pubkey_bytes.clone()); diff --git a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs index ed4d862b0a..ea252c10b1 100644 --- a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs @@ -240,7 +240,7 @@ async fn creator_tag_survives_ownership_transfer() { } /// Ingest a kind:9002 that `keys` signs for `channel_id` with `tags`. -async fn edit_channel( +pub(crate) async fn edit_channel( state: &Arc, tenant: &TenantContext, keys: &Keys, @@ -607,15 +607,16 @@ async fn racing_creates_resolve_to_the_same_oldest_channel() { assert_eq!(pick(vec![&events[1], &events[0]]), Some(expected)); } -/// Run one historical REQ as `reader` and return the IDs of the events sent -/// before EOSE. -async fn req_event_ids( - state: &Arc, +/// An authenticated WebSocket connection for `reader`, and the receiver of +/// the frames the relay sends on it. +pub(crate) fn test_conn( tenant: &TenantContext, reader: &Keys, - filter: serde_json::Value, -) -> Vec { - let (send_tx, mut send_rx) = tokio::sync::mpsc::channel(256); +) -> ( + Arc, + tokio::sync::mpsc::Receiver, +) { + let (send_tx, send_rx) = tokio::sync::mpsc::channel(256); let (ctrl_tx, _ctrl_rx) = tokio::sync::mpsc::channel(4); let cancel = tokio_util::sync::CancellationToken::new(); let conn = Arc::new(crate::connection::ConnectionState { @@ -643,6 +644,18 @@ async fn req_event_ids( nip_fi_gate: crate::nip_fi_gate::SessionAdmissionGate::off_mode(cancel.clone()), community_control: crate::state::CommunityConnectionControl::new(cancel.clone()), }); + (conn, send_rx) +} + +/// Run one historical REQ as `reader` and return the IDs of the events sent +/// before EOSE. +pub(crate) async fn req_event_ids( + state: &Arc, + tenant: &TenantContext, + reader: &Keys, + filter: serde_json::Value, +) -> Vec { + let (conn, mut send_rx) = test_conn(tenant, reader); let filter: nostr::Filter = serde_json::from_value(filter).expect("filter"); crate::handlers::req::handle_req("g".into(), vec![filter], vec![None], conn, state.clone()) .await; @@ -662,7 +675,11 @@ async fn req_event_ids( } /// The current kind:39000 event ID for `channel_id`. -async fn metadata_event_id(state: &AppState, tenant: &TenantContext, channel_id: Uuid) -> String { +pub(crate) async fn metadata_event_id( + state: &AppState, + tenant: &TenantContext, + channel_id: Uuid, +) -> String { state .db .query_events(&EventQuery { diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 8d5175aa45..24c368a564 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -42,6 +42,10 @@ pub mod report_resolution; pub mod req; /// NIP-29 and NIP-25 side-effect handlers. pub mod side_effects; +#[cfg(test)] +mod system_channel_postgres_tests; +/// Per-filter query scope for system channels. +pub(crate) mod system_channels; /// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag, /// applying the server-side override when configured. diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 656cb23b0f..19c6173804 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -126,6 +126,25 @@ pub async fn handle_req( if let Some(allowed) = token_channel_ids.as_deref() { accessible_channels.retain(|channel_id| allowed.contains(channel_id)); } + // System channels are outside `accessible_channels`. A filter that names + // one adds it to its own scope only (see `system_channels`). + let mut readable_system = match super::system_channels::readable_system_channels_for_filters( + &state, + conn.tenant.community(), + &pubkey_bytes, + &filters, + &accessible_channels, + token_channel_ids.as_deref(), + ) + .await + { + Ok(ids) => ids, + Err(e) => { + warn!(conn_id = %conn_id, "Failed to get readable system channels: {e}"); + conn.send(RelayMessage::closed(&sub_id, "error: database error")); + return; + } + }; // Build the conformance `AbstractState` once at request entry. The // `Option` only goes `None` on malformed pubkey bytes (already a @@ -146,11 +165,17 @@ pub async fn handle_req( // is what fixes the search false-miss: a `#h=` search would // otherwise be scoped against the stale vector and return empty. if let Some(requested) = requested_channel_ids.as_ref() { + let repaired_from = accessible_channels.len(); for &ch_id in requested { let token_allows = token_channel_ids .as_deref() .is_none_or(|allowed| allowed.contains(&ch_id)); - let db_is_member = if !token_allows || accessible_channels.contains(&ch_id) { + // A readable system channel is authorized for the filters that + // name it; it must not enter the request-wide vector. + let db_is_member = if !token_allows + || accessible_channels.contains(&ch_id) + || readable_system.contains(&ch_id) + { None } else { match state @@ -185,13 +210,28 @@ pub async fn handle_req( db_is_member, ); } + if let Err(e) = super::system_channels::keep_repaired_system_channels_named( + &state, + conn.tenant.community(), + &mut accessible_channels, + repaired_from, + &mut readable_system, + ) + .await + { + warn!(conn_id = %conn_id, "Failed to classify repaired channels: {e}"); + conn.send(RelayMessage::closed(&sub_id, "error: database error")); + return; + } } let authorized_requested_channels = requested_channel_ids.as_ref().map(|requested| { requested .iter() .copied() - .filter(|channel_id| accessible_channels.contains(channel_id)) + .filter(|channel_id| { + accessible_channels.contains(channel_id) || readable_system.contains(channel_id) + }) .collect::>() }); // Partial authorization preserves NIP-01 OR semantics by omitting only @@ -318,6 +358,7 @@ pub async fn handle_req( owner, &filters, &accessible_channels, + &readable_system, token_channel_ids.is_none(), &conn.tenant, &pubkey_bytes, @@ -416,7 +457,11 @@ pub async fn handle_req( &mut params, filter, per_filter_channel, - &accessible_channels, + &super::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ), ); // Shared-gated visibility pushdown: set reader bytes so query_events // appends the SQL visibility clause before ORDER/LIMIT, preventing @@ -448,6 +493,11 @@ pub async fn handle_req( // Phase 3 — post-processing, strictly in filter order. while let Some((idx, per_filter_channel, filter_events)) = results.next().await { let filter = &filters[idx]; + let filter_scope = super::system_channels::filter_channel_scope( + filter, + &accessible_channels, + &readable_system, + ); let events = match filter_events { Ok(evs) => evs, Err(e) => { @@ -510,7 +560,7 @@ pub async fn handle_req( } if let Some(ch_id) = stored.channel_id { - if !accessible_channels.contains(&ch_id) { + if !filter_scope.contains(&ch_id) { continue; } } @@ -684,7 +734,8 @@ async fn handle_search_req( sub_id: &str, owner: u64, filters: &[Filter], - accessible_channels: &[uuid::Uuid], + implicit_channels: &[uuid::Uuid], + readable_system: &[uuid::Uuid], include_global: bool, tenant: &TenantContext, reader_pubkey_bytes: &[u8], @@ -692,17 +743,15 @@ async fn handle_search_req( state: &AppState, trace_state: Option<&crate::conformance::AbstractState>, ) { - // The community-wide channel scope (no #h tag on the filter). `None` means - // "no accessible channels and no global access" → EOSE, exactly as the - // legacy string-filter helper short-circuited. - let all_channels_scope = - match build_search_channel_scope_filter(accessible_channels, include_global) { - Some(scope) => scope, - None => { - conn.send(RelayMessage::eose(sub_id)); - return; - } - }; + // `None` means "no accessible channels and no global access" → EOSE, + // exactly as the legacy string-filter helper short-circuited. A filter + // that names a readable system channel can still match it. + if build_search_channel_scope_filter(implicit_channels, include_global).is_none() + && readable_system.is_empty() + { + conn.send(RelayMessage::eose(sub_id)); + return; + } let mut seen_ids: HashSet = HashSet::new(); @@ -711,6 +760,13 @@ async fn handle_search_req( Some(s) if !s.is_empty() => s.clone(), _ => continue, }; + // Implicit channels plus the readable system channels this filter + // names; never another filter's. + let accessible_channels = &*super::system_channels::filter_channel_scope( + filter, + implicit_channels, + readable_system, + ); let limit = filter .limit @@ -742,7 +798,11 @@ async fn handle_search_req( } buzz_search::ChannelScope::Channels(valid) } else { - all_channels_scope.clone() + // The community-wide scope (no #h tag on the filter). + match build_search_channel_scope_filter(accessible_channels, include_global) { + Some(scope) => scope, + None => continue, + } }; let kinds = filter.kinds.as_ref().and_then(|ks| { @@ -1164,7 +1224,7 @@ fn filter_to_query_params( .filter(|values| !values.is_empty()) .map(|values| values.iter().map(|v| v.to_string()).collect()); - let tag_values = group_state_tag_values(filter, kinds.as_deref()); + let tag_values = group_state_tag_values(filter); EventQuery { channel_id, @@ -1195,16 +1255,8 @@ fn filter_to_query_params( /// The SQL match is exact NIP-01 tag matching (see /// [`EventQuery::tag_values`]), so no false candidate can take a place in /// the page. The caller's NIP-01 post-filter still runs on each event. -fn group_state_tag_values(filter: &Filter, kinds: Option<&[i32]>) -> Vec<(String, Vec)> { - let group_state_only = kinds.is_some_and(|ks| { - !ks.is_empty() - && ks.iter().all(|&k| { - (buzz_core::kind::KIND_NIP29_GROUP_METADATA as i32 - ..=buzz_core::kind::KIND_NIP29_GROUP_ROLES as i32) - .contains(&k) - }) - }); - if !group_state_only { +fn group_state_tag_values(filter: &Filter) -> Vec<(String, Vec)> { + if !super::system_channels::filter_is_group_state_only(filter) { return Vec::new(); } [ diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index f3dbc8a875..ca4a2304dd 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1036,7 +1036,8 @@ pub fn emit_live_thread_summary( /// Emit a relay-signed membership notification event stored globally (channel_id = None). /// -/// kind:44100 = member added, kind:44101 = member removed. +/// kind:44100 = member added, kind:44101 = member removed. System channels +/// emit none. /// The p tag addresses the target pubkey; the h tag carries the channel UUID as metadata. /// Stored with channel_id = None so global subscribers receive it via slow-path fan-out. pub async fn emit_membership_notification( @@ -1047,6 +1048,16 @@ pub async fn emit_membership_notification( actor_pubkey: &[u8], notification_kind: u32, ) -> anyhow::Result<()> { + // A system channel is left out of clients' channel lists, so it sends no + // membership notifications: a client that receives one would show it. + if !state + .db + .system_channel_ids_among(tenant.community(), &[channel_id]) + .await? + .is_empty() + { + return Ok(()); + } let target_hex = hex::encode(target_pubkey); let actor_hex = hex::encode(actor_pubkey); let channel_id_str = channel_id.to_string(); diff --git a/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs new file mode 100644 index 0000000000..32ab94d925 --- /dev/null +++ b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs @@ -0,0 +1,445 @@ +//! Postgres-backed end-to-end tests for system channels: every read path +//! that does not name a system channel leaves it out, and naming it reads it +//! under the ordinary rules (member, or open channel). + +use std::sync::Arc; + +use buzz_core::TenantContext; +use buzz_db::event::EventQuery; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use uuid::Uuid; + +use super::group_state_postgres_tests::{create_channel, edit_channel, http_auth, test_conn}; +use crate::handlers::ingest::ingest_event; +use crate::state::AppState; + +struct Fixture { + state: Arc, + tenant: TenantContext, + host: String, +} + +async fn fixture() -> Fixture { + let state = crate::api::bridge::postgres_tests::bridge_handler_test_state() + .await + .expect("local Postgres and Redis"); + let host = format!("system-channels-{}.local", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + let tenant = TenantContext::resolved(community, &host); + Fixture { + state, + tenant, + host, + } +} + +/// Ingest an event that `keys` signs and return its ID. +async fn publish(f: &Fixture, keys: &Keys, kind: u16, content: &str, tags: &[&[&str]]) -> String { + let tags: Vec = tags + .iter() + .map(|tag| Tag::parse(tag.iter().copied()).unwrap()) + .collect(); + let event = EventBuilder::new(Kind::Custom(kind), content) + .tags(tags) + .sign_with_keys(keys) + .expect("sign"); + let id = event.id.to_hex(); + let result = ingest_event(&f.state, &f.tenant, event, http_auth(keys)) + .await + .unwrap_or_else(|error| panic!("ingest kind:{kind}: {error:?}")); + assert!(result.accepted, "kind:{kind} refused: {}", result.message); + id +} + +/// The current relay-signed group-state event ID of `kind` for `channel`. +async fn group_state_id(f: &Fixture, channel: Uuid, kind: i32) -> String { + f.state + .db + .query_events(&EventQuery { + kinds: Some(vec![kind]), + d_tag: Some(channel.to_string()), + limit: Some(1), + ..EventQuery::for_community(f.tenant.community()) + }) + .await + .expect("query group state") + .first() + .unwrap_or_else(|| panic!("no kind:{kind} for {channel}")) + .event + .id + .to_hex() +} + +/// Run one historical REQ and return the event IDs sent before EOSE. A +/// CLOSED reply counts as no events. +async fn req(f: &Fixture, reader: &Keys, filter: Value) -> Vec { + let (conn, mut rx) = test_conn(&f.tenant, reader); + let filter: nostr::Filter = serde_json::from_value(filter).expect("filter"); + crate::handlers::req::handle_req("s".into(), vec![filter], vec![None], conn, f.state.clone()) + .await; + let mut ids = Vec::new(); + while let Ok(frame) = rx.try_recv() { + let axum::extract::ws::Message::Text(text) = frame else { + continue; + }; + let message: Value = serde_json::from_str(&text).expect("relay frame"); + match message[0].as_str() { + Some("EVENT") => ids.push(message[2]["id"].as_str().expect("id").to_owned()), + Some("EOSE") | Some("CLOSED") => return ids, + other => panic!("unexpected frame {other:?}: {text}"), + } + } + panic!("no EOSE; events so far: {ids:?}"); +} + +/// Run one WebSocket COUNT and return the count. A CLOSED reply counts as 0. +async fn count(f: &Fixture, reader: &Keys, filter: Value) -> u64 { + let (conn, mut rx) = test_conn(&f.tenant, reader); + let filter: nostr::Filter = serde_json::from_value(filter).expect("filter"); + crate::handlers::count::handle_count("c".into(), vec![filter], conn, f.state.clone()).await; + while let Ok(frame) = rx.try_recv() { + let axum::extract::ws::Message::Text(text) = frame else { + continue; + }; + let message: Value = serde_json::from_str(&text).expect("relay frame"); + match message[0].as_str() { + Some("COUNT") => return message[2]["count"].as_u64().expect("count"), + Some("CLOSED") => return 0, + other => panic!("unexpected frame {other:?}: {text}"), + } + } + panic!("no COUNT reply"); +} + +/// POST `filters` to an HTTP bridge route as `reader`. Returns the JSON body +/// of a 200, or `None` for a refusal. +async fn bridge(f: &Fixture, reader: &Keys, path: &str, filters: Value) -> Option { + use axum::body::Body; + use axum::http::{header, Request}; + use tower::ServiceExt; + + let response = crate::router::build_router(f.state.clone()) + .oneshot( + Request::builder() + .method("POST") + .uri(path) + .header(header::HOST, &f.host) + .header("x-pubkey", reader.public_key().to_hex()) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(&filters).unwrap())) + .expect("request"), + ) + .await + .expect("router"); + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + if status.is_success() { + Some(serde_json::from_slice(&body).expect("json body")) + } else { + assert!( + status.is_client_error(), + "{path}: {status} {}", + String::from_utf8_lossy(&body) + ); + None + } +} + +fn bridge_ids(body: Option) -> Vec { + let Some(body) = body else { + return Vec::new(); + }; + let events = body + .as_array() + .or_else(|| body["events"].as_array()) + .unwrap_or_else(|| panic!("bridge query body: {body}")); + events + .iter() + .map(|event| event["id"].as_str().expect("id").to_owned()) + .collect() +} + +fn bridge_count(body: Option) -> u64 { + body.map_or(0, |body| { + body["count"] + .as_u64() + .unwrap_or_else(|| panic!("bridge count body: {body}")) + }) +} + +fn sorted(mut ids: Vec) -> Vec { + ids.sort(); + ids +} + +/// The relay leaves system channels out of every read that does not name +/// them, and a read that names one follows the ordinary rules. +/// Mutation: drop `channel_type <> 'system'` from either half of +/// `get_accessible_channel_ids` → RED (implicit reads list them). Drop the +/// named-channel scope → RED (members cannot read their own system channel). +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn system_channels_are_read_only_by_name() { + let f = fixture().await; + let owner = Keys::generate(); + let outsider = Keys::generate(); + + let private = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await; + let open = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await; + edit_channel( + &f.state, + &f.tenant, + &owner, + open, + &[&["visibility", "open"]], + ) + .await + .expect("open the system channel"); + let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await; + edit_channel( + &f.state, + &f.tenant, + &owner, + ordinary, + &[&["visibility", "open"]], + ) + .await + .expect("open the ordinary channel"); + + let word = format!("needle{}", Uuid::new_v4().simple()); + let mut message = std::collections::HashMap::new(); + for channel in [private, open, ordinary] { + let id = publish(&f, &owner, 9, &word, &[&["h", &channel.to_string()]]).await; + message.insert(channel, id); + } + let ordinary_message = vec![message[&ordinary].clone()]; + + for reader in [&owner, &outsider] { + let who = if std::ptr::eq(reader, &owner) { + "member" + } else { + "outsider" + }; + + // Implicit scope: no channel named. + assert_eq!( + req(&f, reader, json!({"kinds": [9]})).await, + ordinary_message, + "{who}: REQ without #h" + ); + assert_eq!( + req(&f, reader, json!({"kinds": [9], "search": word})).await, + ordinary_message, + "{who}: NIP-50 search" + ); + assert_eq!( + count(&f, reader, json!({"kinds": [9]})).await, + 1, + "{who}: COUNT" + ); + assert_eq!( + bridge_ids(bridge(&f, reader, "/query", json!([{"kinds": [9]}])).await), + ordinary_message, + "{who}: bridge /query" + ); + assert_eq!( + bridge_ids( + bridge( + &f, + reader, + "/query", + json!([{"kinds": [9], "search": word}]) + ) + .await + ), + ordinary_message, + "{who}: bridge search" + ); + assert_eq!( + bridge_count(bridge(&f, reader, "/count", json!([{"kinds": [9]}])).await), + 1, + "{who}: bridge /count" + ); + assert_eq!( + req(&f, reader, json!({"kinds": [39000]})).await, + vec![group_state_id(&f, ordinary, 39000).await], + "{who}: channel list" + ); + + // Named scope: member, or the channel is open. + for (channel, readable) in [(private, std::ptr::eq(reader, &owner)), (open, true)] { + let h = channel.to_string(); + let expected_message = if readable { + vec![message[&channel].clone()] + } else { + Vec::new() + }; + let expected_count = u64::from(readable); + assert_eq!( + req(&f, reader, json!({"kinds": [9], "#h": [h]})).await, + expected_message, + "{who}: REQ #h {channel}" + ); + assert_eq!( + count(&f, reader, json!({"kinds": [9], "#h": [h]})).await, + expected_count, + "{who}: COUNT #h {channel}" + ); + assert_eq!( + bridge_ids(bridge(&f, reader, "/query", json!([{"kinds": [9], "#h": [h]}])).await), + expected_message, + "{who}: bridge /query #h {channel}" + ); + assert_eq!( + bridge_count( + bridge(&f, reader, "/count", json!([{"kinds": [9], "#h": [h]}])).await + ), + expected_count, + "{who}: bridge /count #h {channel}" + ); + assert_eq!( + req(&f, reader, json!({"kinds": [9], "#h": [h], "search": word})).await, + expected_message, + "{who}: search #h {channel}" + ); + let expected_metadata = if readable { + vec![group_state_id(&f, channel, 39000).await] + } else { + Vec::new() + }; + assert_eq!( + req(&f, reader, json!({"kinds": [39000], "#d": [h]})).await, + expected_metadata, + "{who}: metadata #d {channel}" + ); + } + } + + // The member's own channel list (kind:39002 #p:[me]) leaves them out. + assert_eq!( + req( + &f, + &owner, + json!({"kinds": [39002], "#p": [owner.public_key().to_hex()]}) + ) + .await, + vec![group_state_id(&f, ordinary, 39002).await], + "member list by #p" + ); +} + +/// A reader can join a private system channel after the request loads its +/// readable system channels and before it re-checks a named channel. The +/// re-check then confirms membership. The channel must stay in the scope of +/// the filter that names it, not the request-wide implicit set, so a sibling +/// filter that never names it cannot read it. +/// Mutation: drop `keep_repaired_system_channels_named` from +/// `repair_requested_channel_access` → RED (the sibling filter's scope gets +/// the channel). +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn a_system_channel_joined_mid_request_stays_in_its_named_filter() { + let f = fixture().await; + let owner = Keys::generate(); + let system = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await; + let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await; + + let filters: Vec = serde_json::from_value(json!([ + {"kinds": [9], "#h": [system.to_string()]}, + {"kinds": [39000]}, + ])) + .expect("filters"); + let mut implicit = f + .state + .db + .get_accessible_channel_ids(f.tenant.community(), &owner.public_key().to_bytes()) + .await + .expect("implicit channels"); + assert_eq!(implicit, [ordinary]); + // The request's system-channel lookup ran before the join. + let mut readable_system = Vec::new(); + + crate::api::bridge::repair_requested_channel_access( + &f.state, + &f.tenant, + &filters, + &owner.public_key().to_bytes(), + &mut implicit, + &mut readable_system, + ) + .await + .expect("repair"); + + assert_eq!(implicit, [ordinary], "implicit set stays ordinary"); + assert_eq!(readable_system, [system]); + let scope = |filter| { + crate::handlers::system_channels::filter_channel_scope(filter, &implicit, &readable_system) + .into_owned() + }; + assert_eq!(scope(&filters[0]), [ordinary, system], "naming filter"); + assert_eq!(scope(&filters[1]), [ordinary], "sibling filter"); +} + +/// Adding a member to a system channel sends no kind:44100 notification, +/// which an existing client would show as a new channel. An ordinary +/// channel still sends one, so the check is not vacuous. +/// Mutation: drop the system check in `emit_membership_notification` → RED. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn system_channels_send_no_membership_notifications() { + let f = fixture().await; + let owner = Keys::generate(); + let member = Keys::generate(); + let member_hex = member.public_key().to_hex(); + + let system = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await; + let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await; + for channel in [system, ordinary] { + publish( + &f, + &owner, + 9000, + "", + &[&["h", &channel.to_string()], &["p", &member_hex]], + ) + .await; + } + // Notifications are written after the reply; give the task a moment. + let mut notified = Vec::new(); + for _ in 0..50 { + notified = f + .state + .db + .query_events(&EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32]), + p_tag_hex: Some(member_hex.clone()), + ..EventQuery::for_community(f.tenant.community()) + }) + .await + .expect("query notifications") + .into_iter() + .map(|stored| { + stored + .event + .tags + .iter() + .find(|tag| tag.as_slice().first().map(String::as_str) == Some("h")) + .and_then(|tag| tag.as_slice().get(1).cloned()) + .expect("h tag") + }) + .collect::>(); + if !notified.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert_eq!(sorted(notified), vec![ordinary.to_string()]); +} diff --git a/crates/buzz-relay/src/handlers/system_channels.rs b/crates/buzz-relay/src/handlers/system_channels.rs new file mode 100644 index 0000000000..e7ee917b36 --- /dev/null +++ b/crates/buzz-relay/src/handlers/system_channels.rs @@ -0,0 +1,262 @@ +//! Query scope for system channels. +//! +//! A system channel (`channel_type = 'system'`) exists only for access +//! control. The implicit channel set of a query, from +//! `get_accessible_channel_ids`, leaves system channels out. A filter adds +//! the system channels that the reader can read (member, or open) only when +//! it names them: +//! +//! - `#h` lists the channel, on any kinds; +//! - `#d` lists the channel, on a filter whose kinds are all 39000–39003. +//! +//! The addition is per filter. Another filter in the same request keeps the +//! implicit set, so naming one system channel cannot widen a sibling filter. + +use std::borrow::Cow; + +use nostr::{Alphabet, Filter, SingleLetterTag}; +use uuid::Uuid; + +use buzz_core::CommunityId; + +use crate::state::AppState; + +/// Whether every kind in `filter` is a NIP-29 group-state kind (39000–39003). +pub(crate) fn filter_is_group_state_only(filter: &Filter) -> bool { + filter.kinds.as_ref().is_some_and(|kinds| { + !kinds.is_empty() + && kinds.iter().all(|kind| { + (buzz_core::kind::KIND_NIP29_GROUP_METADATA + ..=buzz_core::kind::KIND_NIP29_GROUP_ROLES) + .contains(&u32::from(kind.as_u16())) + }) + }) +} + +/// Channel IDs that `filter` names: its `#h` values, and its `#d` values when +/// the filter asks only for group-state kinds. Malformed values are skipped. +fn named_channels(filter: &Filter) -> impl Iterator + '_ { + let h = filter + .generic_tags + .get(&SingleLetterTag::lowercase(Alphabet::H)); + let d = filter_is_group_state_only(filter) + .then(|| { + filter + .generic_tags + .get(&SingleLetterTag::lowercase(Alphabet::D)) + }) + .flatten(); + h.into_iter() + .chain(d) + .flatten() + .filter_map(|value| value.parse::().ok()) +} + +/// Whether `filter` could need a system channel added to its scope. +fn filter_needs_system_channels(filter: &Filter, implicit: &[Uuid]) -> bool { + named_channels(filter).any(|id| !implicit.contains(&id)) +} + +/// The channels `filter` may read: the implicit set, plus the readable +/// system channels the filter names. +pub(crate) fn filter_channel_scope<'a>( + filter: &Filter, + implicit: &'a [Uuid], + readable_system: &[Uuid], +) -> Cow<'a, [Uuid]> { + if readable_system.is_empty() { + return Cow::Borrowed(implicit); + } + let mut extra: Vec = Vec::new(); + for id in named_channels(filter) { + if readable_system.contains(&id) && !implicit.contains(&id) && !extra.contains(&id) { + extra.push(id); + } + } + if extra.is_empty() { + return Cow::Borrowed(implicit); + } + let mut scope = implicit.to_vec(); + scope.extend(extra); + Cow::Owned(scope) +} + +/// Load the system channels the reader can read, narrowed to the token's +/// channel scope, when at least one filter could use them. Returns an empty +/// set, without a database read, when no filter names a channel outside the +/// implicit set. +pub(crate) async fn readable_system_channels_for_filters( + state: &AppState, + community: CommunityId, + pubkey: &[u8], + filters: &[Filter], + implicit: &[Uuid], + token_channel_ids: Option<&[Uuid]>, +) -> Result, buzz_db::DbError> { + if !filters + .iter() + .any(|filter| filter_needs_system_channels(filter, implicit)) + { + return Ok(Vec::new()); + } + let mut readable = state + .db + .get_readable_system_channel_ids(community, pubkey) + .await?; + if let Some(allowed) = token_channel_ids { + readable.retain(|id| allowed.contains(id)); + } + Ok(readable) +} + +/// Move system channels out of the repaired tail of the implicit set. +/// +/// A request re-checks membership for named channels missing from its first +/// lookups, and [`super::req::resolve_request_local_access`] appends each +/// confirmed one to the implicit set. A system channel lands there when the +/// reader joins it between the lookups and the re-check. This moves it to +/// `readable_system`, so only the filters that name it can read it. +/// +/// `repaired_from` is the length of `implicit` before the re-check. The tail +/// after it holds the repaired channels; nothing before it is a system +/// channel. +pub(crate) async fn keep_repaired_system_channels_named( + state: &AppState, + community: CommunityId, + implicit: &mut Vec, + repaired_from: usize, + readable_system: &mut Vec, +) -> Result<(), buzz_db::DbError> { + let Some(repaired) = implicit + .get(repaired_from..) + .filter(|tail| !tail.is_empty()) + else { + return Ok(()); + }; + let system = state + .db + .system_channel_ids_among(community, repaired) + .await?; + move_system_channels(implicit, repaired_from, &system, readable_system); + Ok(()) +} + +fn move_system_channels( + implicit: &mut Vec, + repaired_from: usize, + system: &[Uuid], + readable_system: &mut Vec, +) { + if system.is_empty() { + return; + } + let repaired = implicit.split_off(repaired_from); + for id in repaired { + if !system.contains(&id) { + implicit.push(id); + } else if !readable_system.contains(&id) { + readable_system.push(id); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn filter(json: serde_json::Value) -> Filter { + serde_json::from_value(json).expect("filter") + } + + #[test] + fn named_system_channel_widens_only_its_own_filter() { + let open = Uuid::new_v4(); + let system = Uuid::new_v4(); + let implicit = vec![open]; + let readable = vec![system]; + + let by_h = filter(serde_json::json!({"#h": [system.to_string()]})); + assert_eq!( + filter_channel_scope(&by_h, &implicit, &readable).as_ref(), + &[open, system] + ); + + let by_d = filter(serde_json::json!({"kinds": [39000], "#d": [system.to_string()]})); + assert_eq!( + filter_channel_scope(&by_d, &implicit, &readable).as_ref(), + &[open, system] + ); + + // #d names a channel only on group-state kinds. + let d_other_kind = + filter(serde_json::json!({"kinds": [39000, 1], "#d": [system.to_string()]})); + assert_eq!( + filter_channel_scope(&d_other_kind, &implicit, &readable).as_ref(), + &[open] + ); + + // A sibling filter that names nothing keeps the implicit set. + let sibling = filter(serde_json::json!({"kinds": [9]})); + assert_eq!( + filter_channel_scope(&sibling, &implicit, &readable).as_ref(), + &[open] + ); + + // An unreadable system channel is never added. + let unreadable = Uuid::new_v4(); + let other = filter(serde_json::json!({"#h": [unreadable.to_string()]})); + assert_eq!( + filter_channel_scope(&other, &implicit, &readable).as_ref(), + &[open] + ); + } + + #[test] + fn a_repaired_system_channel_stays_in_the_filter_that_names_it() { + let open = Uuid::new_v4(); + let repaired_open = Uuid::new_v4(); + let joined_system = Uuid::new_v4(); + let mut implicit = vec![open, repaired_open, joined_system]; + let mut readable = Vec::new(); + move_system_channels(&mut implicit, 1, &[joined_system], &mut readable); + assert_eq!(implicit, [open, repaired_open]); + assert_eq!(readable, [joined_system]); + + let names_it = filter(serde_json::json!({"#h": [joined_system.to_string()]})); + assert_eq!( + filter_channel_scope(&names_it, &implicit, &readable).as_ref(), + &[open, repaired_open, joined_system] + ); + let sibling = filter(serde_json::json!({"kinds": [39000]})); + assert_eq!( + filter_channel_scope(&sibling, &implicit, &readable).as_ref(), + &[open, repaired_open] + ); + } + + #[test] + fn lookup_runs_only_when_a_filter_names_a_channel_outside_the_implicit_set() { + let open = Uuid::new_v4(); + let implicit = vec![open]; + assert!(!filter_needs_system_channels( + &filter(serde_json::json!({"kinds": [9]})), + &implicit + )); + assert!(!filter_needs_system_channels( + &filter(serde_json::json!({"#h": [open.to_string()]})), + &implicit + )); + assert!(filter_needs_system_channels( + &filter(serde_json::json!({"#h": [Uuid::new_v4().to_string()]})), + &implicit + )); + assert!(filter_needs_system_channels( + &filter(serde_json::json!({"kinds": [39002], "#d": [Uuid::new_v4().to_string()]})), + &implicit + )); + assert!(!filter_needs_system_channels( + &filter(serde_json::json!({"kinds": [1], "#d": [Uuid::new_v4().to_string()]})), + &implicit + )); + } +} diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 42930f6c1f..c89284f948 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -138,11 +138,18 @@ async fn load_match_context( let mut authors: Vec> = leases.iter().map(|lease| lease.author.clone()).collect(); authors.sort_unstable(); authors.dedup(); + // System channels never push: drop their membership pairs, so a channel + // event there matches no lease (channel events require membership). + let system = state + .db + .system_channel_ids_among(batch.community, &channels) + .await?; let memberships = state .db .membership_pairs(batch.community, &channels, &authors) .await? .into_iter() + .filter(|(channel, _)| !system.contains(channel)) .collect(); Ok(MatchContext { leases, diff --git a/docs/nips/NIP-CL.md b/docs/nips/NIP-CL.md index 844d89a3a2..81036df16f 100644 --- a/docs/nips/NIP-CL.md +++ b/docs/nips/NIP-CL.md @@ -166,6 +166,30 @@ channel. It is private unless the kind:9007 also has `["visibility", "open"]`. **Hidden.** While a system channel is private, its kind:39000 carries the NIP-29 `hidden` tag. +### Reads that leave system channels out + +The relay leaves system channels out of every read that does not name them. +Clients that do not know about system channels then stay correct with no +change. This applies to: + +- a filter without `#h`, for example the channels or messages a reader can see; +- kind:39002 `#p:[""]`, the channels a reader belongs to; +- COUNT and NIP-50 search; and +- any relay-specific query API that does the same reads. + +A filter **names** a system channel when it has the channel ID in `#h`, or in +`#d` on a filter whose kinds are all 39000–39003. Then the ordinary read rules +apply: the reader can read the channel if they are a member or if it is open. +A client that has the ID, for example from a link, can open the channel. Thus +a system channel stays an ordinary NIP-29 group. + +The relay decides this for each filter. A filter that names a system channel +does not add it to another filter in the same request. + +The relay MUST NOT send push notifications or kind:44100/44101 membership +notifications for system channels. A client that received one would show the +channel. + ## Trust A client that relies on these tags MUST check that the relay's key signed the diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index e8f4ca54cb..5b85f2c4f2 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -190,6 +190,9 @@ run_unit_tests() { run_test_step "buzz-relay side-effects helper tests" \ cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture + run_test_step "buzz-relay system channel scope tests" \ + cargo test -p buzz-relay --lib handlers::system_channels::tests:: -- --nocapture + run_test_step "buzz-relay storage snapshot tests" \ cargo test -p buzz-relay --lib storage_sweep::tests:: -- --nocapture