diff --git a/.env.example b/.env.example index 61f44af72d..6674deec40 100644 --- a/.env.example +++ b/.env.example @@ -151,7 +151,8 @@ BUZZ_PUSH_GATEWAY_DELIVERY_URL=https://push.buzz.xyz/v1/deliveries/apns # Deployment-global operator-listener routes. Each configured Nostr identity # may register target pubkeys and receives mention notifications at its URL. # URLs may use HTTP or HTTPS. Use HTTP only for operator-controlled internal -# connections: NIP-98 signs requests. +# connections: NIP-98 signs requests but does not encrypt them, and +# notifications carry full message content. # Format: pubkey:deliveryUrl;pubkey:deliveryUrl # BUZZ_OPERATOR_LISTENER_TIMEOUT_MS=5000 # BUZZ_OPERATOR_LISTENERS=<64-char hex pubkey>:https://listener.example/mentions diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index db05ea922f..e645dea011 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -79,6 +79,9 @@ pub const KIND_AUTH: u32 = 22242; pub const KIND_BLOSSOM_AUTH: u32 = 24242; /// Buzz custom one-time identity binding proof (ephemeral, not stored). pub const KIND_NOSTR_IDENTITY_BINDING: u32 = 24243; +/// Operator-listener registration proof: a target pubkey consents to mention +/// delivery to the listener in its `p` tag (HTTP body only, not stored). +pub const KIND_OPERATOR_LISTENER_REGISTRATION: u32 = 24245; /// NIP-98: HTTP auth event (used in nip98.rs, not stored). pub const KIND_HTTP_AUTH: u32 = 27235; @@ -738,6 +741,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_AGENT_PRESENCE, KIND_HUDDLE_REACTION, KIND_BLOSSOM_AUTH, + KIND_OPERATOR_LISTENER_REGISTRATION, KIND_PAIRING, KIND_AGENT_OBSERVER_FRAME, KIND_HTTP_AUTH, diff --git a/crates/buzz-db/src/store/operator_listener.rs b/crates/buzz-db/src/store/operator_listener.rs index b730696902..18d9c1cca0 100644 --- a/crates/buzz-db/src/store/operator_listener.rs +++ b/crates/buzz-db/src/store/operator_listener.rs @@ -11,6 +11,7 @@ use uuid::Uuid; use crate::error::Result; use crate::AdmittedTx; use crate::{observability, CommunityId, Db}; +use buzz_core::StoredEvent; /// Event kinds whose `p` tags represent user-visible message mentions. pub const OPERATOR_LISTENER_MENTION_KINDS: [u32; 4] = [9, 40002, 45001, 45003]; @@ -254,6 +255,84 @@ pub async fn claim_deliveries( .collect() } +/// Community admission policy a delivery target must satisfy, mirroring the +/// relay's connection gate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DeliveryAdmission { + /// The target (or its owner, below) must be a community member. + pub require_membership: bool, + /// An agent whose recorded owner is a member counts as a member (NIP-OA). + pub allow_owner_membership: bool, + /// The target must be in the community's pubkey allowlist. + pub require_allowlist: bool, +} + +/// Load the event for a claimed delivery if the target may still receive it. +/// +/// Returns `None` when the registration was removed, the community is no +/// longer active, the event was deleted, the target or its recorded owner is +/// banned, the target fails `admission`, +/// or the target cannot read the event's channel: the channel was deleted, or +/// it is private (including DMs) and the target is not an active member. Runs +/// on the writer, which committed the outbox row with the event. +pub async fn load_deliverable_event( + pool: &PgPool, + delivery: &ClaimedDelivery, + admission: DeliveryAdmission, +) -> Result> { + let mut connection = acquire_maintenance_writer(pool).await?; + let row = sqlx::query( + "WITH principals AS ( \ + SELECT $2::bytea AS pubkey \ + UNION ALL \ + SELECT agent_owner_pubkey FROM users \ + WHERE community_id = $3 AND pubkey = $2 AND agent_owner_pubkey IS NOT NULL \ + ) \ + SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, \ + e.received_at, e.channel_id \ + FROM operator_listener_pubkeys r \ + JOIN communities co ON co.id = $3 AND co.archived_at IS NULL \ + AND co.deleted_at IS NULL AND co.deletion_state = 'active' \ + JOIN events e ON e.community_id = $3 AND e.created_at = $4 AND e.id = $5 \ + WHERE r.listener_pubkey = $1 AND r.target_pubkey = $2 \ + AND e.deleted_at IS NULL \ + AND (NOT $8 OR EXISTS ( \ + SELECT 1 FROM pubkey_allowlist a \ + WHERE a.community_id = $3 AND a.pubkey = $2)) \ + AND NOT EXISTS ( \ + SELECT 1 FROM community_bans b \ + WHERE b.community_id = $3 AND b.pubkey IN (SELECT pubkey FROM principals) \ + AND b.banned AND (b.ban_expires_at IS NULL OR b.ban_expires_at > now())) \ + AND (NOT $6 OR EXISTS ( \ + SELECT 1 FROM relay_members rm \ + WHERE rm.community_id = $3 AND rm.pubkey IN ( \ + SELECT encode(p.pubkey, 'hex') FROM principals p \ + WHERE $7 OR p.pubkey = $2))) \ + AND (e.channel_id IS NULL OR 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 = r.target_pubkey AND m.removed_at IS NULL))))", + ) + .bind(&delivery.listener_pubkey) + .bind(&delivery.target_pubkey) + .bind(delivery.community.as_uuid()) + .bind(delivery.event_created_at) + .bind(&delivery.event_id) + .bind(admission.require_membership) + .bind(admission.allow_owner_membership) + .bind(admission.require_allowlist) + .fetch_optional(&mut *connection) + .await?; + match row { + Some(row) => crate::event::row_to_stored_event(row), + None => Ok(None), + } +} + /// Release a delivery claim that this pod cannot route. pub async fn release_unroutable_delivery( pool: &PgPool, @@ -368,6 +447,16 @@ impl Db { claim_deliveries(&self.pool, limit, lease_until).await } + /// Load a claimed delivery's event if its target may still receive it. + #[datastore_span(name = "load_operator_listener_delivery_event", system = "postgresql")] + pub async fn load_operator_listener_delivery_event( + &self, + delivery: &ClaimedDelivery, + admission: DeliveryAdmission, + ) -> Result> { + load_deliverable_event(&self.pool, delivery, admission).await + } + /// Mark one operator-listener delivery successful. #[datastore_span(name = "complete_operator_listener_delivery", system = "postgresql")] pub async fn complete_operator_listener_delivery( @@ -438,6 +527,8 @@ mod tests { #[cfg(test)] mod postgres_tests { + use std::collections::HashMap; + use super::*; use nostr::{EventBuilder, Keys, Kind, Tag}; use sqlx::PgPool; @@ -1035,4 +1126,289 @@ mod postgres_tests { .await .expect("delete test registrations"); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deliverable_event_requires_registration_live_event_and_channel_access() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(format!( + "operator-listener-access-{}.example", + community.as_uuid().simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + let listener = Keys::generate(); + let target = Keys::generate(); + let author = Keys::generate(); + let listener_pubkey = listener.public_key().to_bytes().to_vec(); + let target_pubkey = target.public_key().to_bytes().to_vec(); + + let open_channel = Uuid::new_v4(); + let member_channel = Uuid::new_v4(); + let other_channel = Uuid::new_v4(); + for (channel, visibility) in [ + (open_channel, "open"), + (member_channel, "private"), + (other_channel, "private"), + ] { + sqlx::query( + "INSERT INTO channels (community_id, id, name, visibility, channel_type, created_by) \ + VALUES ($1, $2, $3, $4::channel_visibility, 'stream', $5)", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(channel.to_string()) + .bind(visibility) + .bind(author.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("insert test channel"); + } + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role) \ + VALUES ($1, $2, $3, 'member')", + ) + .bind(community.as_uuid()) + .bind(member_channel) + .bind(&target_pubkey) + .execute(&pool) + .await + .expect("insert target membership"); + register_pubkeys( + &pool, + &listener_pubkey, + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("register target"); + + let mut deliveries = HashMap::new(); + for channel in [ + Some(open_channel), + Some(member_channel), + Some(other_channel), + None, + ] { + let event = EventBuilder::new(Kind::Custom(9), format!("hello {channel:?}")) + .tag(Tag::public_key(target.public_key())) + .sign_with_keys(&author) + .expect("sign test event"); + crate::event::insert_event(&pool, community, &event, channel) + .await + .expect("insert event"); + deliveries.insert( + channel, + ClaimedDelivery { + id: Uuid::new_v4(), + claim_id: Uuid::new_v4(), + listener_pubkey: listener_pubkey.clone(), + target_pubkey: target_pubkey.clone(), + community, + community_host: String::new(), + event_id: event.id.as_bytes().to_vec(), + event_kind: 9, + event_created_at: DateTime::from_timestamp( + event.created_at.as_secs() as i64, + 0, + ) + .expect("event timestamp"), + attempt: 1, + }, + ); + } + const OPEN_RELAY: DeliveryAdmission = DeliveryAdmission { + require_membership: false, + allow_owner_membership: false, + require_allowlist: false, + }; + let load_with = |delivery: ClaimedDelivery, admission: DeliveryAdmission| { + let pool = pool.clone(); + async move { + load_deliverable_event(&pool, &delivery, admission) + .await + .expect("load deliverable event") + } + }; + let load = |channel: Uuid| load_with(deliveries[&Some(channel)].clone(), OPEN_RELAY); + + let open = load(open_channel).await.expect("open channel is readable"); + assert_eq!( + open.event.id.as_bytes().as_slice(), + deliveries[&Some(open_channel)].event_id.as_slice() + ); + open.event + .verify() + .expect("stored event round-trips with a valid signature"); + assert!(load(member_channel).await.is_some(), "member may read"); + assert!(load(other_channel).await.is_none(), "non-member may not"); + assert!( + load_with(deliveries[&None].clone(), OPEN_RELAY) + .await + .is_some(), + "channel-less event" + ); + let mut other_community = deliveries[&Some(open_channel)].clone(); + other_community.community = CommunityId::from_uuid(Uuid::new_v4()); + assert!(load_with(other_community, OPEN_RELAY).await.is_none()); + let mut other_timestamp = deliveries[&Some(open_channel)].clone(); + other_timestamp.event_created_at += TimeDelta::seconds(1); + assert!(load_with(other_timestamp, OPEN_RELAY).await.is_none()); + + // Closed community: direct membership, or a member owner when NIP-OA is on. + let closed = DeliveryAdmission { + require_membership: true, + ..OPEN_RELAY + }; + let closed_with_owner = DeliveryAdmission { + allow_owner_membership: true, + ..closed + }; + let open_delivery = deliveries[&Some(open_channel)].clone(); + assert!(load_with(open_delivery.clone(), closed).await.is_none()); + let owner = Keys::generate(); + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) \ + VALUES ($1, $3, NULL), ($1, $2, $3)", + ) + .bind(community.as_uuid()) + .bind(&target_pubkey) + .bind(owner.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("record target owner"); + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')", + ) + .bind(community.as_uuid()) + .bind(owner.public_key().to_hex()) + .execute(&pool) + .await + .expect("add owner membership"); + assert!(load_with(open_delivery.clone(), closed).await.is_none()); + assert!(load_with(open_delivery.clone(), closed_with_owner) + .await + .is_some()); + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')", + ) + .bind(community.as_uuid()) + .bind(target.public_key().to_hex()) + .execute(&pool) + .await + .expect("add target membership"); + assert!(load_with(open_delivery.clone(), closed).await.is_some()); + + // A ban on the target's owner cascades to the target. + sqlx::query( + "INSERT INTO community_bans (community_id, pubkey, banned, actor_pubkey) \ + VALUES ($1, $2, true, $2)", + ) + .bind(community.as_uuid()) + .bind(owner.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("ban target owner"); + assert!(load_with(open_delivery.clone(), OPEN_RELAY).await.is_none()); + sqlx::query("DELETE FROM community_bans WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("lift ban"); + assert!(load_with(open_delivery.clone(), OPEN_RELAY).await.is_some()); + + sqlx::query( + "UPDATE channel_members SET removed_at = now() \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community.as_uuid()) + .bind(member_channel) + .bind(&target_pubkey) + .execute(&pool) + .await + .expect("remove target membership"); + assert!(load(member_channel).await.is_none(), "removed member"); + + sqlx::query("UPDATE channels SET deleted_at = now() WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(open_channel) + .execute(&pool) + .await + .expect("delete open channel"); + assert!(load(open_channel).await.is_none(), "deleted channel"); + sqlx::query("UPDATE channels SET deleted_at = NULL WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(open_channel) + .execute(&pool) + .await + .expect("restore open channel"); + + sqlx::query("UPDATE events SET deleted_at = now() WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(&deliveries[&Some(open_channel)].event_id) + .execute(&pool) + .await + .expect("delete open event"); + assert!(load(open_channel).await.is_none(), "deleted event"); + + sqlx::query("UPDATE events SET deleted_at = NULL WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(&deliveries[&Some(open_channel)].event_id) + .execute(&pool) + .await + .expect("restore open event"); + + // The allowlist gate requires the target itself to be allowlisted. + let allowlisted = DeliveryAdmission { + require_allowlist: true, + ..OPEN_RELAY + }; + assert!(load_with(open_delivery.clone(), allowlisted) + .await + .is_none()); + sqlx::query("INSERT INTO pubkey_allowlist (community_id, pubkey) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(&target_pubkey) + .execute(&pool) + .await + .expect("allowlist target"); + assert!(load_with(open_delivery.clone(), allowlisted) + .await + .is_some()); + + // An archived community delivers nothing. Deletion is one-way, so + // only the archive predicate is exercised here. + for (statement, expected) in [ + ( + "UPDATE communities SET archived_at = now() WHERE id = $1", + false, + ), + ( + "UPDATE communities SET archived_at = NULL WHERE id = $1", + true, + ), + ] { + sqlx::query(statement) + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("update community archive state"); + assert_eq!( + load_with(open_delivery.clone(), OPEN_RELAY).await.is_some(), + expected, + "{statement}" + ); + } + remove_pubkeys( + &pool, + &listener_pubkey, + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("remove registration"); + assert!(load(open_channel).await.is_none(), "removed registration"); + } } diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 4a1fb717c1..ea5a4e65c1 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -70,6 +70,17 @@ struct ListenerPubkeysRequest { pubkeys: Vec, } +/// Registration body: one proof signed by each target pubkey. +#[derive(Debug, Deserialize)] +struct ListenerRegistrationRequest { + registrations: Vec, +} + +const MAX_LISTENER_PUBKEYS: usize = 1_000; +const LISTENER_PUBKEYS_PATH: &str = "/operator/listener/pubkeys"; +/// Matches the NIP-98 request window: proofs are signed just before sending. +const REGISTRATION_PROOF_MAX_SKEW_SECS: u64 = 60; + fn parse_listener_pubkeys(body: &[u8]) -> Result>, (StatusCode, Json)> { let request: ListenerPubkeysRequest = serde_json::from_slice(body).map_err(|e| { api_error( @@ -77,7 +88,7 @@ fn parse_listener_pubkeys(body: &[u8]) -> Result>, (StatusCode, Json &format!("invalid operator-listener pubkeys JSON: {e}"), ) })?; - if request.pubkeys.is_empty() || request.pubkeys.len() > 1_000 { + if request.pubkeys.is_empty() || request.pubkeys.len() > MAX_LISTENER_PUBKEYS { return Err(api_error( StatusCode::BAD_REQUEST, "pubkeys must contain between 1 and 1000 entries", @@ -103,6 +114,75 @@ fn parse_listener_pubkeys(body: &[u8]) -> Result>, (StatusCode, Json .collect() } +/// Parse a registration body and return the target pubkeys whose proofs +/// verify for `listener` at `url`. +fn parse_listener_registrations( + body: &[u8], + listener: &nostr::PublicKey, + url: &str, + now: u64, +) -> Result>, (StatusCode, Json)> { + let request: ListenerRegistrationRequest = serde_json::from_slice(body).map_err(|e| { + api_error( + StatusCode::BAD_REQUEST, + &format!("invalid operator-listener registration JSON: {e}"), + ) + })?; + if request.registrations.is_empty() || request.registrations.len() > MAX_LISTENER_PUBKEYS { + return Err(api_error( + StatusCode::BAD_REQUEST, + "registrations must contain between 1 and 1000 entries", + )); + } + request + .registrations + .iter() + .map(|proof| verify_registration_proof(proof, listener, url, now)) + .collect::>() + .map_err(|reason| api_error(StatusCode::BAD_REQUEST, reason)) +} + +/// Verify one target-signed registration proof and return the target pubkey. +/// +/// The proof binds the target's consent to this listener (`p`) and this +/// deployment's registration endpoint (`u`). The outer NIP-98 request is +/// replay-checked; within the skew window a proof can only re-register the +/// pair it was signed for. +fn verify_registration_proof( + proof: &nostr::Event, + listener: &nostr::PublicKey, + url: &str, + now: u64, +) -> Result, &'static str> { + if u32::from(proof.kind.as_u16()) != buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION { + return Err("registration proof has the wrong kind"); + } + if proof.verify().is_err() { + return Err("registration proof signature is invalid"); + } + if proof.created_at.as_secs().abs_diff(now) > REGISTRATION_PROOF_MAX_SKEW_SECS { + return Err("registration proof is outside the timestamp window"); + } + let listener_hex = listener.to_hex(); + let tag_values = |name: &str| { + proof + .tags + .iter() + .filter_map(|tag| match tag.as_slice() { + [tag_name, value, ..] if tag_name == name => Some(value.as_str()), + _ => None, + }) + .collect::>() + }; + if tag_values("p") != [listener_hex.as_str()] { + return Err("registration proof must name exactly this listener in one p tag"); + } + if tag_values("u") != [url] { + return Err("registration proof must name this endpoint in one u tag"); + } + Ok(proof.pubkey.to_bytes().to_vec()) +} + /// Shared deployment-global operator auth prelude. The canonical management /// origin and replay namespace are configuration, never tenant registry state /// or an inbound proxy `Host` header. @@ -187,15 +267,20 @@ pub async fn register_listener_pubkeys( headers: HeaderMap, body: axum::body::Bytes, ) -> Result, (StatusCode, Json)> { - let listener = authorize_operator_listener_request( - &state, - &headers, - "POST", - "/operator/listener/pubkeys", + let listener = + authorize_operator_listener_request(&state, &headers, "POST", LISTENER_PUBKEYS_PATH, &body) + .await?; + let origin = state + .config + .relay_operator_api_origin + .as_deref() + .ok_or_else(|| internal_error("operator API origin is not configured"))?; + let target_pubkeys = parse_listener_registrations( &body, - ) - .await?; - let target_pubkeys = parse_listener_pubkeys(&body)?; + &listener, + &format!("{origin}{LISTENER_PUBKEYS_PATH}"), + nostr::Timestamp::now().as_secs(), + )?; state .db .register_operator_listener_pubkeys(listener.as_bytes(), &target_pubkeys) @@ -214,7 +299,7 @@ pub async fn remove_listener_pubkeys( &state, &headers, "DELETE", - "/operator/listener/pubkeys", + LISTENER_PUBKEYS_PATH, &body, ) .await?; @@ -816,6 +901,130 @@ pub async fn community_availability( }))) } +#[cfg(test)] +mod tests { + use nostr::{EventBuilder, Keys, Kind, Tag, TagKind, Timestamp}; + + use super::*; + + const URL: &str = "https://operator.example/operator/listener/pubkeys"; + const NOW: u64 = 1_800_000_000; + + fn proof(target: &Keys, kind: u16, created_at: u64, tags: Vec) -> nostr::Event { + EventBuilder::new(Kind::Custom(kind), "") + .tags(tags) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(target) + .expect("sign registration proof") + } + + fn proof_tags(listener: &Keys, url: &str) -> Vec { + vec![ + Tag::public_key(listener.public_key()), + Tag::custom(TagKind::custom("u"), [url]), + ] + } + + const PROOF_KIND: u16 = buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION as u16; + + #[test] + fn registration_proof_returns_signing_target() { + let listener = Keys::generate(); + let target = Keys::generate(); + let valid = proof(&target, PROOF_KIND, NOW - 30, proof_tags(&listener, URL)); + + assert_eq!( + verify_registration_proof(&valid, &listener.public_key(), URL, NOW), + Ok(target.public_key().to_bytes().to_vec()) + ); + } + + #[test] + fn registration_proof_rejects_unbound_or_forged_consent() { + let listener = Keys::generate(); + let other_listener = Keys::generate(); + let target = Keys::generate(); + let mut forged = proof(&target, PROOF_KIND, NOW, proof_tags(&listener, URL)); + forged.pubkey = Keys::generate().public_key(); + let mut two_listeners = proof_tags(&listener, URL); + two_listeners.push(Tag::public_key(other_listener.public_key())); + + for (case, event) in [ + ("forged signer", forged), + ( + "wrong kind", + proof(&target, 27235, NOW, proof_tags(&listener, URL)), + ), + ( + "stale", + proof(&target, PROOF_KIND, NOW - 61, proof_tags(&listener, URL)), + ), + ( + "future", + proof(&target, PROOF_KIND, NOW + 61, proof_tags(&listener, URL)), + ), + ( + "other listener", + proof(&target, PROOF_KIND, NOW, proof_tags(&other_listener, URL)), + ), + ( + "ambiguous listener", + proof(&target, PROOF_KIND, NOW, two_listeners), + ), + ( + "other endpoint", + proof( + &target, + PROOF_KIND, + NOW, + proof_tags(&listener, "https://other.example/operator/listener/pubkeys"), + ), + ), + ( + "missing endpoint", + proof( + &target, + PROOF_KIND, + NOW, + vec![Tag::public_key(listener.public_key())], + ), + ), + ] { + assert!( + verify_registration_proof(&event, &listener.public_key(), URL, NOW).is_err(), + "{case} proof was accepted" + ); + } + } + + #[test] + fn registration_body_requires_signed_proofs() { + let listener = Keys::generate(); + let target = Keys::generate(); + let valid = proof(&target, PROOF_KIND, NOW, proof_tags(&listener, URL)); + let parse = |body: serde_json::Value| { + parse_listener_registrations( + body.to_string().as_bytes(), + &listener.public_key(), + URL, + NOW, + ) + }; + + assert_eq!( + parse(serde_json::json!({"registrations": [valid]})).expect("signed registration"), + vec![target.public_key().to_bytes().to_vec()] + ); + for body in [ + serde_json::json!({}), + serde_json::json!({"registrations": []}), + serde_json::json!({"pubkeys": [target.public_key().to_hex()]}), + ] { + assert!(parse(body.clone()).is_err(), "accepted {body}"); + } + } +} + #[cfg(test)] mod postgres_tests { use std::{ @@ -2200,7 +2409,28 @@ mod postgres_tests { .iter() .map(|target| target.public_key().to_hex()) .collect::>(); - let register_body = serde_json::json!({"pubkeys": target_hex}).to_string(); + let registration_proof = |target: &Keys, listener: &Keys| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION as u16), + "", + ) + .tags([ + Tag::public_key(listener.public_key()), + Tag::custom( + nostr::TagKind::custom("u"), + [format!("http://{INGRESS_HOST}/operator/listener/pubkeys")], + ), + ]) + .sign_with_keys(target) + .expect("sign registration proof") + }; + let register_body = serde_json::json!({ + "registrations": targets + .iter() + .map(|target| registration_proof(target, &listener)) + .collect::>() + }) + .to_string(); let response = signed_operator_request( Arc::clone(&state), &listener, @@ -2228,6 +2458,26 @@ mod postgres_tests { assert_eq!(listener_count, 2); assert_eq!(other_listener_count, 0); + // A proof signed for another listener, and the old unsigned form. + for rejected_body in [ + serde_json::json!({"registrations": [registration_proof(&targets[1], &other_listener)]}), + serde_json::json!({"pubkeys": [target_hex[1]]}), + ] { + let rejected = signed_operator_request( + Arc::clone(&state), + &listener, + "POST", + "/operator/listener/pubkeys", + Some(rejected_body.to_string()), + ) + .await; + assert_eq!( + rejected.status(), + StatusCode::BAD_REQUEST, + "{rejected_body}" + ); + } + let body = serde_json::json!({"pubkeys": [target_hex[0]]}).to_string(); let missing_payload_auth = nip98_auth_header_without_payload( &listener, @@ -2259,7 +2509,7 @@ mod postgres_tests { &listener, "POST", "/operator/listener/pubkeys", - Some(r#"{"pubkeys":["not-a-pubkey"]}"#.to_string()), + Some(r#"{"registrations":["not-an-event"]}"#.to_string()), ) .await; assert_eq!(malformed.status(), StatusCode::BAD_REQUEST); diff --git a/crates/buzz-relay/src/operator_listener.rs b/crates/buzz-relay/src/operator_listener.rs index 8545497e38..8adeb8df57 100644 --- a/crates/buzz-relay/src/operator_listener.rs +++ b/crates/buzz-relay/src/operator_listener.rs @@ -8,7 +8,7 @@ use serde::Serialize; use tracing::{error, warn}; use uuid::Uuid; -use crate::{nip98::nip98_header, state::AppState}; +use crate::{handlers::req::event_visible_to_reader, nip98::nip98_header, state::AppState}; use reqwest::StatusCode; @@ -20,14 +20,17 @@ const OUTBOX_REAP_INTERVAL: Duration = Duration::from_secs(5 * 60); const REGISTRATION_REAP_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); /// Notification sent to the configured operator-listener endpoint. +/// +/// Version 2 adds the full signed `event` to version 1's metadata. #[derive(Debug, Serialize)] -struct MentionNotification { +struct MentionNotification<'a> { v: u8, pubkey: String, community_host: String, event_id: String, event_kind: i32, event_created_at: i64, + event: &'a nostr::Event, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -105,6 +108,10 @@ trait DeliveryTransport: Sync { #[async_trait::async_trait] trait DeliveryStore: Sync { + async fn load( + &self, + delivery: &buzz_db::operator_listener::ClaimedDelivery, + ) -> Result, String>; async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result; async fn retry(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; @@ -113,10 +120,21 @@ trait DeliveryStore: Sync { struct DbDeliveryStore<'a> { db: &'a buzz_db::Db, + admission: buzz_db::operator_listener::DeliveryAdmission, } #[async_trait::async_trait] impl DeliveryStore for DbDeliveryStore<'_> { + async fn load( + &self, + delivery: &buzz_db::operator_listener::ClaimedDelivery, + ) -> Result, String> { + self.db + .load_operator_listener_delivery_event(delivery, self.admission) + .await + .map_err(|error| error.to_string()) + } + async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result { self.db .release_operator_listener_delivery(id, claim_id, next) @@ -191,7 +209,14 @@ async fn run_delivery_once_with_transport( if claimed.is_empty() { return WorkerIteration::Idle; } - let store = DbDeliveryStore { db: &state.db }; + let store = DbDeliveryStore { + db: &state.db, + admission: buzz_db::operator_listener::DeliveryAdmission { + require_membership: state.config.require_relay_membership, + allow_owner_membership: state.config.allow_nip_oa_auth, + require_allowlist: state.config.pubkey_allowlist_enabled, + }, + }; join_all(claimed.into_iter().map(|delivery| { deliver_one( &state.config.operator_listener_delivery_urls, @@ -233,13 +258,29 @@ async fn deliver_one( .increment(1); return; }; + let stored = match store.load(&delivery).await { + Ok(Some(stored)) if event_visible_to_reader(&stored.event, &delivery.target_pubkey) => { + stored + } + // Retry rather than drop: a mention can commit just before the + // target is added to the channel. + Ok(_) => { + retry_or_fail(store, &delivery, "target cannot read event".to_owned()).await; + return; + } + Err(error) => { + retry_or_fail(store, &delivery, format!("event load failed: {error}")).await; + return; + } + }; let body = match serde_json::to_vec(&MentionNotification { - v: 1, + v: 2, pubkey: hex::encode(&delivery.target_pubkey), community_host: delivery.community_host.clone(), event_id: hex::encode(&delivery.event_id), event_kind: delivery.event_kind, event_created_at: delivery.event_created_at.timestamp(), + event: &stored.event, }) { Ok(body) => body, Err(error) => { @@ -350,7 +391,7 @@ mod tests { use super::*; use buzz_core::tenant::CommunityId; - use nostr::Keys; + use nostr::{EventBuilder, Keys, Kind}; #[test] fn reapers_use_their_expected_intervals() { @@ -369,19 +410,46 @@ mod tests { Fail(Uuid, Uuid), } - #[derive(Default)] struct MockStore { calls: Mutex>, + loaded: Result, String>, + } + + impl Default for MockStore { + fn default() -> Self { + Self::loading(Ok(Some(stored_event(Kind::Custom(9))))) + } } impl MockStore { + fn loading(loaded: Result, String>) -> Self { + Self { + calls: Mutex::default(), + loaded, + } + } + fn calls(&self) -> Vec { self.calls.lock().expect("store calls lock").clone() } } + fn stored_event(kind: Kind) -> buzz_core::StoredEvent { + let event = EventBuilder::new(kind, "hello agent") + .sign_with_keys(&Keys::generate()) + .expect("sign test event"); + buzz_core::StoredEvent::new(event, None) + } + #[async_trait::async_trait] impl DeliveryStore for MockStore { + async fn load( + &self, + _delivery: &buzz_db::operator_listener::ClaimedDelivery, + ) -> Result, String> { + self.loaded.clone() + } + async fn release( &self, id: Uuid, @@ -546,7 +614,9 @@ mod tests { #[tokio::test] async fn successful_delivery_posts_payload_and_completes_claim() { let (routes, relay_keypair, delivery) = delivery_fixture(1); - let store = MockStore::default(); + let loaded = stored_event(Kind::Custom(9)); + let expected = loaded.event.clone(); + let store = MockStore::loading(Ok(Some(loaded))); let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT)); deliver_one( @@ -562,7 +632,7 @@ mod tests { assert_eq!(request.url, routes[&hex::encode(&delivery.listener_pubkey)]); assert!(request.authorization.starts_with("Nostr ")); let body: serde_json::Value = serde_json::from_slice(&request.body).expect("JSON body"); - assert_eq!(body["v"], 1); + assert_eq!(body["v"], 2); assert_eq!(body["pubkey"], hex::encode(&delivery.target_pubkey)); assert_eq!(body["community_host"], delivery.community_host); assert_eq!(body["event_id"], hex::encode(&delivery.event_id)); @@ -571,12 +641,75 @@ mod tests { body["event_created_at"], delivery.event_created_at.timestamp() ); + let event: nostr::Event = + serde_json::from_value(body["event"].clone()).expect("signed event in body"); + assert_eq!(event, expected); + event + .verify() + .expect("delivered event keeps a valid signature"); assert_eq!( store.calls(), [StoreCall::Complete(delivery.id, delivery.claim_id)] ); } + #[tokio::test] + async fn unreadable_event_retries_then_fails_without_http_request() { + // Author-only kinds pass the database check but not the reader check. + for loaded in [None, Some(stored_event(Kind::Custom(30300)))] { + for (attempt, expect_retry) in [ + (1, true), + (buzz_db::operator_listener::MAX_DELIVERY_ATTEMPTS, false), + ] { + let (routes, relay_keypair, delivery) = delivery_fixture(attempt); + let store = MockStore::loading(Ok(loaded.clone())); + let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT)); + + deliver_one( + &routes, + &relay_keypair, + &store, + &transport, + delivery.clone(), + ) + .await; + + assert_eq!(transport.request_count(), 0); + match store.calls().as_slice() { + [StoreCall::Retry(id, claim_id, _)] if expect_retry => { + assert_eq!((*id, *claim_id), (delivery.id, delivery.claim_id)); + } + [StoreCall::Fail(id, claim_id)] if !expect_retry => { + assert_eq!((*id, *claim_id), (delivery.id, delivery.claim_id)); + } + calls => panic!("attempt {attempt}: unexpected store calls {calls:?}"), + } + } + } + } + + #[tokio::test] + async fn event_load_error_retries_without_http_request() { + let (routes, relay_keypair, delivery) = delivery_fixture(1); + let store = MockStore::loading(Err("database unavailable".to_owned())); + let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT)); + + deliver_one( + &routes, + &relay_keypair, + &store, + &transport, + delivery.clone(), + ) + .await; + + assert_eq!(transport.request_count(), 0); + assert!(matches!( + store.calls().as_slice(), + [StoreCall::Retry(id, claim_id, _)] if *id == delivery.id && *claim_id == delivery.claim_id + )); + } + #[tokio::test] async fn rate_limit_and_server_errors_retry_with_exponential_delay() { for status in [