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<RedisNip98ReplayGuard> + Arc<RedisRateLimiter>,
  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 <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

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<buzz_core::TenantContext, (StatusCode, Json<Value>)> {
+    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<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_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<Value> = 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());

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,

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 —

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<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], ()>>,
+    /// 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<RedisNip98ReplayGuard>,
+
+    /// 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<RedisRateLimiter>,
 
     /// 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(