From fba8a7a94febd4fbfb97df90f7c6be8fb258cf4c Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Thu, 5 Nov 2026 19:46:24 -0500 Subject: [PATCH] feat(relay): log WebSocket EVENT writes on the access target (#9154) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Logs every WebSocket `EVENT` on the access target (`buzz_relay::access`). Before this, accepted writes logged only `Event ingested` on the handler target, and refused or ephemeral writes logged nothing. ### What's logged Each `EVENT` frame writes one `event=write` line when its handler finishes. The line has `conn_id`, `community`, `pubkey` (the authenticated login, if any), `kind`, `event_id`, and `outcome`/`reason`, taken from the OK the relay sent: - `accepted` - `rejected` (INFO), with `reason` set to the OK's NIP-01 prefix (`invalid`, `restricted`, `auth-required`, `duplicate`, `rate-limited`, …). Any other message is logged as `other`. - `error` (WARN), when the OK starts with `error` - `aborted`, when the handler ended before deciding an OK `outcome` is always the relay's decision. If the OK couldn't be queued because the outbound channel was full or closed, the line adds `ok_queued=false`; the field is left off otherwise. `EVENT`s refused before the handler runs are logged too, where their OK is sent: admission rate limits, with admission's reason, and a full handler pool, as `rate-limited`. The OK text after the prefix, the event's content and its tag values are never logged. Lines join to `auth`/`close` lines (and so to `peer_addr`) through `conn_id`, the same way `req` lines do. Rejected writes are capped per connection by the same refusal-budget mechanism as refused reads, on their own counter: the first 16 rejections are logged and the rest are only counted, in a new `writes_suppressed` field on the `close` line. Accepted writes are never capped. ### How `access::WriteLog` sends the OK and records it in one call (`write.ok(...)`). That call replaces every `conn.send(RelayMessage::ok(...))` in `handle_event` and the observer-frame handler, so the line always shows what the client was told. ### Tests - `access::tests`: the reason comes from the OK prefix only; each outcome logs at the right level and without content; rejected lines are capped by the budget and accepted ones are not (draining the channel so every OK is queued); an OK refused by a full or a closed outbound channel keeps its outcome and adds `ok_queued=false`. - `handlers::event::tests::fanout_access`: an unauthenticated `EVENT` going through `handle_event` writes one `rejected`/`auth-required` line, with no content or tag values. --------- Signed-off-by: Will Pfleger Co-authored-by: Duncan --- Justfile | 2 +- crates/buzz-relay/src/access.rs | 252 +++++++++++++++++++++++- crates/buzz-relay/src/connection.rs | 56 +++++- crates/buzz-relay/src/handlers/event.rs | 143 ++++++++------ crates/buzz-relay/src/rejection.rs | 57 +++++- desktop/src-tauri/Cargo.lock | 24 ++- scripts/run-tests.sh | 1 + 7 files changed, 470 insertions(+), 65 deletions(-) diff --git a/Justfile b/Justfile index 48f1079a97..80a6a56af7 100644 --- a/Justfile +++ b/Justfile @@ -558,7 +558,7 @@ test-unit: # subscription::tests covers the in-memory registry and its fan-out # indexes; it is infra-free. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^subscription::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + 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(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + 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::failed_search_closes_without_eose_or_trying_later_filters) + 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::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^subscription::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + 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(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + 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::failed_search_closes_without_eose_or_trying_later_filters) + 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::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (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-relay/src/access.rs b/crates/buzz-relay/src/access.rs index b14ada9e67..2174d0d2d6 100644 --- a/crates/buzz-relay/src/access.rs +++ b/crates/buzz-relay/src/access.rs @@ -13,6 +13,8 @@ //! except public documents and probes; a request that matched no route is //! logged only at status 400 or above, as `route=""`. `route` is the //! matched template, never the raw path. +//! `event=write` lines cover WebSocket EVENT: one per frame, with the OK the +//! relay answered (see [`WriteLog`]). //! A successful bridge read that returned nothing (`result_count=0`) logs at //! DEBUG: polling clients make most of them. WebSocket `req` lines stay at //! INFO, because an empty subscription can still deliver live events. @@ -188,12 +190,13 @@ impl Listed { const REFUSAL_LINES: u64 = 16; /// Per-connection bound on the access lines a client can trigger without -/// being served: reads refused or denied, and AUTH frames after the login was -/// decided. The first [`REFUSAL_LINES`] of each are logged; the rest are +/// being served: reads refused or denied, EVENTs rejected, and AUTH frames +/// after the login was decided. Each kind has its own counter. The first [`REFUSAL_LINES`] of each are logged; the rest are /// counted and reported on the close line. Served reads are never capped. #[derive(Debug, Default)] pub(crate) struct RefusalBudget { reads: AtomicU64, + writes: AtomicU64, auths: AtomicU64, } @@ -203,6 +206,11 @@ impl RefusalBudget { Self::charge(&self.reads) } + /// Whether this rejected EVENT may write its line. + pub(crate) fn write(&self) -> bool { + Self::charge(&self.writes) + } + /// Whether this post-decision AUTH frame may write its line. pub(crate) fn auth(&self) -> bool { Self::charge(&self.auths) @@ -477,6 +485,126 @@ impl Drop for ReadLog { } } +/// The NIP-01 OK prefixes a write's `reason` may take; any other message +/// logs `other`, so the text after the prefix never reaches the log. +const OK_PREFIXES: [&str; 9] = [ + "duplicate", + "pow", + "blocked", + "rate-limited", + "invalid", + "restricted", + "mute", + "error", + "auth-required", +]; + +/// The fixed reason for a refused write: its OK message's NIP-01 prefix. +fn ok_reason(message: &str) -> &'static str { + let prefix = message.split_once(':').map_or(message, |(p, _)| p); + OK_PREFIXES + .into_iter() + .find(|p| *p == prefix) + .unwrap_or("other") +} + +/// One access line per WebSocket EVENT, written when its handler ends and +/// recording the OK the relay answered: `accepted`, `rejected` or `error` +/// (WARN) with the OK prefix as `reason`, or `aborted` if the handler ended +/// before deciding one. `outcome` is the relay's decision; an OK that could not +/// be queued (outbound channel full or closed) adds `ok_queued=false`. +/// Only the event's `kind` and id are logged, never its content or tags. +/// Rejected lines are capped per connection by [`RefusalBudget`], on their own +/// counter, separate from refused reads'. +pub(crate) struct WriteLog { + conn_id: Uuid, + community: CommunityId, + pubkey: Option, + kind: u32, + event_id: String, + /// The OK decided, as (accepted, reason, queued); `None` until one is. + answer: Option<(bool, &'static str, bool)>, + budget: Arc, +} + +impl WriteLog { + pub(crate) fn new(conn: &crate::connection::ConnectionState, event: &nostr::Event) -> Self { + let pubkey = match conn.auth_state_snapshot() { + crate::connection::AuthState::Authenticated(ctx) => Some(ctx.pubkey.to_hex()), + _ => None, + }; + Self { + conn_id: conn.conn_id, + community: conn.tenant.community(), + pubkey, + kind: buzz_core::kind::event_kind_u32(event), + event_id: event.id.to_hex(), + answer: None, + budget: Arc::clone(&conn.refusal_budget), + } + } + + /// Log an EVENT refused before its handler ran, under a fixed `reason`; + /// `queued` is whether its OK was queued. + pub(crate) fn refused( + conn: &crate::connection::ConnectionState, + event: &nostr::Event, + reason: &'static str, + queued: bool, + ) { + Self::new(conn, event).answer = Some((false, reason, queued)); + } + + /// Send the EVENT's OK and record it for the line. + pub(crate) fn ok( + &mut self, + conn: &crate::connection::ConnectionState, + event_id: &str, + accepted: bool, + message: &str, + ) { + let queued = conn.send(crate::protocol::RelayMessage::ok( + event_id, accepted, message, + )); + let reason = if accepted { "" } else { ok_reason(message) }; + self.answer = Some((accepted, reason, queued)); + } +} + +impl Drop for WriteLog { + fn drop(&mut self) { + let (outcome, reason) = match self.answer { + None => ("aborted", None), + Some((true, ..)) => ("accepted", None), + Some((false, "error", _)) => ("error", Some("error")), + Some((false, reason, _)) => ("rejected", Some(reason)), + }; + // Present only when an OK was decided but could not be queued. + let ok_queued = matches!(self.answer, Some((.., false))).then_some(false); + if outcome == "rejected" && !self.budget.write() { + return; + } + let level = if outcome == "error" { + Level::WARN + } else { + Level::INFO + }; + emit!( + level, + event = "write", + conn_id = %self.conn_id, + community = %self.community, + pubkey = self.pubkey.as_deref(), + kind = self.kind, + event_id = self.event_id.as_str(), + outcome, + reason, + ok_queued, + "access" + ); + } +} + /// Which credential an auth attempt presented. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum AuthMethod { @@ -999,6 +1127,7 @@ impl Drop for CloseLine { events_queued = self.queued.events(), bytes_queued = self.queued.bytes(), reads_suppressed = RefusalBudget::close(&self.refusals.reads), + writes_suppressed = RefusalBudget::close(&self.refusals.writes), auth_suppressed = RefusalBudget::close(&self.refusals.auths), "access" ); @@ -1781,6 +1910,125 @@ pub(crate) mod tests { assert!(line.get("pubkey").is_none()); } + /// A rejected write's reason is only its OK message's NIP-01 prefix. + #[test] + fn write_reason_is_the_ok_prefix_only() { + assert_eq!(ok_reason("invalid: bad signature"), "invalid"); + assert_eq!( + ok_reason("auth-required: not authenticated"), + "auth-required" + ); + assert_eq!(ok_reason("rate-limited"), "rate-limited"); + assert_eq!(ok_reason("something: secret"), "other"); + assert_eq!(ok_reason(""), "other"); + } + + /// Each answer logs its outcome and level; no OK logs `aborted`. + #[test] + fn write_lines_record_the_ok_answered() { + let (conn, _rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "secret words") + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let id = event.id.to_hex(); + let (_, out) = capture(|| { + WriteLog::new(&conn, &event).ok(&conn, &id, true, ""); + WriteLog::new(&conn, &event).ok(&conn, &id, false, "duplicate: already have it"); + WriteLog::new(&conn, &event).ok(&conn, &id, false, "error: internal error"); + drop(WriteLog::new(&conn, &event)); + WriteLog::refused(&conn, &event, "messages", true); + }); + assert!(!out.raw().contains("secret"), "{}", out.raw()); + let got: Vec = out + .lines() + .iter() + .map(|l| { + assert_eq!(l["event"], "write", "{l}"); + assert_eq!(l["kind"], 9, "{l}"); + assert_eq!(l["event_id"], id.as_str(), "{l}"); + assert!(l.get("ok_queued").is_none(), "{l}"); + format!("{}/{}/{}", l["outcome"], l["reason"], l["level"]) + }) + .collect(); + assert_eq!( + got, + [ + r#""accepted"/null/"INFO""#, + r#""rejected"/"duplicate"/"INFO""#, + r#""error"/"error"/"WARN""#, + r#""aborted"/null/"INFO""#, + r#""rejected"/"messages"/"INFO""#, + ] + ); + } + + /// Rejected writes past the refusal budget are only counted; accepted + /// writes are never capped. + #[test] + fn rejected_writes_have_a_refusal_budget() { + let (conn, mut rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "") + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let (_, out) = capture(|| { + for _ in 0..REFUSAL_LINES + 3 { + WriteLog::refused(&conn, &event, "messages", true); + WriteLog::new(&conn, &event).ok(&conn, "x", true, ""); + // Drain, so every OK is queued and only the cap is under test. + rx.try_recv().expect("OK queued"); + } + }); + let lines = out.lines(); + assert!( + lines.iter().all(|l| l.get("ok_queued").is_none()), + "{lines:?}" + ); + let count = |o: &str| lines.iter().filter(|l| l["outcome"] == o).count(); + assert_eq!(count("rejected"), REFUSAL_LINES as usize); + assert_eq!(count("accepted"), REFUSAL_LINES as usize + 3); + assert_eq!(RefusalBudget::close(&conn.refusal_budget.writes), 3); + } + + /// An OK the outbound channel could not take (full or closed) keeps the + /// relay's decision as `outcome` and adds `ok_queued=false`. + #[test] + fn write_lines_mark_an_ok_that_could_not_be_queued() { + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "") + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let id = event.id.to_hex(); + + // Full: the test channel holds 4 frames and nothing drains it. + let (full, _rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + while full.send(String::new()) {} + // Closed: the receiver is gone. + let (closed, rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + drop(rx); + + let (_, out) = capture(|| { + WriteLog::new(&full, &event).ok(&full, &id, true, ""); + WriteLog::new(&closed, &event).ok(&closed, &id, false, "invalid: bad"); + WriteLog::refused(&closed, &event, "rate-limited", false); + }); + let got: Vec = out + .lines() + .iter() + .map(|l| format!("{}/{}/{}", l["outcome"], l["reason"], l["ok_queued"])) + .collect(); + assert_eq!( + got, + [ + r#""accepted"/null/false"#, + r#""rejected"/"invalid"/false"#, + r#""rejected"/"rate-limited"/false"#, + ] + ); + } + /// Every read is measured under its type, outcome and reason, including /// the refusals whose lines the budget suppresses. /// Mutation: record after the budget check in `Drop`, or drop the diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index b4fb90270c..a9067da122 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -1372,6 +1372,7 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar } if let Err(reason) = enforce_ws_admission(&msg, &conn, &state).await { + // An EVENT's write line is written where its OK is sent. crate::access::ReadLog::refused(&msg, &conn, reason); return; } @@ -1391,10 +1392,11 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar Err(_) => { // Correlate to the event id: a bare NOTICE here strands the // client's pending publish exactly as an over-quota one did. - conn.send(request_rejection_message( + let queued = conn.send(request_rejection_message( RejectionTarget::Event(event.id), "rate-limited: too many concurrent requests", )); + crate::access::WriteLog::refused(&conn, &event, "rate-limited", queued); return; } }; @@ -2426,6 +2428,58 @@ pub(crate) mod tests { assert_eq!(frame[3], "rate-limited: too many concurrent requests"); } + /// The saturation refusal's write line records whether its OK was queued, + /// with the outbound channel open, full or closed. Mutation: pass `true` + /// instead of the send result at the refusal site → RED. + #[tokio::test] + async fn saturated_handler_write_line_records_whether_its_ok_was_queued() { + let state = crate::state::tests::test_state().await; + let permits = state.handler_semaphore.available_permits(); + let _held = Arc::clone(&state.handler_semaphore) + .acquire_many_owned(permits as u32) + .await + .expect("hold every handler permit"); + let mut got = Vec::new(); + for channel in ["open", "full", "closed"] { + let (conn, rx) = test_conn_with_auth(AuthState::Failed); + let _rx = match channel { + "full" => { + while conn.send(String::new()) {} + Some(rx) + } + "closed" => { + drop(rx); + None + } + _ => Some(rx), + }; + let event = EventBuilder::new(Kind::TextNote, "hello") + .sign_with_keys(&Keys::generate()) + .expect("sign event"); + let raw = serde_json::json!(["EVENT", event]).to_string(); + let (guard, out) = crate::access::tests::capture_access_only(tracing::Level::INFO); + handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await; + drop(guard); + let line = out + .lines() + .into_iter() + .find(|l| l["event"] == "write") + .expect("write line"); + got.push(format!( + "{channel}:{}/{}/{}", + line["outcome"], line["reason"], line["ok_queued"] + )); + } + assert_eq!( + got, + [ + r#"open:"rejected"/"rate-limited"/null"#, + r#"full:"rejected"/"rate-limited"/false"#, + r#"closed:"rejected"/"rate-limited"/false"#, + ] + ); + } + /// The REQ arm of the same branch still settles on CLOSED. #[tokio::test] async fn saturated_handler_rejects_a_req_on_the_closed_channel() { diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 9056dfe416..72a51140b4 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -21,7 +21,6 @@ use buzz_pubsub::EventTopic; use nostr::{Event, PublicKey}; use crate::connection::{AuthState, ConnectionState}; -use crate::protocol::RelayMessage; use crate::state::AppState; use super::ingest::{reject_with_transport, IngestAuth, IngestError}; @@ -656,6 +655,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc { reject("auth"); - conn.send(RelayMessage::ok( + write.ok( + &conn, &event_id_hex, false, "auth-required: not authenticated", - )); + ); return; } } @@ -705,21 +706,23 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc (m, "auth"), }; reject(reason); - conn.send(RelayMessage::ok(&event_id_hex, false, &message)); + write.ok(&conn, &event_id_hex, false, &message); return; } } @@ -753,11 +756,12 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc, state: Arc, state: Arc {} Ok(false) => { reject("restricted"); - conn.send(RelayMessage::ok( + write.ok( + &conn, &event_id_hex, false, "restricted: community writes are fenced", - )); + ); return; } Err(error) => { reject("error"); tracing::warn!(%error, event_id = %event_id_hex, "failed to check ephemeral-event community lifecycle"); - conn.send(RelayMessage::ok( - &event_id_hex, - false, - "error: internal server error", - )); + write.ok(&conn, &event_id_hex, false, "error: internal server error"); return; } } @@ -841,11 +844,12 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc permit, Err(crate::nip_fi_gate::SessionExpired) => { // Fix 4: [FI-TRACE-DENIAL-ORACLE] - conn.send(RelayMessage::ok( + write.ok( + &conn, &event_id_hex, false, "restricted: authorization denied", - )); + ); return; } }; @@ -860,7 +864,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc { - conn.send(RelayMessage::ok(&event_id_hex, true, "")); + write.ok(&conn, &event_id_hex, true, ""); } Err(e) => { // Rejections carry the ingest taxonomy so backend failures @@ -882,7 +886,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc permit, Err(crate::nip_fi_gate::SessionExpired) => { // Fix 4: [FI-TRACE-DENIAL-ORACLE] - conn.send(RelayMessage::ok( + write.ok( + &conn, &event_id_hex, false, "restricted: authorization denied", - )); + ); return; } }; @@ -938,11 +943,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc { let e = e.into_legacy(); @@ -958,7 +959,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc, + write: &mut crate::access::WriteLog, ) { let event_clone = event.clone(); let verify_result = tokio::task::spawn_blocking(move || verify_event(&event_clone)).await; match verify_result { Ok(Ok(())) => {} Ok(Err(e)) => { - conn.send(RelayMessage::ok( - event_id_hex, - false, - &format!("invalid: {e}"), - )); + write.ok(&conn, event_id_hex, false, &format!("invalid: {e}")); return; } Err(_) => { - conn.send(RelayMessage::ok( - event_id_hex, - false, - "error: internal error", - )); + write.ok(&conn, event_id_hex, false, "error: internal error"); return; } } @@ -1349,11 +1343,12 @@ async fn handle_agent_observer_event( let now = chrono::Utc::now().timestamp(); let event_ts = event.created_at.as_secs() as i64; if (event_ts - now).unsigned_abs() > 300 { - conn.send(RelayMessage::ok( + write.ok( + &conn, event_id_hex, false, "invalid: observer frame timestamp outside ±5 minute freshness window", - )); + ); return; } @@ -1361,12 +1356,12 @@ async fn handle_agent_observer_event( Ok(Some(route)) => route, Ok(None) => { // Unknown frame value — silently drop, no error to publisher. - conn.send(RelayMessage::ok(event_id_hex, true, "")); + write.ok(&conn, event_id_hex, true, ""); return; } Err(message) => { reject("invalid"); - conn.send(RelayMessage::ok(event_id_hex, false, &message)); + write.ok(&conn, event_id_hex, false, &message); return; } }; @@ -1405,11 +1400,7 @@ async fn handle_agent_observer_event( } Err(e) => { warn!(conn_id = %conn_id, event_id = %event_id_hex, "agent observer owner check failed: {e}"); - conn.send(RelayMessage::ok( - event_id_hex, - false, - "error: internal server error", - )); + write.ok(&conn, event_id_hex, false, "error: internal server error"); return; } } @@ -1418,11 +1409,12 @@ async fn handle_agent_observer_event( }; if !is_owner { reject("auth"); - conn.send(RelayMessage::ok( + write.ok( + &conn, event_id_hex, false, "restricted: observer frame is not authorized for this agent owner", - )); + ); return; } @@ -1432,11 +1424,12 @@ async fn handle_agent_observer_event( if matches!(route.direction, AgentObserverDirection::Telemetry) { let agent_key: [u8; 32] = agent_bytes.as_slice().try_into().unwrap_or([0u8; 32]); if observer_frame_rate_limited(&state, conn.tenant.community(), agent_key) { - conn.send(RelayMessage::ok( + write.ok( + &conn, event_id_hex, false, "rate-limited: observer frame rate exceeded (100/sec per agent)", - )); + ); return; } } @@ -1463,7 +1456,7 @@ async fn handle_agent_observer_event( ); fan_out_event_to_local_subscribers(&state, conn.tenant.community(), &stored_event).await; - conn.send(RelayMessage::ok(event_id_hex, true, "")); + write.ok(&conn, event_id_hex, true, ""); } fn agent_observer_route(event: &Event) -> Result, String> { @@ -1831,12 +1824,14 @@ mod tests { ), }); + let mut write = crate::access::WriteLog::new(&conn, &event); super::handle_agent_observer_event( event.clone(), conn.conn_id, &event.id.to_hex(), conn, state, + &mut write, ) .await; @@ -3486,6 +3481,38 @@ mod tests { "membership lookup" ); } + + /// A refused WebSocket EVENT writes one `write` access line with the + /// OK's fixed prefix as its reason, and never the event's content or + /// tag values. + #[tokio::test] + async fn refused_ws_event_writes_one_access_line_without_its_content() { + let state = test_state().await; + let (conn, mut send_rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + let event = EventBuilder::new(Kind::Custom(9), "secret words") + .tag(nostr::Tag::parse(["h", "secret-channel"]).unwrap()) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let (guard, out) = crate::access::tests::capture_scoped(); + super::super::handle_event(event.clone(), conn, state).await; + drop(guard); + + let axum::extract::ws::Message::Text(ack) = send_rx.try_recv().expect("OK") else { + panic!("expected text OK"); + }; + assert!(ack.contains("auth-required"), "{ack}"); + assert!(!out.raw().contains("secret"), "{}", out.raw()); + let lines = out.lines(); + assert_eq!(lines.len(), 1, "{lines:?}"); + let line = &lines[0]; + assert_eq!(line["event"], "write", "{line}"); + assert_eq!(line["outcome"], "rejected", "{line}"); + assert_eq!(line["reason"], "auth-required", "{line}"); + assert_eq!(line["kind"], 9, "{line}"); + assert_eq!(line["event_id"], event.id.to_hex(), "{line}"); + assert!(line.get("pubkey").is_none(), "{line}"); + } } // --------------------------------------------------------------------- diff --git a/crates/buzz-relay/src/rejection.rs b/crates/buzz-relay/src/rejection.rs index 72b228e97e..ee0bfd9998 100644 --- a/crates/buzz-relay/src/rejection.rs +++ b/crates/buzz-relay/src/rejection.rs @@ -145,10 +145,13 @@ async fn send_admission_result( crate::handlers::close::refuse(sub_id, &reason, conn, state).await; } _ => { - conn.send(request_rejection_message( + let queued = conn.send(request_rejection_message( rejection_target_for(msg), &reason, )); + if let ClientMessage::Event(event) = msg { + crate::access::WriteLog::refused(conn, event, logged, queued); + } } } Err(logged) @@ -263,6 +266,58 @@ mod tests { assert_eq!(frame[2], false); } + /// The write line's `ok_queued` for an over-quota EVENT, with the outbound + /// channel `open`, full or closed: absent when the OK was queued, `false` + /// when it was not. Mutation: pass `true` instead of the send result at + /// the refusal site → RED. + #[tokio::test] + async fn admission_write_line_records_whether_its_ok_was_queued() { + let state = crate::state::tests::test_state().await; + let mut got = Vec::new(); + for channel in ["open", "full", "closed"] { + let (conn, rx) = test_conn(); + let _rx = match channel { + "full" => { + while conn.send(String::new()) {} + Some(rx) + } + "closed" => { + drop(rx); + None + } + _ => Some(rx), + }; + let (msg, _) = parsed_event_message(); + let (guard, out) = crate::access::tests::capture_access_only(tracing::Level::INFO); + let _ = send_admission_result( + &conn, + &state, + Err(AdmissionError::Unavailable), + &msg, + "ws_operations", + ) + .await; + drop(guard); + let line = out + .lines() + .into_iter() + .find(|l| l["event"] == "write") + .expect("write line"); + got.push(format!( + "{channel}:{}/{}", + line["outcome"], line["ok_queued"] + )); + } + assert_eq!( + got, + [ + r#"open:"rejected"/null"#, + r#"full:"rejected"/false"#, + r#"closed:"rejected"/false"# + ] + ); + } + /// A REQ still settles on CLOSED, which carries the subscription id. This /// pins the pre-existing behavior the fix must not disturb. #[tokio::test] diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index 18a8845752..f1aabae104 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -1240,6 +1240,7 @@ dependencies = [ "tracing", "ulid", "uuid", + "webp", ] [[package]] @@ -2396,7 +2397,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.118", ] [[package]] @@ -4978,6 +4979,16 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "libwebp-sys" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54cd30df7c7165ce74a456e4ca9732c603e8dc5e60784558c1c6dc047f876733" +dependencies = [ + "cc", + "glob", +] + [[package]] name = "line-clipping" version = "0.3.7" @@ -11132,7 +11143,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -12632,6 +12643,15 @@ dependencies = [ "system-deps", ] +[[package]] +name = "webp" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c071456adef4aca59bf6a583c46b90ff5eb0b4f758fc347cea81290288f37ce1" +dependencies = [ + "libwebp-sys", +] + [[package]] name = "webpki-root-certs" version = "1.0.10" diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index dc87f85af7..b872ac1900 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -333,6 +333,7 @@ run_unit_tests() { connection::tests::dispatcher_counts_each_req_and_count_frame_once \ connection::tests::read_before_login_logs_one_denial_from_the_handler \ connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial \ + connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued \ connection::tests::read_refused_by_unavailable_admission_logs_one_denial \ handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial \ nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line \