fix(relay): share NIP-98 replay guard via Redis

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

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<Output = Result<bool, AuthError>> + Send;
+    ) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + 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<bool, AuthError> {
-        Ok(true)
+    ) -> Pin<Box<dyn Future<Output = Result<bool, AuthError>> + Send + 'a>> {
+        Box::pin(async { Ok(true) })
     }
 }
 

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.

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<bool, AuthError> {
-        // §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<Box<dyn std::future::Future<Output = Result<bool, AuthError>> + 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 <ttl>. 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<String> = 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 <ttl>. 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<String> = 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}"
+                    )))
+                }
             }
-        }
+        })
     }
 }
 

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<Value>)> {
+    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<Value>)> {
     // 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<Json<Value>, (StatusCode, Json<Value>)> {
-    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<nostr::Filter> = 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(

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<AtomicBool>,
     /// 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<moka::sync::Cache<[u8; 32], ()>>,
+    /// 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<dyn Nip98ReplayGuard>,
 
     /// 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<dyn Nip98ReplayGuard> =
+            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(