mirror of
https://github.com/block/buzz.git
synced 2026-11-14 05:12:27 +00:00
feat(relay): leave system channels out of reads that do not name them
The implicit channel set of a query (get_accessible_channel_ids, the union of the reader's memberships and the open channels) now leaves system channels out of both halves. So REQ, COUNT, NIP-50 search, the HTTP bridge /query and /count, the sidebar projection and the channel list skip them, and an existing client never sees one. A filter that names a system channel (#h, or #d on kinds 39000-39003) can read it under the ordinary rules: the reader is a member, or the channel is open. The addition applies to that filter only, so naming one system channel does not widen a sibling filter. Thread windows and the workflows API name their channel, so they follow the same rule. Push matching and kind:44100/44101 membership notifications skip system channels. Signed-off-by: Logan Johnson <loganj@squareup.com> Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -1107,6 +1107,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,
|
||||
@@ -1123,10 +1127,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())
|
||||
@@ -1142,6 +1148,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<Vec<Uuid>> {
|
||||
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<Vec<Uuid>> {
|
||||
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)]
|
||||
@@ -1367,6 +1439,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'
|
||||
"#
|
||||
);
|
||||
|
||||
@@ -1772,6 +1845,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<Vec<Uuid>> {
|
||||
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<Vec<Uuid>> {
|
||||
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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1403,12 +1403,25 @@ 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 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,
|
||||
&readable_system,
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -1424,6 +1437,7 @@ async fn query_events_authed(
|
||||
&raw_filters,
|
||||
&filters,
|
||||
&accessible_channels,
|
||||
&readable_system,
|
||||
tenant,
|
||||
&authed_pubkey_hex,
|
||||
&pubkey_bytes,
|
||||
@@ -1455,7 +1469,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);
|
||||
}
|
||||
}
|
||||
@@ -1471,7 +1490,11 @@ async fn query_events_authed(
|
||||
tenant,
|
||||
raw,
|
||||
filter,
|
||||
&accessible_channels,
|
||||
&crate::handlers::system_channels::filter_channel_scope(
|
||||
filter,
|
||||
&accessible_channels,
|
||||
&readable_system,
|
||||
),
|
||||
&mut events,
|
||||
)
|
||||
.await?;
|
||||
@@ -1592,6 +1615,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) {
|
||||
@@ -1681,6 +1709,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) {
|
||||
@@ -1782,6 +1815,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 {
|
||||
@@ -1819,6 +1857,7 @@ async fn repair_requested_channel_access(
|
||||
filters: &[nostr::Filter],
|
||||
pubkey_bytes: &[u8],
|
||||
accessible_channels: &mut Vec<uuid::Uuid>,
|
||||
readable_system: &[uuid::Uuid],
|
||||
) -> Result<(), (StatusCode, Json<Value>)> {
|
||||
for filter in filters {
|
||||
let Some(requested) =
|
||||
@@ -1827,7 +1866,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
|
||||
@@ -1999,17 +2039,35 @@ 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 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,
|
||||
&readable_system,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let mut total: u64 = 0;
|
||||
for filter in &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
|
||||
@@ -2228,26 +2286,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<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
// 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<Value> = Vec::new();
|
||||
let mut seen_ids: std::collections::HashSet<[u8; 32]> = std::collections::HashSet::new();
|
||||
|
||||
@@ -2263,6 +2312,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.
|
||||
@@ -2279,7 +2335,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| {
|
||||
|
||||
@@ -84,6 +84,40 @@ fn append(events: &mut Vec<Value>, 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<uuid::Uuid>, Vec<uuid::Uuid>), 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::<HashSet<_>>() == b.iter().collect::<HashSet<_>>()
|
||||
}
|
||||
|
||||
/// 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<Item = &'a Request>,
|
||||
) -> Result<Vec<Value>, 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::<HashSet<_>>() != current.iter().collect::<HashSet<_>>() {
|
||||
if !same_set(&implicit, ¤t_implicit) || !same_set(&system, ¤t_system) {
|
||||
return Err(unavailable("thread authorization changed; retry query"));
|
||||
}
|
||||
Ok(events)
|
||||
|
||||
@@ -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(api_error(StatusCode::FORBIDDEN, "workflow is not accessible").into_response());
|
||||
}
|
||||
|
||||
|
||||
@@ -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 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.
|
||||
@@ -145,7 +164,7 @@ pub async fn handle_count(
|
||||
|
||||
if let Some(requested_channels) = requested_channels {
|
||||
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 +193,15 @@ pub async fn handle_count(
|
||||
db_is_member,
|
||||
);
|
||||
}
|
||||
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 +219,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 +298,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());
|
||||
|
||||
@@ -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<AppState>,
|
||||
tenant: &TenantContext,
|
||||
keys: &Keys,
|
||||
@@ -551,15 +551,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<AppState>,
|
||||
/// 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<String> {
|
||||
let (send_tx, mut send_rx) = tokio::sync::mpsc::channel(256);
|
||||
) -> (
|
||||
Arc<crate::connection::ConnectionState>,
|
||||
tokio::sync::mpsc::Receiver<axum::extract::ws::Message>,
|
||||
) {
|
||||
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 {
|
||||
@@ -587,6 +588,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<AppState>,
|
||||
tenant: &TenantContext,
|
||||
reader: &Keys,
|
||||
filter: serde_json::Value,
|
||||
) -> Vec<String> {
|
||||
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;
|
||||
@@ -606,7 +619,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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 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
|
||||
@@ -150,7 +169,12 @@ pub async fn handle_req(
|
||||
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
|
||||
@@ -191,7 +215,9 @@ pub async fn handle_req(
|
||||
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::<Vec<_>>()
|
||||
});
|
||||
// Partial authorization preserves NIP-01 OR semantics by omitting only
|
||||
@@ -318,6 +344,7 @@ pub async fn handle_req(
|
||||
owner,
|
||||
&filters,
|
||||
&accessible_channels,
|
||||
&readable_system,
|
||||
token_channel_ids.is_none(),
|
||||
&conn.tenant,
|
||||
&pubkey_bytes,
|
||||
@@ -416,7 +443,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 +479,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 +546,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 +720,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 +729,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<nostr::EventId> = HashSet::new();
|
||||
|
||||
@@ -711,6 +746,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 +784,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| {
|
||||
|
||||
@@ -985,7 +985,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(
|
||||
@@ -996,6 +997,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();
|
||||
|
||||
393
crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs
Normal file
393
crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs
Normal file
@@ -0,0 +1,393 @@
|
||||
//! 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<AppState>,
|
||||
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<Tag> = 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<String> {
|
||||
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<Value> {
|
||||
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<Value>) -> Vec<String> {
|
||||
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<Value>) -> u64 {
|
||||
body.map_or(0, |body| {
|
||||
body["count"]
|
||||
.as_u64()
|
||||
.unwrap_or_else(|| panic!("bridge count body: {body}"))
|
||||
})
|
||||
}
|
||||
|
||||
fn sorted(mut ids: Vec<String>) -> Vec<String> {
|
||||
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"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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::<Vec<_>>();
|
||||
if !notified.is_empty() {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
|
||||
}
|
||||
assert_eq!(sorted(notified), vec![ordinary.to_string()]);
|
||||
}
|
||||
188
crates/buzz-relay/src/handlers/system_channels.rs
Normal file
188
crates/buzz-relay/src/handlers/system_channels.rs
Normal file
@@ -0,0 +1,188 @@
|
||||
//! 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<Item = Uuid> + '_ {
|
||||
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::<Uuid>().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<Uuid> = 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<Vec<Uuid>, 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)
|
||||
}
|
||||
|
||||
#[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 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
|
||||
));
|
||||
}
|
||||
}
|
||||
@@ -138,11 +138,18 @@ async fn load_match_context(
|
||||
let mut authors: Vec<Vec<u8>> = 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,
|
||||
|
||||
Reference in New Issue
Block a user