Give each sender its own traversal offer allowance, and enforce the freshness bound at load

The incoming-offer semaphore was global with no per-sender accounting and
the permit was taken before any identity check, so one sender could hold
every slot and deny rendezvous to everyone else. Each sender now has its
own allowance with the global count kept as the outer bound. Note what
this does and does not do: it raises the cost from one keypair to a small
number of them, so a sender willing to spend throwaway identities can
still saturate the pool at unchanged total offer rate.

The signal freshness bound is only sound while the acceptance window
stays strictly inside the replay window, or an offer evicted from the
replay cache is still fresh enough to be accepted twice. The relation was
stated in a comment and enforced nowhere. Config validation now rejects
the bad combination at load, derived from the skew constant rather than a
literal, and checked regardless of whether the feature is enabled so that
turning it on later cannot surface an error at a surprising moment. The
NAT lab config generator produced a combination the new rule rejects and
is corrected in the same change.

The punch-target filter shipped with no test that would fail if it were
reverted. Loopback, link-local and multicast candidates and an oversized
list are now covered, and the cap assertion is tightened from a bound to
an equality. The private-range inclusion that LAN traversal depends on is
pinned as a healthy path so a blanket ban cannot pass.

Green: fmt, build, clippy and test --lib, 1533 passed.

CHANGELOG.md

@@ -24,8 +24,50 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   burst or a non-positive rate is rejected at config validation rather than
   silently refusing all rekey traffic.
 
+- `node.discovery.nostr.max_concurrent_offers_per_npub`, defaulting to 4, which
+  bounds how many inbound traversal offers one sender npub may have in flight
+  at once. It sits inside `max_concurrent_incoming_offers`, which remains the
+  outer bound, so a value above that is inert; zero is rejected at config
+  validation, since it refuses every inbound offer rather than disabling the
+  limit. Existing configurations parse unchanged, the key being optional.
+
 ### Changed
 
+- Inbound traversal offers are now admitted against a per-sender allowance as
+  well as the global pool. The intake path previously took a permit from a
+  single semaphore before any identity check, with the sender's npub used only
+  as a log field, so one sender could hold every slot and deny traversal
+  onboarding to every other peer for as long as it kept offering. Admission now
+  takes a per-npub permit and a global permit together. A sender over its own
+  allowance is refused at debug rather than warn, because the party tripping it
+  is by definition sending faster than the node wants and a record per
+  rejection would turn the spam into log volume; the global bound being reached
+  keeps its warn, which is the operator's signal that the node is genuinely
+  saturated. **This does not make the pool inexhaustible.** Nostr identities
+  cost nothing to generate and the signal subscription carries no author
+  restriction, so an attacker running four throwaway npubs still saturates the
+  shipped 16-slot pool at an unchanged total offer rate. What the change buys
+  is that one identity can no longer do it alone, and that the two refusals are
+  distinguishable in the log. The permit is still held across the whole
+  attempt; that duration remains inferred from the attempt timeout rather than
+  measured.
+
+- Config validation now rejects a `node.discovery.nostr.signal_ttl_secs` that
+  is too large for the configured `replay_window_secs`. A traversal signal is
+  acceptable over its TTL plus 60s of clock-skew grace on each side, and that
+  span has to stay strictly inside the replay window, or a session id evicted
+  from the replay cache on expiry is still fresh enough to be accepted a second
+  time. The relation was documented but unenforced, so raising the TTL past
+  180s silently voided it. The bound is derived from the skew constant rather
+  than restated, and is checked whether or not nostr discovery is enabled, for
+  the same reason the rekey rules are. The shipped defaults (120s against 300s)
+  are unaffected, but a configuration that had widened the TTL or narrowed the
+  replay window now fails to load, with an error naming the concrete floor for
+  `replay_window_secs`. The NAT lab's config generator was one such
+  configuration and its generated `replay_window_secs` moves from 60 to 180.
+  Note that this covers eviction on expiry only: `seen_sessions_max_entries`
+  remains a separate capacity-eviction route that no config relation bounds.
+
 - Config validation now rejects two `node.rekey` settings that appear to
   disable the trigger and in fact fire it continuously. `after_messages` of
   zero makes the message-count arm true on every poll, because the trigger

docs/design/fips-nostr-discovery.md

@@ -212,11 +212,20 @@ NIP-17 DM relay list (kind 10050), and falls back to `dm_relays` if
 the inbox-relays fetch fails. Each side also publishes its own inbox
 relay list on startup so dialers can discover it.
 
-On the receiving side, an inbound semaphore bounds concurrent offer
-processing at `max_concurrent_incoming_offers`. When the semaphore is
-full, the offer is dropped with a warn log; this is the primary guard
-against offer-spam from a misbehaving or compromised relay. A
-`sessionId` replay cache (bounded by `seen_sessions_max_entries`, with
+On the receiving side, admission is a pair of bounds taken together: a
+per-sender allowance of `max_concurrent_offers_per_npub`, keyed on the
+npub that signed the gift wrap, nested inside a global
+`max_concurrent_incoming_offers`. A sender over its own allowance is
+refused at debug, since by definition it is sending faster than the node
+wants and a record per rejection would turn the spam into log volume; the
+global bound being reached is the operator-visible warn, because that one
+says the node is genuinely saturated. Together they keep one identity
+from holding the whole pool. They do not make the pool inexhaustible:
+nostr identities are free to generate, so an attacker running
+`ceil(max_concurrent_incoming_offers / max_concurrent_offers_per_npub)`
+throwaway npubs still saturates it at the same total offer rate. Raising
+the attacker's cost beyond keypairs would mean pricing the offer itself.
+A `sessionId` replay cache (bounded by `seen_sessions_max_entries`, with
 entries valid for `replay_window_secs`) rejects duplicates.
 
 The responder runs its own STUN query and replies with a
@@ -306,6 +315,7 @@ machinery:
 | Mechanism | Default | What it prevents | Behavior at limit |
 | --- | --- | --- | --- |
 | Offer semaphore (`max_concurrent_incoming_offers`) | 16 | CPU and memory exhaustion from offer spam on DM relays. | Warn log, offer dropped. |
+| Per-npub offer allowance (`max_concurrent_offers_per_npub`) | 4 | One sender identity holding every offer slot and denying traversal onboarding to everyone else. Does not prevent the same denial from several throwaway npubs. | Debug log, offer dropped. |
 | Advert cache (`advert_cache_max_entries`) | 2048 | Memory growth from ambient advert traffic under `policy: open`. | LRU-by-expiry eviction. |
 | Seen-sessions (`seen_sessions_max_entries`) | 2048 | Replay of stale `sessionId` values. | Oldest entry evicted. |
 | Signal TTL (`signal_ttl_secs`) | 120 s | Indefinite in-flight offers on relays. | Expired offers rejected at validation. |

docs/reference/configuration.md

@@ -207,6 +207,7 @@ inert otherwise.
 | `node.discovery.nostr.policy` | string | `"configured_only"` | Advert discovery policy: `disabled`, `configured_only`, `open` |
 | `node.discovery.nostr.open_discovery_max_pending` | usize | `64` | Max open-discovery peers queued in outbound retry/connection state at once |
 | `node.discovery.nostr.max_concurrent_incoming_offers` | usize | `16` | Max concurrent inbound traversal offers processed at once (rate limit against offer spam) |
+| `node.discovery.nostr.max_concurrent_offers_per_npub` | usize | `4` | Max concurrent inbound traversal offers accepted from any one sender npub, so a single identity cannot hold the whole pool. Sits inside `max_concurrent_incoming_offers`, which stays the outer bound; a larger value is inert. Zero is rejected, since it refuses every inbound offer rather than disabling the limit |
 | `node.discovery.nostr.advert_cache_max_entries` | usize | `2048` | Max cached overlay adverts retained from relay traffic |
 | `node.discovery.nostr.seen_sessions_max_entries` | usize | `2048` | Max seen-session IDs retained for replay detection |
 | `node.discovery.nostr.advertise` | bool | `true` | Publish local endpoint adverts |

src/config/mod.rs

@@ -25,6 +25,7 @@ mod node;
 mod peer;
 mod transport;
 
+use crate::discovery::nostr::FRESHNESS_SKEW_TOLERANCE_MS;
 use crate::node::REKEY_JITTER_SECS;
 use crate::upper::config::{DnsConfig, TunConfig};
 use crate::{Identity, IdentityError};
@@ -972,6 +973,48 @@ impl Config {
             )));
         }
 
+        // The freshness window backstops session-id replay protection: an
+        // offer evicted from the replay cache must already be too old to pass
+        // the freshness check, or it can be accepted a second time. A signal
+        // is acceptable over `signal_ttl_secs` plus the skew tolerance on each
+        // side, so that span has to stay strictly inside `replay_window_secs`.
+        // Checked regardless of `nostr.enabled`, for the reason the rekey
+        // block above gives: enabling the feature later must not surface a
+        // config error at a surprising moment.
+        let skew_secs = FRESHNESS_SKEW_TOLERANCE_MS / 1000;
+        let freshness_window_secs = nostr.signal_ttl_secs.saturating_add(2 * skew_secs);
+        if freshness_window_secs >= nostr.replay_window_secs {
+            return Err(ConfigError::Validation(format!(
+                "`node.discovery.nostr.signal_ttl_secs` is {}, which with {skew_secs}s of clock-skew grace on each side makes a traversal signal acceptable over a {freshness_window_secs}s window, \
+                 but `node.discovery.nostr.replay_window_secs` is {}. \
+                 The freshness window must be strictly narrower than the replay window, or a session id evicted from the replay cache is still fresh enough to be accepted a second time. \
+                 Raise `replay_window_secs` above {freshness_window_secs}, or lower `signal_ttl_secs`.",
+                nostr.signal_ttl_secs, nostr.replay_window_secs
+            )));
+        }
+
+        // Zero here is the same trap as `established_handshake_burst`: it
+        // reads as "no limit" and in fact refuses every inbound offer. The
+        // upper bound exists because the per-npub semaphore is built lazily
+        // inside the intake path rather than at startup, so an oversized
+        // value would panic there instead of failing loudly at load.
+        if nostr.max_concurrent_offers_per_npub == 0 {
+            return Err(ConfigError::Validation(
+                "`node.discovery.nostr.max_concurrent_offers_per_npub` is 0, which refuses every inbound traversal offer rather than disabling the per-sender limit. \
+                 Omit the key for the default, or set a positive allowance; `max_concurrent_incoming_offers` remains the outer bound."
+                    .to_string(),
+            ));
+        }
+
+        if nostr.max_concurrent_offers_per_npub > tokio::sync::Semaphore::MAX_PERMITS {
+            return Err(ConfigError::Validation(format!(
+                "`node.discovery.nostr.max_concurrent_offers_per_npub` is {}, which exceeds the maximum {} permits a semaphore can hold. \
+                 Use a value at or below `max_concurrent_incoming_offers`, which is the outer bound anything larger is inert against.",
+                nostr.max_concurrent_offers_per_npub,
+                tokio::sync::Semaphore::MAX_PERMITS
+            )));
+        }
+
         Ok(())
     }
 
@@ -2172,6 +2215,74 @@ peers:
         assert!(err.to_string().contains("after_secs"));
     }
 
+    #[test]
+    fn test_validate_signal_ttl_at_or_above_the_replay_window_margin_rejected() {
+        // 180 is the boundary: 180 + 2 * 60 = 300, which is not strictly less
+        // than the default 300s replay window.
+        for signal_ttl_secs in [180, 181, 3600, u64::MAX] {
+            let mut config = Config::default();
+            config.node.discovery.nostr.signal_ttl_secs = signal_ttl_secs;
+
+            match config.validate() {
+                Err(e) => {
+                    let msg = e.to_string();
+                    assert!(msg.contains("signal_ttl_secs"), "got: {msg}");
+                    assert!(msg.contains("replay_window_secs"), "got: {msg}");
+                }
+                Ok(()) => panic!("signal_ttl_secs = {signal_ttl_secs} should be rejected"),
+            }
+        }
+    }
+
+    #[test]
+    fn test_validate_signal_ttl_just_inside_the_replay_window_margin_accepted() {
+        let mut config = Config::default();
+        config.node.discovery.nostr.signal_ttl_secs = 179;
+
+        config
+            .validate()
+            .expect("179 + 2 * 60 = 299 leaves the freshness window inside the 300s replay window");
+    }
+
+    #[test]
+    fn test_validate_per_npub_offer_allowance_of_zero_rejected() {
+        let mut config = Config::default();
+        config.node.discovery.nostr.max_concurrent_offers_per_npub = 0;
+
+        let err = config.validate().expect_err("validation should fail");
+        assert!(
+            err.to_string().contains("max_concurrent_offers_per_npub"),
+            "got: {err}"
+        );
+    }
+
+    #[test]
+    fn test_validate_per_npub_offer_allowance_of_one_accepted() {
+        let mut config = Config::default();
+        config.node.discovery.nostr.max_concurrent_offers_per_npub = 1;
+
+        config
+            .validate()
+            .expect("an allowance of one offer per sender is restrictive but well defined");
+    }
+
+    #[test]
+    fn test_validate_shipped_defaults_satisfy_the_freshness_invariant() {
+        Config::default()
+            .validate()
+            .expect("the shipped defaults must satisfy every validation rule");
+
+        // Stated against the constant rather than a literal, so this reds if
+        // anyone changes FRESHNESS_SKEW_TOLERANCE_MS or either default without
+        // re-checking the relation they jointly have to satisfy.
+        let defaults = Config::default();
+        let nostr = &defaults.node.discovery.nostr;
+        assert!(
+            nostr.signal_ttl_secs + 2 * (FRESHNESS_SKEW_TOLERANCE_MS / 1000)
+                < nostr.replay_window_secs
+        );
+    }
+
     #[test]
     fn test_outbound_only_forces_ephemeral_bind() {
         let cfg = UdpConfig {

src/config/node.rs

@@ -349,6 +349,11 @@ pub struct NostrDiscoveryConfig {
     /// Acts as a rate limit against offer spam from relays.
     #[serde(default = "NostrDiscoveryConfig::default_max_concurrent_incoming_offers")]
     pub max_concurrent_incoming_offers: usize,
+    /// Max concurrent inbound traversal offers accepted from any one sender
+    /// npub. Sits inside `max_concurrent_incoming_offers`, which remains the
+    /// outer bound.
+    #[serde(default = "NostrDiscoveryConfig::default_max_concurrent_offers_per_npub")]
+    pub max_concurrent_offers_per_npub: usize,
     /// Max cached overlay adverts retained from relay traffic.
     /// Bounds memory under ambient advert volume.
     #[serde(default = "NostrDiscoveryConfig::default_advert_cache_max_entries")]
@@ -437,6 +442,7 @@ impl Default for NostrDiscoveryConfig {
             policy: NostrDiscoveryPolicy::default(),
             open_discovery_max_pending: Self::default_open_discovery_max_pending(),
             max_concurrent_incoming_offers: Self::default_max_concurrent_incoming_offers(),
+            max_concurrent_offers_per_npub: Self::default_max_concurrent_offers_per_npub(),
             advert_cache_max_entries: Self::default_advert_cache_max_entries(),
             seen_sessions_max_entries: Self::default_seen_sessions_max_entries(),
             attempt_timeout_secs: Self::default_attempt_timeout_secs(),
@@ -502,6 +508,20 @@ impl NostrDiscoveryConfig {
         16
     }
 
+    /// Four, derived rather than picked. The initiator side already admits at
+    /// most one in-flight traversal per peer npub, so one concurrent offer per
+    /// peer is the honest steady state. An offer is published to and consumed
+    /// from the whole DM relay set, three URLs by default, and whether the
+    /// notification stream deduplicates one event delivered by three relays is
+    /// not established here — if it does not, one honest offer can present as
+    /// three near-simultaneous admissions before the replay check rejects the
+    /// duplicates. Four is that worst-case fan-out plus one, so a retry
+    /// overlapping a still-timing-out attempt is still admitted, and it is a
+    /// quarter of the default global bound.
+    fn default_max_concurrent_offers_per_npub() -> usize {
+        4
+    }
+
     fn default_advert_cache_max_entries() -> usize {
         2048
     }

src/discovery/nostr/mod.rs

@@ -1,4 +1,5 @@
 mod failure_state;
+mod offer_admission;
 mod runtime;
 mod signal;
 mod stun;
@@ -8,6 +9,8 @@ mod types;
 #[cfg(test)]
 mod tests;
 
+pub(crate) use signal::FRESHNESS_SKEW_TOLERANCE_MS;
+
 pub use runtime::NostrDiscovery;
 pub use types::{
     ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent,

src/discovery/nostr/offer_admission.rs

@@ -0,0 +1,202 @@
+//! Per-npub admission control for inbound traversal offers.
+//!
+//! The intake path used to hold a single global semaphore sized by
+//! `max_concurrent_incoming_offers`, acquired before any identity check, so
+//! one sender could hold every slot and deny traversal onboarding to every
+//! other peer. This type takes a per-npub permit and a global permit
+//! together, so a single npub can occupy at most its own allowance.
+//!
+//! **What this does not fix.** The global pool stays exhaustible. Nostr
+//! identities are free to generate and the signal subscription carries no
+//! author restriction, so an attacker willing to run
+//! `ceil(global_limit / per_sender_limit)` throwaway npubs still saturates
+//! the pool at an unchanged total offer rate, and an honest peer still gets
+//! the same drop. What the per-npub allowance buys is that one identity can
+//! no longer do it alone, and that the refusal a spamming sender receives is
+//! distinguishable in the log from genuine saturation. A defence that raises
+//! the attacker's cost in more than keypairs would have to price the offer
+//! itself, which is a larger change than this one.
+//!
+//! **Lock discipline, which the correctness of the map bound rests on.**
+//! `try_admit` does the prune, the map lookup and both permit acquisitions
+//! under one `std::sync::Mutex` and never awaits inside it. The prune uses
+//! `Arc::strong_count` as an exact idle test: an `OwnedSemaphorePermit` owns
+//! an `Arc<Semaphore>`, so a count of 1 means the map holds the only
+//! reference and no permit for that npub is outstanding. Moving
+//! `try_acquire_owned` outside the lock breaks this — a concurrent prune
+//! could evict an entry whose permit was still live, and the next offer from
+//! that npub would build a fresh semaphore and over-admit. No test in this
+//! module catches that rearrangement, so it has to be preserved by reading.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+
+/// Why an inbound offer was refused a slot.
+///
+/// The two classes stay apart because they are different operator stories:
+/// one says the node is saturated, the other says a single sender is over its
+/// allowance and everyone else is unaffected.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum AdmissionReject {
+    /// The sender already holds its whole per-npub allowance.
+    SenderFull,
+    /// The node is at `max_concurrent_incoming_offers` across all senders.
+    GlobalFull,
+}
+
+/// A granted admission. Both permits release on drop, so the caller only has
+/// to keep this alive for as long as the offer is being processed.
+pub(super) struct OfferPermit {
+    _sender: OwnedSemaphorePermit,
+    _global: OwnedSemaphorePermit,
+}
+
+/// Admits inbound offers against a per-sender allowance nested inside a
+/// global bound.
+pub(super) struct OfferAdmission {
+    global: Arc<Semaphore>,
+    per_sender: Mutex<HashMap<String, Arc<Semaphore>>>,
+    per_sender_limit: usize,
+}
+
+impl OfferAdmission {
+    /// Build an admission gate bounded globally by `global_limit` and per
+    /// sender npub by `per_sender_limit`.
+    pub(super) fn new(global_limit: usize, per_sender_limit: usize) -> Self {
+        Self {
+            global: Arc::new(Semaphore::new(global_limit)),
+            per_sender: Mutex::new(HashMap::new()),
+            per_sender_limit,
+        }
+    }
+
+    /// Try to take one slot for `npub`, or say which bound refused it.
+    ///
+    /// The per-sender check runs first so a spamming sender is turned away
+    /// without ever touching the global pool, and therefore causes no churn
+    /// there.
+    pub(super) fn try_admit(&self, npub: &str) -> Result<OfferPermit, AdmissionReject> {
+        let mut map = self
+            .per_sender
+            .lock()
+            .expect("offer-admission mutex poisoned");
+
+        // Drop entries with no outstanding permit. Every live per-sender
+        // permit implies a live global permit, so at most `global_limit`
+        // senders survive a prune and the map is bounded by
+        // `global_limit + 1` after the insert below. A benign race with a
+        // permit dropping concurrently can only retain an idle entry for one
+        // more round, or drop an entry all of whose permits are already free;
+        // neither over-admits.
+        map.retain(|_, sem| Arc::strong_count(sem) > 1);
+
+        let sem = map
+            .entry(npub.to_string())
+            .or_insert_with(|| Arc::new(Semaphore::new(self.per_sender_limit)))
+            .clone();
+        let sender = sem
+            .try_acquire_owned()
+            .map_err(|_| AdmissionReject::SenderFull)?;
+        let global = self
+            .global
+            .clone()
+            .try_acquire_owned()
+            .map_err(|_| AdmissionReject::GlobalFull)?;
+
+        Ok(OfferPermit {
+            _sender: sender,
+            _global: global,
+        })
+    }
+
+    /// How many sender npubs the map currently holds.
+    #[cfg(test)]
+    pub(super) fn tracked_senders(&self) -> usize {
+        self.per_sender
+            .lock()
+            .expect("offer-admission mutex poisoned")
+            .len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn one_sender_cannot_take_more_than_its_allowance_or_deny_another_sender() {
+        let admission = OfferAdmission::new(16, 4);
+
+        // The attacker pushes for the whole global pool, which is what makes
+        // the victim's assertion below discriminating: without per-npub
+        // accounting it takes all sixteen and the victim is refused.
+        let mut held = Vec::new();
+        let mut refusals = Vec::new();
+        for _ in 0..16 {
+            match admission.try_admit("npub1attacker") {
+                Ok(permit) => held.push(permit),
+                Err(reject) => refusals.push(reject),
+            }
+        }
+
+        assert_eq!(
+            held.len(),
+            4,
+            "one npub must not hold more than its own allowance"
+        );
+        assert!(
+            refusals
+                .iter()
+                .all(|reject| *reject == AdmissionReject::SenderFull),
+            "a sender over its allowance is refused on its own account, not on the pool's: {refusals:?}"
+        );
+        assert!(
+            admission.try_admit("npub1victim").is_ok(),
+            "a second sender must still be admitted while the pool has slots"
+        );
+    }
+
+    #[test]
+    fn several_peers_bootstrapping_at_once_are_all_admitted() {
+        let admission = OfferAdmission::new(16, 4);
+
+        let mut held = Vec::new();
+        for peer in 0..8 {
+            held.push(
+                admission
+                    .try_admit(&format!("npub1peer{peer}"))
+                    .unwrap_or_else(|e| panic!("peer {peer} should be admitted, got {e:?}")),
+            );
+        }
+    }
+
+    #[test]
+    fn an_idle_sender_is_reclaimed_so_the_map_cannot_grow_without_bound() {
+        let admission = OfferAdmission::new(16, 4);
+
+        // Three senders stay busy for the whole test, so the map bound and
+        // not a collapse to a single entry is what the assertion measures.
+        let mut held = Vec::new();
+        for peer in 0..3 {
+            held.push(
+                admission
+                    .try_admit(&format!("npub1busy{peer}"))
+                    .expect("a busy peer is inside both bounds"),
+            );
+        }
+
+        for sender in 0..100 {
+            drop(
+                admission
+                    .try_admit(&format!("npub1transient{sender}"))
+                    .expect("a transient sender is inside both bounds"),
+            );
+        }
+
+        // Three still-held entries, plus the last transient sender, which was
+        // inserted after the prune that would have removed it.
+        assert_eq!(admission.tracked_senders(), 4);
+    }
+}

src/discovery/nostr/runtime.rs

@@ -12,11 +12,12 @@ use nostr::prelude::{
 };
 use nostr_sdk::{Client, ClientOptions, prelude::RelayPoolNotification};
 use serde::Serialize;
-use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, mpsc, oneshot};
+use tokio::sync::{Mutex, Notify, RwLock, broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 use tracing::{debug, info, trace, warn};
 
 use super::failure_state::FailureState;
+use super::offer_admission::{AdmissionReject, OfferAdmission};
 use super::signal::{
     FreshnessOutcome, SignalEnvelope, build_signal_event, create_traversal_answer,
     create_traversal_offer, estimate_clock_skew, unwrap_signal_event, validate_offer_freshness,
@@ -241,7 +242,7 @@ pub struct NostrDiscovery {
     pending_answers: Mutex<HashMap<String, oneshot::Sender<SignalEnvelope<TraversalAnswer>>>>,
     active_initiators: Mutex<HashSet<String>>,
     seen_sessions: Mutex<HashMap<String, u64>>,
-    offer_slots: Arc<Semaphore>,
+    admission: OfferAdmission,
     event_tx: mpsc::UnboundedSender<BootstrapEvent>,
     event_rx: Mutex<mpsc::UnboundedReceiver<BootstrapEvent>>,
     connect_task: Mutex<Option<JoinHandle<()>>>,
@@ -292,7 +293,10 @@ impl NostrDiscovery {
         let pubkey = keys.public_key();
         let npub = crate::encode_npub(&identity.pubkey());
         let (event_tx, event_rx) = mpsc::unbounded_channel();
-        let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));
+        let admission = OfferAdmission::new(
+            config.max_concurrent_incoming_offers,
+            config.max_concurrent_offers_per_npub,
+        );
 
         let failure_state = FailureState::new(
             config.failure_streak_threshold,
@@ -313,7 +317,7 @@ impl NostrDiscovery {
             pending_answers: Mutex::new(HashMap::new()),
             active_initiators: Mutex::new(HashSet::new()),
             seen_sessions: Mutex::new(HashMap::new()),
-            offer_slots,
+            admission,
             event_tx,
             event_rx: Mutex::new(event_rx),
             connect_task: Mutex::new(None),
@@ -826,13 +830,30 @@ impl NostrDiscovery {
                         && offer.message_type == "offer"
                         && offer.recipient_npub == self.npub
                     {
-                        let Ok(permit) = self.offer_slots.clone().try_acquire_owned() else {
-                            warn!(
-                                sender_npub = %sender_npub,
-                                limit = self.config.max_concurrent_incoming_offers,
-                                "rate-limited inbound traversal offer (max_concurrent_incoming_offers reached); offer dropped"
-                            );
-                            continue;
+                        let permit = match self.admission.try_admit(&sender_npub) {
+                            Ok(permit) => permit,
+                            Err(AdmissionReject::GlobalFull) => {
+                                warn!(
+                                    sender_npub = %sender_npub,
+                                    limit = self.config.max_concurrent_incoming_offers,
+                                    "rate-limited inbound traversal offer (max_concurrent_incoming_offers reached); offer dropped"
+                                );
+                                continue;
+                            }
+                            // Debug, not warn: the party that trips this is by
+                            // definition sending faster than the node wants, so
+                            // a record per rejection turns the spam into log
+                            // volume. The global-full arm above stays at warn
+                            // and remains the operator's signal that the node
+                            // is actually saturated.
+                            Err(AdmissionReject::SenderFull) => {
+                                debug!(
+                                    sender_npub = %sender_npub,
+                                    limit = self.config.max_concurrent_offers_per_npub,
+                                    "inbound traversal offer refused: sender is at its per-npub offer allowance"
+                                );
+                                continue;
+                            }
                         };
                         let runtime = Arc::clone(&self);
                         let peer_short = short_npub(&sender_npub);
@@ -1894,7 +1915,10 @@ impl NostrDiscovery {
             .opts(ClientOptions::new().autoconnect(false))
             .build();
         let config = NostrDiscoveryConfig::default();
-        let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));
+        let admission = OfferAdmission::new(
+            config.max_concurrent_incoming_offers,
+            config.max_concurrent_offers_per_npub,
+        );
         let (event_tx, event_rx) = mpsc::unbounded_channel();
         let failure_state = FailureState::new(
             config.failure_streak_threshold,
@@ -1914,7 +1938,7 @@ impl NostrDiscovery {
             pending_answers: Mutex::new(HashMap::new()),
             active_initiators: Mutex::new(HashSet::new()),
             seen_sessions: Mutex::new(HashMap::new()),
-            offer_slots,
+            admission,
             event_tx,
             event_rx: Mutex::new(event_rx),
             connect_task: Mutex::new(None),

src/discovery/nostr/signal.rs

@@ -11,7 +11,11 @@ use super::types::{BootstrapError, PunchHint, SIGNAL_KIND, TraversalAnswer, Trav
 /// past ~minutes erodes the freshness guarantee that backstops session-id
 /// replay protection. Tightening it below the size of a typical un-NTP'd
 /// drift defeats the purpose. 60s sits comfortably between those.
-pub(super) const FRESHNESS_SKEW_TOLERANCE_MS: u64 = 60_000;
+///
+/// `pub(crate)` because `Config::validate` derives the freshness window from
+/// it rather than restating the number, so changing it here moves the
+/// validation boundary with it.
+pub(crate) const FRESHNESS_SKEW_TOLERANCE_MS: u64 = 60_000;
 
 pub(super) struct SignalEnvelope<T> {
     pub(super) payload: T,

src/discovery/nostr/tests.rs

@@ -522,15 +522,76 @@ fn planned_remote_endpoints_cap_targets_from_an_oversized_candidate_list() {
     )
     .expect("endpoint planning should succeed");
 
+    // The exact figures are determined by the inputs: the plan is one
+    // reflexive-to-reflexive pair plus 300 reflexive-to-candidate pairs, all
+    // distinct, so the cap lands on exactly MAX_PUNCH_TARGETS. An inequality
+    // here could not tell the cap from a collapse to a single target.
     assert!(tally.capped > 0, "the cap should have discarded targets");
     assert!(tally.suspicious());
-    assert!(
-        endpoints.len() <= 8,
-        "expected at most 8 endpoints, got {}",
-        endpoints.len()
+    assert_eq!(tally.admitted, 8);
+    assert_eq!(endpoints.len(), 8);
+}
+
+/// The IPv6 arm of `is_never_punchable_ip` is exercised as a pure predicate
+/// elsewhere; this drives it end to end through the planner, which is the
+/// path the reflector attack actually uses.
+///
+/// The local address is a ULA so `lan_refs` is non-empty and `same_subnet_24`
+/// is genuinely called with two IPv6 strings. It splits on `.` and requires
+/// four parts, so no IPv6 candidate can ever satisfy the /24 gate and
+/// `fd00::1` is refused off-subnet rather than admitted.
+#[test]
+fn planned_remote_endpoints_reject_never_punchable_ipv6_candidates() {
+    let (endpoints, _tally) = planned_remote_endpoints(
+        &[addr("fd00::2", 62000)],
+        Some(&addr("203.0.113.10", 62000)),
+        &[
+            addr("::1", 63000),
+            addr("::", 63000),
+            addr("ff02::1", 63000),
+            addr("fe80::1", 63000),
+            addr("fd00::1", 63000),
+        ],
+        Some(&addr("198.51.100.20", 63000)),
+    )
+    .expect("endpoint planning should succeed");
+
+    assert_eq!(
+        endpoints,
+        vec!["198.51.100.20:63000".parse::<SocketAddr>().unwrap()]
     );
 }
 
+/// The cap test above uses public candidates so the cap and not the filter is
+/// what bounds the output. The attack shape is the opposite: several hundred
+/// attacker-chosen unroutable addresses. This is the only test that pins the
+/// filter and the cap acting together on that input, and its failure would
+/// mean the reflector is back.
+#[test]
+fn planned_remote_endpoints_bound_an_oversized_list_of_unroutable_candidates() {
+    let mut remotes = Vec::new();
+    for host in 1..=150u8 {
+        remotes.push(addr(&format!("127.0.0.{host}"), 63000));
+        remotes.push(addr(&format!("224.0.0.{host}"), 63000));
+    }
+
+    let (endpoints, tally) = planned_remote_endpoints(
+        &[],
+        Some(&addr("203.0.113.10", 62000)),
+        &remotes,
+        Some(&addr("198.51.100.20", 63000)),
+    )
+    .expect("endpoint planning should succeed");
+
+    assert_eq!(
+        endpoints,
+        vec!["198.51.100.20:63000".parse::<SocketAddr>().unwrap()]
+    );
+    assert_eq!(tally.unroutable, 300);
+    assert_eq!(tally.admitted, 1);
+    assert!(tally.suspicious());
+}
+
 /// Guards the deployment whose STUN server sits inside the private network,
 /// so the observed reflexive address is itself private. Applying the /24 gate
 /// to a peer's reflexive address would drop it and remove the only branch

src/discovery/nostr/traversal.rs

@@ -471,14 +471,26 @@ pub(super) fn nonce() -> String {
 /// step, which is what a resume produces, saturates the punch start delay to
 /// zero so punching begins immediately; the attempt's own bounds are monotonic
 /// `Instant` deadlines, so its length is unaffected. A backward step lengthens
-/// that delay instead and can cost a single punch attempt, which retries. Early
-/// eviction from the replay window cannot admit a replay under the shipped
-/// defaults, because the freshness window a replayed offer would also have to
-/// satisfy (`signal_ttl_secs` plus `FRESHNESS_SKEW_TOLERANCE_MS` on each side,
-/// 240s under the shipped defaults) is strictly narrower than the replay window
-/// itself (`replay_window_secs`, 300s). The margin holds while
-/// `signal_ttl_secs + 120 < replay_window_secs`; nothing in config validation
-/// enforces that relation today.
+/// that delay instead and can cost a single punch attempt, which retries.
+/// Expiry-driven eviction from the replay window cannot admit a replay, because
+/// the freshness window a replayed offer would also have to satisfy
+/// (`signal_ttl_secs` plus `FRESHNESS_SKEW_TOLERANCE_MS` on each side, 240s
+/// under the shipped defaults) is strictly narrower than the replay window
+/// itself (`replay_window_secs`, 300s). `Config::validate` enforces
+/// `signal_ttl_secs + 2 * FRESHNESS_SKEW_TOLERANCE_MS < replay_window_secs`, so
+/// that margin can no longer be configured away.
+///
+/// That covers the expiry route only. `mark_session_seen` also evicts on
+/// capacity, dropping the entries nearest expiry once the cache exceeds
+/// `seen_sessions_max_entries`, and a session id dropped that way stays
+/// replayable for the rest of its freshness window whatever the time relation
+/// is. Nothing bounds that route: whether it is reachable depends on the
+/// admitted-offer rate against the cache size. The shipped defaults leave a
+/// wide margin — filling 2048 entries inside 300s needs about 6.8 admitted
+/// offers per second, against roughly 1.2/s from a 16-slot pool whose permits
+/// are held for the length of an attempt — but raising
+/// `max_concurrent_incoming_offers` narrows it. That admission rate is inferred
+/// from the attempt timeout, not measured.
 pub(super) fn now_ms() -> u64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)

testing/nat/scripts/generate-configs.sh

@@ -102,7 +102,10 @@ node:
         - "$stun_addr"
       signal_ttl_secs: 30
       attempt_timeout_secs: 6
-      replay_window_secs: 60
+      # Must stay above signal_ttl_secs plus 60s of clock-skew grace on each
+      # side, or config validation refuses to start the node. 180 keeps a 30s
+      # margin over the 150s freshness window the 30s TTL implies.
+      replay_window_secs: 180
       punch_start_delay_ms: 500
       punch_interval_ms: 100
       punch_duration_ms: 2500