From ed33878b7dd1cf8f3d4d5208ae4917b50ff6105e Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Fri, 26 Jun 2026 13:29:28 -0400 Subject: [PATCH] feat(relay): wire shared NIP-98 replay guard + per-IP connection fence Land the auth fences from buzz-auth/buzz-pubsub into the relay's request paths, replacing the per-pod moka NIP-98 cache that does not carry the freshness proof under any-pod-any-connection (bus-scoping B). - AppState: add Arc + Arc, constructed from the existing redis_pool. Drop the dead nip98_seen moka cache (its only caller was the bridge replay check). - bridge.rs: check_nip98_replay is now async + tenant-scoped, calling the shared seen-set's try_mark under the resolved community (the seen-set is community-scoped per S1). Resolve the tenant BEFORE the replay check in all three NIP-98 handlers; DRY the duplicated host-resolve into resolve_request_tenant. /count now resolves a tenant (required for the per-community replay check). Replay and invalid responses are wire-indistinguishable (Quinn P2) and fail closed on Redis error. - router.rs: per-IP connection fence (check_ip_connection) runs in the WS upgrade path BEFORE host resolution, so an unmappable Host cannot bypass the cap. Operator-global, tenant-free; fail-closed -> 429. - config.rs: BUZZ_MAX_CONNECTIONS_PER_IP (default 60) + BUZZ_IP_CONNECTION_WINDOW_SECS (default 60). cargo test -p buzz-relay: 378 passed / 0 failed. clippy clean (one pre-existing unrelated warning in side_effects.rs). Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-relay/src/api/bridge.rs | 104 +++++++++++++++++----------- crates/buzz-relay/src/config.rs | 20 ++++++ crates/buzz-relay/src/router.rs | 20 ++++++ crates/buzz-relay/src/state.rs | 27 +++++--- 4 files changed, 122 insertions(+), 49 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 3c73d0ac0..86e0feac0 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -11,6 +11,7 @@ use axum::{ response::Json, }; use base64::Engine; +use buzz_auth::Nip98ReplayGuard; use serde_json::Value; use crate::handlers::ingest::{IngestAuth, IngestError}; @@ -67,29 +68,65 @@ fn verify_bridge_auth( Err(api_error(StatusCode::UNAUTHORIZED, "missing Nostr auth")) } -/// Check NIP-98 replay and record the event ID atomically. +/// Resolve the request tenant from the connection host. /// -/// Uses moka's `entry` API for atomic insert-if-absent — no race window -/// between "check if seen" and "mark as seen". -fn check_nip98_replay( +/// The HTTP twin of the WS upgrade bind: community comes from the connection +/// host, never the authenticated key. An unmapped host fails closed with +/// `404` so the bridge never reveals whether a host maps to a community. +async fn resolve_request_tenant( state: &AppState, + headers: &HeaderMap, +) -> Result)> { + let host = crate::router::normalize_host( + headers, + &crate::api::nip05::extract_domain(&state.config.relay_url), + ); + state + .resolve_tenant(&host) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "not found")) +} + +/// Check NIP-98 replay and record the event ID atomically, scoped to `ctx`'s +/// community. +/// +/// Backed by the shared Redis seen-set (`SET NX EX`): under +/// any-pod-any-connection the freshness fence must be cluster-wide, not +/// per-pod. The seen-set is community-scoped (S1) — the same event id is a +/// distinct claim in each community — so this MUST run under the resolved +/// tenant. Fails closed: a Redis error rejects the request rather than +/// admitting a possible replay. Dev-mode X-Pubkey auth (zero hash) skips the +/// check. +async fn check_nip98_replay( + state: &AppState, + ctx: &buzz_core::TenantContext, event_id_bytes: [u8; 32], ) -> Result<(), (StatusCode, Json)> { // Skip replay detection for dev-mode X-Pubkey auth (zero hash). if event_id_bytes == [0u8; 32] { return Ok(()); } - // Atomic: get_with inserts the value if absent and returns it. - // If the entry already existed, this is a replay. - let entry = state.nip98_seen.entry(event_id_bytes); - let result = entry.or_insert(()); - if !result.is_fresh() { - return Err(api_error( + let event_id = nostr::EventId::from_slice(&event_id_bytes) + .map_err(|_| api_error(StatusCode::UNAUTHORIZED, "NIP-98: invalid event id"))?; + match state + .nip98_replay + .try_mark(ctx, &event_id, buzz_auth::DEFAULT_REPLAY_TTL_SECS) + .await + { + Ok(true) => Ok(()), + // Wire-indistinguishable from an invalid NIP-98 event (Quinn P2): a + // distinct "replay detected" reply would turn the community-scoped + // seen-set into a presence oracle on event ids. + Ok(false) => Err(api_error( StatusCode::UNAUTHORIZED, - "NIP-98: replay detected", - )); + "NIP-98 verification failed", + )), + // Fail closed — never admit on a Redis error. + Err(_) => Err(api_error( + StatusCode::UNAUTHORIZED, + "NIP-98 verification failed", + )), } - Ok(()) } /// Reconstruct the canonical URL for NIP-98 verification from the relay config. @@ -171,7 +208,12 @@ pub async fn submit_event( Some(&body), state.config.require_auth_token, )?; - check_nip98_replay(&state, event_id_bytes)?; + // Resolve the tenant from the request host (the HTTP twin of the WS upgrade + // bind) BEFORE the replay check: the seen-set is community-scoped, so the + // replay mark must run under the resolved tenant. An unmapped host fails + // closed here. + let ctx = resolve_request_tenant(&state, &headers).await?; + check_nip98_replay(&state, &ctx, event_id_bytes).await?; let pubkey_bytes = pubkey.to_bytes().to_vec(); // Enforce relay membership (with NIP-OA fallback via x-auth-tag header). @@ -181,19 +223,6 @@ pub async fn submit_event( let event: nostr::Event = serde_json::from_slice(&body) .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid event JSON: {e}")))?; - // Resolve the tenant from the request host (the HTTP twin of the WS - // upgrade bind): community comes from the connection host, never the - // authenticated key. An unmapped host fails closed. Resolved before any - // event routing so both the mesh and ingest paths bind the same tenant. - let host = crate::router::normalize_host( - &headers, - &crate::api::nip05::extract_domain(&state.config.relay_url), - ); - let ctx = state - .resolve_tenant(&host) - .await - .map_err(|_| api_error(StatusCode::NOT_FOUND, "not found"))?; - // Mesh signaling kinds (24620 status report, 24621 connect request) are // ephemeral and deliberately absent from ingest_event's per-kind allowlist. // The desktop's Rust coordinator publishes them via this bridge, so route @@ -255,23 +284,14 @@ pub async fn query_events( Some(&body), state.config.require_auth_token, )?; - check_nip98_replay(&state, event_id_bytes)?; + // Resolve the tenant before the replay check (community-scoped seen-set). + let ctx = resolve_request_tenant(&state, &headers).await?; + check_nip98_replay(&state, &ctx, event_id_bytes).await?; let pubkey_bytes = pubkey.to_bytes().to_vec(); let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok()); super::relay_members::enforce_relay_membership(&state, &pubkey_bytes, auth_tag).await?; - // Resolve the tenant from the request host (community from the connection, - // never the authenticated key); unmapped host fails closed. - let host = crate::router::normalize_host( - &headers, - &crate::api::nip05::extract_domain(&state.config.relay_url), - ); - let ctx = state - .resolve_tenant(&host) - .await - .map_err(|_| api_error(StatusCode::NOT_FOUND, "not found"))?; - // Two-pass parse: preserve raw JSON for custom extension fields (before_id, // depth_limit, feed_types) that nostr::Filter silently drops. let raw_filters: Vec = serde_json::from_slice(&body) @@ -519,7 +539,11 @@ pub async fn count_events( Some(&body), state.config.require_auth_token, )?; - check_nip98_replay(&state, event_id_bytes)?; + // Resolve the tenant before the replay check (community-scoped seen-set); + // unmapped host fails closed. /count previously skipped tenant resolution — + // it's required now because the NIP-98 seen-set is per-community. + let ctx = resolve_request_tenant(&state, &headers).await?; + check_nip98_replay(&state, &ctx, event_id_bytes).await?; let pubkey_bytes = pubkey.to_bytes().to_vec(); let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok()); diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 248e3f419..fcf0b8aa1 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -39,6 +39,12 @@ pub struct Config { pub relay_url: String, /// Maximum number of concurrent WebSocket connections. pub max_connections: usize, + /// Maximum WebSocket connection attempts permitted per source IP within + /// `ip_connection_window_secs`. Operator-global fence (pre-tenant) — runs + /// before host resolution so an unmappable Host can't bypass the cap. + pub max_connections_per_ip: u64, + /// Sliding-window length (seconds) for the per-IP connection fence. + pub ip_connection_window_secs: u64, /// Maximum number of concurrently executing message handlers. pub max_concurrent_handlers: usize, /// Per-connection outbound message buffer size (number of messages). @@ -157,6 +163,18 @@ impl Config { .and_then(|v| v.parse().ok()) .unwrap_or(10_000); + let max_connections_per_ip = std::env::var("BUZZ_MAX_CONNECTIONS_PER_IP") + .ok() + .and_then(|v| v.parse().ok()) + .filter(|&v| v > 0) + .unwrap_or(60); + + let ip_connection_window_secs = std::env::var("BUZZ_IP_CONNECTION_WINDOW_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .filter(|&v| v > 0) + .unwrap_or(60); + let max_concurrent_handlers = std::env::var("BUZZ_MAX_CONCURRENT_HANDLERS") .ok() .and_then(|v| v.parse().ok()) @@ -377,6 +395,8 @@ impl Config { typesense_key, relay_url, max_connections, + max_connections_per_ip, + ip_connection_window_secs, max_concurrent_handlers, send_buffer_size, max_frame_bytes, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index d84c592bd..94315c97f 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -23,6 +23,7 @@ use crate::connection::handle_connection; use crate::metrics::track_metrics; use crate::nip11::{nip11_facts, relay_info_handler, RelayInfo}; use crate::state::AppState; +use buzz_auth::RateLimiter; /// Build the axum [`Router`] with all relay routes, middleware, and CORS configuration. /// @@ -162,6 +163,25 @@ async fn nip11_or_ws_handler( match WebSocketUpgrade::from_request(req, &state).await { Ok(ws) => { + // Operator-global IP connection fence — runs BEFORE host resolution + // so an attacker can't bypass the per-IP cap by sending an + // unmappable Host (every upgrade attempt counts, mapped or not). + // Tenant-free by construction (`check_ip_connection`). Fail-closed: + // a Redis error rejects rather than admits. + match state + .rate_limiter + .check_ip_connection( + &addr.ip(), + state.config.ip_connection_window_secs, + state.config.max_connections_per_ip, + ) + .await + { + Ok(result) if result.allowed => {} + Ok(_) => return StatusCode::TOO_MANY_REQUESTS.into_response(), + Err(_) => return StatusCode::TOO_MANY_REQUESTS.into_response(), + } + // Conformance row-zero: resolve the tenant from the connection host // BEFORE upgrading, so a connection that upgrades already carries a // resolved `TenantContext`. An unmapped host is rejected fail-closed — diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index de763e2bb..354aa6df5 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -18,6 +18,8 @@ use buzz_core::event::StoredEvent; use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; +use buzz_pubsub::nip98_replay::RedisNip98ReplayGuard; +use buzz_pubsub::rate_limiter::RedisRateLimiter; use buzz_pubsub::PubSubManager; use buzz_search::SearchService; use buzz_workflow::WorkflowEngine; @@ -247,9 +249,18 @@ pub struct AppState { pub shutting_down: Arc, /// Process start time — used by `/_status` endpoint. pub started_at: Instant, - /// NIP-98 replay prevention: recently-seen event IDs. - /// 2× the ±60s tolerance window so entries outlive the acceptance window. - pub nip98_seen: Arc>, + /// NIP-98 replay prevention: shared, community-scoped seen-set backed by + /// Redis (`SET NX EX`). Replaces the old per-pod moka cache — under + /// any-pod-any-connection (bus-scoping B), a NIP-98 mint can land on any + /// pod, so the freshness fence must be cluster-wide. Fail-closed on Redis + /// error (see `Nip98ReplayGuard::try_mark`). + pub nip98_replay: Arc, + + /// Operator-global per-IP connection fence + per-(community, pubkey) + /// rate limiter, backed by Redis. `check_ip_connection` is tenant-free and + /// runs before host resolution; `check_and_increment` scopes by resolved + /// community. + pub rate_limiter: Arc, /// Per-agent sliding-window rate limiter for observer frames (kind 24200). /// Key: agent pubkey bytes (32). Value: (count, window_start). @@ -357,6 +368,8 @@ impl AppState { &config.media.s3_bucket, ) .expect("media storage was already constructed with this S3 config"); + let nip98_replay = Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); + let rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone())); let state = Self { config: Arc::new(config), db, @@ -404,12 +417,8 @@ impl AppState { audio_rooms: Arc::new(AudioRoomManager::new()), shutting_down: Arc::new(AtomicBool::new(false)), started_at: Instant::now(), - nip98_seen: Arc::new( - moka::sync::Cache::builder() - .max_capacity(10_000) - .time_to_live(std::time::Duration::from_secs(120)) - .build(), - ), + nip98_replay, + rate_limiter, observer_rate_limiter: Arc::new(DashMap::new()), mesh_connect_rate_limiter: Arc::new(DashMap::new()), observer_owner_cache: Arc::new(