diff --git a/crates/buzz-auth/src/nip98_replay.rs b/crates/buzz-auth/src/nip98_replay.rs index 9910399d2..aece4b0c7 100644 --- a/crates/buzz-auth/src/nip98_replay.rs +++ b/crates/buzz-auth/src/nip98_replay.rs @@ -20,7 +20,7 @@ //! //! ```ignore //! let pubkey = buzz_auth::verify_nip98_event(json, url, method, body)?; -//! if !replay.try_mark(&ctx, &event_id).await? { +//! if !replay.try_mark(&ctx, &event_id, buzz_auth::DEFAULT_REPLAY_TTL_SECS).await? { //! return Err(AuthError::Nip98Replay); //! } //! // safe to honor the request as `pubkey` @@ -30,6 +30,8 @@ //! the window over which a duplicate event id is even plausible is 2×60 = 120s). //! [`DEFAULT_REPLAY_TTL_SECS`] is the floor; deployments may raise it. +use std::{future::Future, pin::Pin}; + use buzz_core::TenantContext; use nostr::EventId; @@ -82,12 +84,12 @@ pub trait Nip98ReplayGuard: Send + Sync { /// The replay window's natural maximum is the verifier's ±60s tolerance; /// values past an hour are implausible and risk Redis `EX` parse failures /// (Redis interprets `EX` as a signed 64-bit integer). - fn try_mark( - &self, - ctx: &TenantContext, - event_id: &EventId, + fn try_mark<'a>( + &'a self, + ctx: &'a TenantContext, + event_id: &'a EventId, ttl_secs: u64, - ) -> impl std::future::Future> + Send; + ) -> Pin> + Send + 'a>>; } /// Redis key for a NIP-98 replay marker: @@ -110,13 +112,13 @@ pub struct AlwaysFreshReplayGuard; #[cfg(any(test, feature = "test-utils"))] impl Nip98ReplayGuard for AlwaysFreshReplayGuard { - async fn try_mark( - &self, - _ctx: &TenantContext, - _event_id: &EventId, + fn try_mark<'a>( + &'a self, + _ctx: &'a TenantContext, + _event_id: &'a EventId, _ttl_secs: u64, - ) -> Result { - Ok(true) + ) -> Pin> + Send + 'a>> { + Box::pin(async { Ok(true) }) } } diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index e8e00176d..550f32d65 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -27,6 +27,7 @@ pub mod cache_invalidation; pub mod error; /// Redis-backed NIP-98 replay seen-set. pub mod nip98_replay; +pub use nip98_replay::RedisNip98ReplayGuard; /// Online/offline presence tracking in Redis. pub mod presence; /// Redis PUBLISH for channel event fan-out. diff --git a/crates/buzz-pubsub/src/nip98_replay.rs b/crates/buzz-pubsub/src/nip98_replay.rs index 79df78db1..b51f1074f 100644 --- a/crates/buzz-pubsub/src/nip98_replay.rs +++ b/crates/buzz-pubsub/src/nip98_replay.rs @@ -33,66 +33,69 @@ impl RedisNip98ReplayGuard { } impl Nip98ReplayGuard for RedisNip98ReplayGuard { - async fn try_mark( - &self, - ctx: &TenantContext, - event_id: &EventId, + fn try_mark<'a>( + &'a self, + ctx: &'a TenantContext, + event_id: &'a EventId, ttl_secs: u64, - ) -> Result { - // §5 gate floor + safety ceiling. Sub-floor values are lifted to the - // floor (contract permits clamping); above-ceiling values are pushed - // down to MAX_REPLAY_TTL_SECS (contract REQUIRES clamping) so a buggy - // caller cannot send a Redis-incompatible `EX` arg or pin a slot for - // implausibly long. - let ttl = ttl_secs.clamp(DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS); + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async move { + // §5 gate floor + safety ceiling. Sub-floor values are lifted to the + // floor (contract permits clamping); above-ceiling values are pushed + // down to MAX_REPLAY_TTL_SECS (contract REQUIRES clamping) so a buggy + // caller cannot send a Redis-incompatible `EX` arg or pin a slot for + // implausibly long. + let ttl = ttl_secs.clamp(DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS); - let mut conn = self.pool.get().await.map_err(|e| { - // Structured field for ops; the user-facing AuthError stays a - // bounded category string. - tracing::warn!( - community = %ctx.community(), - error = %e, - "nip98 replay: redis pool acquire failed — caller MUST fail closed" - ); - AuthError::Internal(format!("Redis pool: {e}")) - })?; - - let key = nip98_replay_key(ctx, event_id); - - // SET key 1 NX EX . redis-rs typed return: Some("OK") on first - // claim, None on existing key. Any other value would be a Redis-side - // bug; treat it as internal error. - let result: Option = redis::cmd("SET") - .arg(&key) - .arg("1") - .arg("NX") - .arg("EX") - .arg(ttl) - .query_async(&mut *conn) - .await - .map_err(|e| { + let mut conn = self.pool.get().await.map_err(|e| { + // Structured field for ops; the user-facing AuthError stays a + // bounded category string. tracing::warn!( community = %ctx.community(), error = %e, - "nip98 replay: redis SET NX EX failed — caller MUST fail closed" + "nip98 replay: redis pool acquire failed — caller MUST fail closed" ); - AuthError::Internal(format!("Redis SET NX EX: {e}")) + AuthError::Internal(format!("Redis pool: {e}")) })?; - match result.as_deref() { - Some("OK") => Ok(true), - None => Ok(false), - Some(other) => { - tracing::error!( - community = %ctx.community(), - reply = %other, - "nip98 replay: redis SET NX EX returned an unexpected reply — investigate" - ); - Err(AuthError::Internal(format!( - "unexpected SET NX EX reply: {other}" - ))) + let key = nip98_replay_key(ctx, event_id); + + // SET key 1 NX EX . redis-rs typed return: Some("OK") on first + // claim, None on existing key. Any other value would be a Redis-side + // bug; treat it as internal error. + let result: Option = redis::cmd("SET") + .arg(&key) + .arg("1") + .arg("NX") + .arg("EX") + .arg(ttl) + .query_async(&mut *conn) + .await + .map_err(|e| { + tracing::warn!( + community = %ctx.community(), + error = %e, + "nip98 replay: redis SET NX EX failed — caller MUST fail closed" + ); + AuthError::Internal(format!("Redis SET NX EX: {e}")) + })?; + + match result.as_deref() { + Some("OK") => Ok(true), + None => Ok(false), + Some(other) => { + tracing::error!( + community = %ctx.community(), + reply = %other, + "nip98 replay: redis SET NX EX returned an unexpected reply — investigate" + ); + Err(AuthError::Internal(format!( + "unexpected SET NX EX reply: {other}" + ))) + } } - } + }) } } diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 47133c8ea..944bcabc7 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -13,6 +13,9 @@ use axum::{ use base64::Engine; use serde_json::Value; +use buzz_auth::{Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS}; +use buzz_core::TenantContext; + use crate::handlers::ingest::{IngestAuth, IngestError}; use crate::state::AppState; @@ -69,27 +72,50 @@ fn verify_bridge_auth( /// Check NIP-98 replay and record the event ID atomically. /// -/// 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 correctness boundary is the shared, community-scoped Redis seen-set on +/// `AppState`, not process-local memory. Any Redis/guard error fails closed: +/// without the shared `SET NX EX` proof, a stateless worker cannot admit the +/// NIP-98 request safely. +async fn check_nip98_replay( state: &AppState, + tenant: &TenantContext, + event_id_bytes: [u8; 32], +) -> Result<(), (StatusCode, Json)> { + check_nip98_replay_with_guard(state.nip98_replay.as_ref(), tenant, event_id_bytes).await +} + +async fn check_nip98_replay_with_guard( + replay_guard: &dyn Nip98ReplayGuard, + tenant: &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_byte_array(event_id_bytes); + match replay_guard + .try_mark(tenant, &event_id, DEFAULT_REPLAY_TTL_SECS) + .await + { + Ok(true) => Ok(()), + Ok(false) => Err(api_error( StatusCode::UNAUTHORIZED, "NIP-98: replay detected", - )); + )), + Err(e) => { + tracing::warn!( + community = %tenant.community(), + error = %e, + "NIP-98 replay guard failed; rejecting request fail-closed" + ); + Err(api_error( + StatusCode::UNAUTHORIZED, + "NIP-98: replay check unavailable", + )) + } } - Ok(()) } /// Reconstruct the canonical URL for NIP-98 verification from the relay config. @@ -188,7 +214,7 @@ pub async fn submit_event( Some(&body), state.config.require_auth_token, )?; - check_nip98_replay(&state, event_id_bytes)?; + check_nip98_replay(&state, &tenant, event_id_bytes).await?; let pubkey_bytes = pubkey.to_bytes().to_vec(); // Enforce relay membership (with NIP-OA fallback via x-auth-tag header). @@ -277,7 +303,7 @@ pub async fn query_events( Some(&body), state.config.require_auth_token, )?; - check_nip98_replay(&state, event_id_bytes)?; + check_nip98_replay(&state, &tenant, 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()); @@ -527,20 +553,6 @@ pub async fn count_events( headers: HeaderMap, body: axum::body::Bytes, ) -> Result, (StatusCode, Json)> { - let url = canonical_url(&state.config.relay_url, "/count"); - let (pubkey, event_id_bytes) = verify_bridge_auth( - &headers, - "POST", - &url, - Some(&body), - state.config.require_auth_token, - )?; - check_nip98_replay(&state, event_id_bytes)?; - 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?; - // Row zero: bind this HTTP request to its community from the request host // before any tenant-scoped read, identical to the WS door in `router.rs` // and `query_events`/`submit_event` above. Fail-closed; never a default @@ -558,6 +570,20 @@ pub async fn count_events( ) })?; + let url = canonical_url(&state.config.relay_url, "/count"); + let (pubkey, event_id_bytes) = verify_bridge_auth( + &headers, + "POST", + &url, + Some(&body), + state.config.require_auth_token, + )?; + check_nip98_replay(&state, &tenant, 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?; + let filters: Vec = serde_json::from_slice(&body) .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?; @@ -1088,6 +1114,57 @@ mod tests { use super::*; use nostr::{Alphabet, EventBuilder, Keys, Kind, SingleLetterTag, Tag}; + fn redis_pool() -> deadpool_redis::Pool { + let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); + deadpool_redis::Config::from_url(url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("create redis pool") + } + + fn fresh_tenant(host: &str) -> TenantContext { + TenantContext::resolved( + buzz_core::CommunityId::from_uuid(uuid::Uuid::new_v4()), + host, + ) + } + + fn fresh_nip98_event_id_bytes() -> [u8; 32] { + EventBuilder::new(Kind::HttpAuth, "") + .sign_with_keys(&Keys::generate()) + .expect("sign auth event") + .id + .to_bytes() + } + + /// Attack 3 proof: two stateless relay pods sharing Redis must share one + /// community-scoped NIP-98 seen-set. Pod A's first claim succeeds; pod B's + /// replay of the same event id in the same community is rejected. The same + /// id in a different community still succeeds, proving the key is scoped by + /// server-resolved tenant rather than global process memory. + #[tokio::test] + #[ignore = "requires Redis"] + async fn nip98_replay_guard_rejects_cross_pod_replay_on_bridge_path() { + let pool = redis_pool(); + let pod_a = buzz_pubsub::RedisNip98ReplayGuard::new(pool.clone()); + let pod_b = buzz_pubsub::RedisNip98ReplayGuard::new(pool); + let tenant_a = fresh_tenant("relay-a.example"); + let tenant_b = fresh_tenant("relay-b.example"); + let event_id_bytes = fresh_nip98_event_id_bytes(); + + check_nip98_replay_with_guard(&pod_a, &tenant_a, event_id_bytes) + .await + .expect("first pod should claim fresh NIP-98 event id"); + + let (status, _) = check_nip98_replay_with_guard(&pod_b, &tenant_a, event_id_bytes) + .await + .expect_err("second pod must reject same-community replay"); + assert_eq!(status, StatusCode::UNAUTHORIZED); + + check_nip98_replay_with_guard(&pod_b, &tenant_b, event_id_bytes) + .await + .expect("same event id in a different community uses a distinct seen-set"); + } + /// Build a kind:30174 engram envelope authored by `agent`, tagged with `owner`. fn engram_envelope(agent: &Keys, owner_hex: &str) -> buzz_core::StoredEvent { let d_tag = Tag::custom( diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index c9a4559bd..ac43e9a3c 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -13,13 +13,13 @@ use tokio_util::sync::CancellationToken; use uuid::Uuid; use buzz_audit::AuditService; -use buzz_auth::AuthService; +use buzz_auth::{AuthService, Nip98ReplayGuard}; use buzz_core::tenant::TenantContext; use buzz_core::CommunityId; use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; -use buzz_pubsub::PubSubManager; +use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard}; use buzz_search::SearchService; use buzz_workflow::WorkflowEngine; use deadpool_redis; @@ -258,9 +258,13 @@ 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>, + /// Shared, community-scoped NIP-98 replay prevention. + /// + /// Correctness boundary for stateless workers: every pod must consult the + /// same Redis `SET NX EX` seen-set, keyed by resolved community. Do not + /// replace this with process-local caching; replay freshness must survive + /// cross-pod routing. + pub nip98_replay: Arc, /// Per-agent sliding-window rate limiter for observer frames (kind 24200). /// Key: agent pubkey bytes (32). Value: (count, window_start). @@ -353,6 +357,8 @@ impl AppState { &config.media.s3_bucket, ) .expect("media storage was already constructed with this S3 config"); + let nip98_replay: Arc = + Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); let state = Self { config: Arc::new(config), db, @@ -399,12 +405,7 @@ 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, observer_rate_limiter: Arc::new(DashMap::new()), mesh_connect_rate_limiter: Arc::new(DashMap::new()), observer_owner_cache: Arc::new(