From bcc9c525d35cf6614dbd608f84b82180faed0c43 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 4 May 2026 12:39:51 +0000 Subject: [PATCH 1/2] nostr: per-peer NAT-traversal failure suppression and clock-skew handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Public-test daemons with populous open-discovery caches generate sustained NAT-traversal-failure WARN volume (~140/hour, ~3500/day) against cache-learned peers that have gone offline — their adverts are absent from major Nostr relays but cached entries persist until their advertised `valid_until` expires. The daemon kept publishing offers indefinitely under exponential backoff with no per-peer suppression, drowning operator signal and hammering relays. A parallel concern: the strict freshness check at signal.rs silently rejected offers under modest clock skew (now_ms() anchors to SystemTime once at startup, so post-startup NTP step adjustments don't propagate on long-uptime daemons), indistinguishable from "peer is offline." Six independent improvements layered on the existing retry logic. Per-npub WARN log rate-limit ---------------------------- New `FailureState` struct on `NostrDiscovery` records per-npub `last_warn_at_ms`. Subsequent failures inside `warn_log_interval_secs` (default 5 min) emit DEBUG instead of WARN. Each WARN now also carries `consecutive_failures` and remaining `cooldown_secs` so operators can read the trajectory without grepping multiple lines. Per-npub consecutive-failure counter + extended cooldown -------------------------------------------------------- After `failure_streak_threshold` (default 5) consecutive failures against a peer, the next `extended_cooldown_secs` (default 1800) of attempts are suppressed by pushing `retry_pending[npub].retry_after_ms` past the cooldown wall. The open-discovery sweep also consults `cooldown_until` and increments a new `skipped_cooldown` counter so a peer whose `retry_pending` was cleared by max_retries doesn't get re-enqueued during the cooldown window. Caps offer-publish rate per dead peer regardless of how often the sweep tries to re-enqueue. Stale-advert eviction on streak-threshold transition ---------------------------------------------------- On the threshold-crossing transition (one-shot, not every subsequent failure), `tokio::spawn` an active re-fetch of the peer's Kind 37195 advert from `advert_relays`. Three outcomes: - absent on relays → cache evicted; sweep won't re-enqueue (peer is genuinely gone). - newer `created_at` → cache refreshed + streak reset (peer republished; allowed to retry immediately). - same → cache untouched; cooldown stands. Cost: ~one fetch per dead peer per 30-min cooldown cycle, vs hundreds of offer publishes/hour today. Clock-skew tolerance on freshness check --------------------------------------- `signal.rs` `validate_offer_freshness` and `validate_traversal_answer_for_offer` now allow ±60s grace beyond strict TTL. Both return a new `FreshnessOutcome` enum so callers can DEBUG-log when an offer/answer was only accepted via the grace window. `FRESHNESS_SKEW_TOLERANCE_MS` is hard-coded — loosening this past minutes erodes the freshness/replay security boundary and operators tend to tune in the wrong direction. NTP-style skew estimate (offer_received_at echo) ------------------------------------------------ Added optional `offerReceivedAt: Option` field to `TraversalAnswer` payload. Responder fills it with `now_ms()` at offer-receipt time. Initiator computes the standard NTP offset formula `((T2-T1) + (T3-T4)) / 2` against the round-trip and DEBUG-logs when `|skew| ≥ 30s`. Skew is also stashed in `FailureState` and surfaced in `show_peers`. Non-breaking — older responders that don't fill the field still produce valid answers, and `estimate_clock_skew` returns `None`. Per-peer state in `show_peers` JSON ----------------------------------- Each peer entry in `show_peers` now carries: "nostr_traversal": { "consecutive_failures": , "in_cooldown": , "cooldown_until_ms": , "last_observed_skew_ms": } Always emitted (schema-stable); values populated when discovery is enabled and the npub has a recorded entry. Required a new public `Node::nostr_discovery_handle()` accessor and refactored `FailureState`'s internal Mutex from `tokio::sync` to `std::sync` (operations never hold across await), which lets the synchronous `show_peers` handler call `snapshot()` directly without the dispatcher becoming async. New config knobs (under `node.discovery.nostr`) ----------------------------------------------- failure_streak_threshold: 5 extended_cooldown_secs: 1800 warn_log_interval_secs: 300 failure_state_max_entries: 4096 Tests ----- 12 new unit tests: - 5 in `tests.rs` covering freshness strict / tolerated / rejected outcomes, NTP skew estimation, and the backward-compat None case when the responder didn't fill `offer_received_at`. - 7 in `failure_state.rs` covering streak/warn-rate-limit state transitions, cooldown active vs expired semantics, success-resets-streak, observed-skew records, and size-cap eviction by oldest `last_failure_at`. CHANGELOG entries added under `[Unreleased]` Fixed. --- CHANGELOG.md | 25 +++ src/config/node.rs | 43 ++++ src/control/queries.rs | 37 ++++ src/discovery/nostr/failure_state.rs | 311 +++++++++++++++++++++++++++ src/discovery/nostr/mod.rs | 8 +- src/discovery/nostr/runtime.rs | 198 ++++++++++++++++- src/discovery/nostr/signal.rs | 101 +++++++-- src/discovery/nostr/tests.rs | 172 ++++++++++++++- src/discovery/nostr/types.rs | 42 ++++ src/node/lifecycle.rs | 71 +++++- src/node/mod.rs | 7 + 11 files changed, 987 insertions(+), 28 deletions(-) create mode 100644 src/discovery/nostr/failure_state.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e5e04df..b27e449 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -274,6 +274,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Nostr-discovery now tolerates ±60s of clock skew on offer/answer + freshness checks so a responder whose wall clock leads the + initiator's by less than that no longer silently rejects every + offer. Previously, a public-test daemon with un-NTP'd peers (or + long uptime — `now_ms()` anchors to `SystemTime` once at startup, + then advances monotonically; post-startup NTP step adjustments + don't propagate) would see ~100% signal-timeout rate against + skewed peers, indistinguishable from "peer is offline." New + optional `offerReceivedAt` field on the answer payload lets the + initiator log per-peer NTP-style skew estimates (DEBUG when ≥30s) + for operator visibility. Backward-compatible — older responders + that don't fill the field still produce valid answers +- Nostr-discovery NAT-traversal failure suppression: per-npub + consecutive-failure counter triggers a 30-min extended cooldown + after 5 failures, preventing the daemon from hammering Nostr + relays with offers to peers that have gone away. WARN log lines + rate-limited to one per peer per 5 min (subsequent failures + emit DEBUG with `consecutive_failures` + remaining `cooldown_secs`). + Threshold-crossing also fires a one-shot active re-check of the + peer's Kind 37195 advert against `advert_relays`; absent → + evict cache; newer → refresh + reset streak; same → cooldown + stands. New `failure_streak_threshold`, `extended_cooldown_secs`, + `warn_log_interval_secs`, `failure_state_max_entries` config + fields under `node.discovery.nostr`. Per-peer state visible in + `fipsctl show peers` JSON under `nostr_traversal` - Tor onion adverts published over Nostr overlay discovery now include the public-facing port (`.onion:`) instead of just the bare onion hostname. The publisher previously emitted a diff --git a/src/config/node.rs b/src/config/node.rs index 6bfcfd5..9b95748 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -365,6 +365,29 @@ pub struct NostrDiscoveryConfig { /// `policy: open`. Default: 3600 (1 hour). #[serde(default = "NostrDiscoveryConfig::default_startup_sweep_max_age_secs")] pub startup_sweep_max_age_secs: u64, + /// Number of consecutive NAT-traversal failures against a peer before + /// an extended cooldown is applied to throttle further offer publishes. + /// At this threshold the daemon also actively re-fetches the peer's + /// advert from `advert_relays` to evict cache entries for peers that + /// have gone away. Default: 5. + #[serde(default = "NostrDiscoveryConfig::default_failure_streak_threshold")] + pub failure_streak_threshold: u32, + /// Cooldown applied to a peer once `failure_streak_threshold` is hit. + /// Suppresses both open-discovery sweep enqueues and per-attempt + /// retry firings until elapsed. Default: 1800 (30 minutes). + #[serde(default = "NostrDiscoveryConfig::default_extended_cooldown_secs")] + pub extended_cooldown_secs: u64, + /// Minimum interval between `NAT traversal failed` WARN log lines for + /// the same peer. Subsequent failures inside the window log at DEBUG. + /// Reduces log spam on public-test nodes with many cache-learned + /// peers. Default: 300 (5 minutes). + #[serde(default = "NostrDiscoveryConfig::default_warn_log_interval_secs")] + pub warn_log_interval_secs: u64, + /// Maximum entries retained in the per-npub failure-state map. + /// Bounds memory under high cache turnover. Oldest entries (by last + /// failure time) evicted when the cap is exceeded. Default: 4096. + #[serde(default = "NostrDiscoveryConfig::default_failure_state_max_entries")] + pub failure_state_max_entries: usize, } impl Default for NostrDiscoveryConfig { @@ -392,6 +415,10 @@ impl Default for NostrDiscoveryConfig { advert_refresh_secs: Self::default_advert_refresh_secs(), startup_sweep_delay_secs: Self::default_startup_sweep_delay_secs(), startup_sweep_max_age_secs: Self::default_startup_sweep_max_age_secs(), + failure_streak_threshold: Self::default_failure_streak_threshold(), + extended_cooldown_secs: Self::default_extended_cooldown_secs(), + warn_log_interval_secs: Self::default_warn_log_interval_secs(), + failure_state_max_entries: Self::default_failure_state_max_entries(), } } } @@ -484,6 +511,22 @@ impl NostrDiscoveryConfig { fn default_startup_sweep_max_age_secs() -> u64 { 3_600 } + + fn default_failure_streak_threshold() -> u32 { + 5 + } + + fn default_extended_cooldown_secs() -> u64 { + 1_800 + } + + fn default_warn_log_interval_secs() -> u64 { + 300 + } + + fn default_failure_state_max_entries() -> usize { + 4_096 + } } /// Spanning tree (`node.tree.*`). diff --git a/src/control/queries.rs b/src/control/queries.rs index 7547192..b7d3575 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -114,6 +114,18 @@ pub fn show_peers(node: &Node) -> Value { let parent_id = *tree.my_declaration().parent_id(); let is_root = tree.is_root(); + // Per-npub Nostr-traversal failure-state snapshot, indexed by npub + // for O(1) per-peer lookup. Empty if Nostr discovery is disabled. + let nostr_state: std::collections::HashMap = node + .nostr_discovery_handle() + .map(|d| { + d.failure_state_snapshot() + .into_iter() + .map(|view| (view.npub.clone(), view)) + .collect() + }) + .unwrap_or_default(); + let peers: Vec = node .peers() .map(|peer| { @@ -175,6 +187,31 @@ pub fn show_peers(node: &Node) -> Value { peer_json["replay_suppressed"] = json!(peer.replay_suppressed_count()); peer_json["consecutive_decrypt_failures"] = json!(peer.consecutive_decrypt_failures()); + // Nostr-traversal state if this peer's npub appears in + // failure-state. Always emitted (even null) so the schema + // stays stable; values populated only when Nostr discovery + // is enabled and the npub has been seen. + let npub = peer.npub(); + let mut nostr_obj = json!({ + "consecutive_failures": 0, + "in_cooldown": false, + "cooldown_until_ms": Value::Null, + "last_observed_skew_ms": Value::Null, + }); + if let Some(state) = nostr_state.get(&npub) { + nostr_obj["consecutive_failures"] = json!(state.consecutive_failures); + nostr_obj["in_cooldown"] = json!(state.cooldown_until_ms.is_some()); + nostr_obj["cooldown_until_ms"] = state + .cooldown_until_ms + .map(|t| json!(t)) + .unwrap_or(Value::Null); + nostr_obj["last_observed_skew_ms"] = state + .last_observed_skew_ms + .map(|s| json!(s)) + .unwrap_or(Value::Null); + } + peer_json["nostr_traversal"] = nostr_obj; + // Noise session counters (rekey urgency, replay window state) if let Some(session) = peer.noise_session() { peer_json["noise"] = json!({ diff --git a/src/discovery/nostr/failure_state.rs b/src/discovery/nostr/failure_state.rs new file mode 100644 index 0000000..57d2b0e --- /dev/null +++ b/src/discovery/nostr/failure_state.rs @@ -0,0 +1,311 @@ +//! Per-npub NAT-traversal failure tracking. +//! +//! Records consecutive offer/answer signal-timeout (and other) failures +//! against each peer. Drives three operator-visible behaviors: +//! +//! - **WARN log rate-limit (B1).** Suppresses repeat WARN lines for the +//! same peer inside a configurable window; subsequent failures inside +//! the window log at DEBUG instead. +//! - **Extended cooldown (B2).** Once a peer accumulates +//! `failure_streak_threshold` consecutive failures, the next +//! `extended_cooldown_secs` worth of attempts are suppressed by pushing +//! the retry timer out, capping how aggressively a public-test node can +//! hammer Nostr relays with offers to dead peers. +//! - **Stale-advert eviction (B6).** At streak threshold, the daemon +//! actively re-fetches the peer's advert; outcomes (`Evicted`, +//! `Refreshed`, `SameAdvert`, `Skipped`) feed back into the cache so +//! peers that have actually disappeared stop being retried after +//! eviction (`prune_advert_cache` semantics). +//! +//! Also stores last-observed clock skew (from B5a) so operators can +//! surface it via `fipsctl show peers` (B3). + +use std::collections::HashMap; +use std::sync::Mutex; + +/// One peer's failure-tracking state. Keyed by bech32 npub string in the +/// owning `FailureState` map. +#[derive(Debug, Clone)] +pub(super) struct NpubFailureRecord { + /// Number of consecutive failures since the last success (or fresh + /// advert that reset the streak). + pub consecutive_failures: u32, + /// When this entry was last touched, used for size-cap eviction. + pub last_failure_at_ms: u64, + /// When the last WARN was emitted for this peer; controls WARN + /// rate-limit (B1). + pub last_warn_at_ms: Option, + /// When the extended cooldown was applied; while + /// `cooldown_until_ms.is_some_and(|t| t > now)`, retries are + /// suppressed. + pub cooldown_until_ms: Option, + /// Most recent NTP-style skew estimate (B5a), in ms (positive = + /// peer ahead of us). `None` if the peer hasn't successfully + /// answered an offer with `offerReceivedAt` populated, or if + /// successful traversal cleared the streak (we keep the last-seen + /// skew, but only on records that are still in the map). + pub last_observed_skew_ms: Option, +} + +impl NpubFailureRecord { + fn new(now_ms: u64) -> Self { + Self { + consecutive_failures: 0, + last_failure_at_ms: now_ms, + last_warn_at_ms: None, + cooldown_until_ms: None, + last_observed_skew_ms: None, + } + } +} + +/// What the lifecycle layer should do based on the recorded failure. +#[derive(Debug, Clone, Copy)] +pub(super) struct FailureDecision { + /// Updated streak count (post-increment). + pub consecutive_failures: u32, + /// True iff lifecycle should log at WARN; false → log at DEBUG. + pub should_warn: bool, + /// If set, retry_after_ms for this peer should not fire before this + /// wall-clock ms. + pub cooldown_until_ms: Option, + /// True only on the streak-threshold-crossing transition. Lifecycle + /// should run a one-shot stale-advert check (B6) when this fires. + pub crossed_threshold: bool, +} + +pub(super) struct FailureState { + inner: Mutex>, + threshold: u32, + extended_cooldown_ms: u64, + warn_log_interval_ms: u64, + max_entries: usize, +} + +impl FailureState { + pub(super) fn new( + threshold: u32, + extended_cooldown_secs: u64, + warn_log_interval_secs: u64, + max_entries: usize, + ) -> Self { + Self { + inner: Mutex::new(HashMap::new()), + threshold, + extended_cooldown_ms: extended_cooldown_secs.saturating_mul(1000), + warn_log_interval_ms: warn_log_interval_secs.saturating_mul(1000), + max_entries, + } + } + + /// Record a traversal failure against `npub`. Returns the resulting + /// FailureDecision for the lifecycle layer to act on. + pub(super) fn record_failure(&self, npub: &str, now_ms: u64) -> FailureDecision { + let mut map = self.inner.lock().expect("failure-state mutex poisoned"); + let entry = map + .entry(npub.to_string()) + .or_insert_with(|| NpubFailureRecord::new(now_ms)); + entry.consecutive_failures = entry.consecutive_failures.saturating_add(1); + entry.last_failure_at_ms = now_ms; + + let crossed_threshold = entry.consecutive_failures == self.threshold; + let cooldown_until_ms = if entry.consecutive_failures >= self.threshold { + let cooldown = now_ms.saturating_add(self.extended_cooldown_ms); + entry.cooldown_until_ms = Some(cooldown); + Some(cooldown) + } else { + None + }; + + let should_warn = !matches!( + entry.last_warn_at_ms, + Some(last) if now_ms.saturating_sub(last) < self.warn_log_interval_ms + ); + if should_warn { + entry.last_warn_at_ms = Some(now_ms); + } + + let decision = FailureDecision { + consecutive_failures: entry.consecutive_failures, + should_warn, + cooldown_until_ms, + crossed_threshold, + }; + + if map.len() > self.max_entries { + evict_oldest(&mut map, self.max_entries); + } + + decision + } + + /// Record a successful traversal — clears the streak and cooldown. + /// Last-observed skew is retained until next eviction since it's + /// useful to display in `show_peers` even for healthy peers. + pub(super) fn record_success(&self, npub: &str, now_ms: u64) { + let mut map = self.inner.lock().expect("failure-state mutex poisoned"); + if let Some(entry) = map.get_mut(npub) { + entry.consecutive_failures = 0; + entry.cooldown_until_ms = None; + entry.last_failure_at_ms = now_ms; + } + // No insert if absent — successful peers don't need a record. + } + + /// Record an observed clock-skew estimate from a successful answer + /// receipt (B5a). Creates an entry if needed so we can surface the + /// skew via `show_peers` even when the peer is healthy. + pub(super) fn note_observed_skew(&self, npub: &str, skew_ms: i64, now_ms: u64) { + let mut map = self.inner.lock().expect("failure-state mutex poisoned"); + let entry = map + .entry(npub.to_string()) + .or_insert_with(|| NpubFailureRecord::new(now_ms)); + entry.last_observed_skew_ms = Some(skew_ms); + + if map.len() > self.max_entries { + evict_oldest(&mut map, self.max_entries); + } + } + + /// Reset streak/cooldown after a successful B6 advert refresh. + pub(super) fn reset_streak_after_refresh(&self, npub: &str) { + let mut map = self.inner.lock().expect("failure-state mutex poisoned"); + if let Some(entry) = map.get_mut(npub) { + entry.consecutive_failures = 0; + entry.cooldown_until_ms = None; + } + } + + /// Return cooldown_until_ms if the peer is currently in extended + /// cooldown. + pub(super) fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option { + let map = self.inner.lock().expect("failure-state mutex poisoned"); + map.get(npub) + .and_then(|e| e.cooldown_until_ms) + .filter(|&t| t > now_ms) + } + + /// Snapshot for `show_peers` rendering (B3). + pub(super) fn snapshot(&self) -> Vec<(String, NpubFailureRecord)> { + let map = self.inner.lock().expect("failure-state mutex poisoned"); + map.iter() + .map(|(npub, rec)| (npub.clone(), rec.clone())) + .collect() + } +} + +fn evict_oldest(map: &mut HashMap, target: usize) { + if map.len() <= target { + return; + } + let overflow = map.len() - target; + let mut entries: Vec<(String, u64)> = map + .iter() + .map(|(k, v)| (k.clone(), v.last_failure_at_ms)) + .collect(); + entries.sort_by_key(|(_, t)| *t); + for (k, _) in entries.into_iter().take(overflow) { + map.remove(&k); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fs() -> FailureState { + // threshold=3, cooldown=10s, warn-interval=5s, cap=8 + FailureState::new(3, 10, 5, 8) + } + + #[test] + fn first_failure_warns_and_no_cooldown() { + let s = fs(); + let d = s.record_failure("npub1a", 1000); + assert_eq!(d.consecutive_failures, 1); + assert!(d.should_warn); + assert!(d.cooldown_until_ms.is_none()); + assert!(!d.crossed_threshold); + } + + #[test] + fn warn_suppressed_inside_window_then_unsuppressed_after() { + let s = fs(); + let d1 = s.record_failure("npub1a", 1000); + let d2 = s.record_failure("npub1a", 1500); + assert!(d1.should_warn); + assert!( + !d2.should_warn, + "second failure inside 5s window must DEBUG" + ); + // 5s warn-interval = 5000 ms; bump beyond. + let d3 = s.record_failure("npub1a", 1000 + 5_500); + assert!(d3.should_warn, "after window, must WARN again"); + } + + #[test] + fn streak_threshold_triggers_cooldown_and_signals_crossing() { + let s = fs(); + let _ = s.record_failure("npub1a", 1000); + let _ = s.record_failure("npub1a", 1100); + let d3 = s.record_failure("npub1a", 1200); + assert_eq!(d3.consecutive_failures, 3); + assert!(d3.crossed_threshold); + assert_eq!(d3.cooldown_until_ms, Some(1200 + 10_000)); + // Subsequent failure does NOT re-fire crossed_threshold. + let d4 = s.record_failure("npub1a", 1300); + assert!(!d4.crossed_threshold); + assert!(d4.cooldown_until_ms.is_some()); + } + + #[test] + fn record_success_clears_streak() { + let s = fs(); + for t in [1000u64, 1100, 1200, 1300] { + let _ = s.record_failure("npub1a", t); + } + s.record_success("npub1a", 2000); + let d = s.record_failure("npub1a", 3000); + assert_eq!(d.consecutive_failures, 1, "streak reset after success"); + assert!(!d.crossed_threshold); + } + + #[test] + fn cooldown_until_returns_only_active_cooldowns() { + let s = fs(); + for t in [1000u64, 1100, 1200] { + let _ = s.record_failure("npub1a", t); + } + // Mid-cooldown + assert!(s.cooldown_until("npub1a", 5_000).is_some()); + // Past cooldown + assert!(s.cooldown_until("npub1a", 1200 + 10_001).is_none()); + } + + #[test] + fn note_observed_skew_creates_entry_for_healthy_peer() { + let s = fs(); + s.note_observed_skew("npub1healthy", 250, 1000); + let snap = s.snapshot(); + assert_eq!(snap.len(), 1); + let (npub, rec) = &snap[0]; + assert_eq!(npub, "npub1healthy"); + assert_eq!(rec.last_observed_skew_ms, Some(250)); + assert_eq!(rec.consecutive_failures, 0); + } + + #[test] + fn size_cap_evicts_oldest_by_last_failure_at() { + let s = fs(); // cap = 8 + for i in 0..10 { + let npub = format!("npub1{i}"); + let _ = s.record_failure(&npub, 1000 + i as u64); + } + let snap = s.snapshot(); + assert!(snap.len() <= 8, "cap not enforced: {}", snap.len()); + // Oldest two (npub10, npub11) should be evicted; newer kept. + let names: std::collections::HashSet<_> = snap.iter().map(|(n, _)| n.clone()).collect(); + assert!(!names.contains("npub10")); + assert!(names.contains("npub19")); + } +} diff --git a/src/discovery/nostr/mod.rs b/src/discovery/nostr/mod.rs index c08f6f0..881fb2e 100644 --- a/src/discovery/nostr/mod.rs +++ b/src/discovery/nostr/mod.rs @@ -1,3 +1,4 @@ +mod failure_state; mod runtime; mod signal; mod stun; @@ -10,7 +11,8 @@ mod tests; pub use runtime::NostrDiscovery; pub use types::{ ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent, - CachedOverlayAdvert, OverlayAdvert, OverlayEndpointAdvert, OverlayTransportKind, - PROTOCOL_VERSION, PUNCH_ACK_MAGIC, PUNCH_MAGIC, PunchHint, PunchPacket, PunchPacketKind, - SIGNAL_KIND, TraversalAddress, TraversalAnswer, TraversalOffer, + CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome, + OverlayAdvert, OverlayEndpointAdvert, OverlayTransportKind, PROTOCOL_VERSION, PUNCH_ACK_MAGIC, + PUNCH_MAGIC, PunchHint, PunchPacket, PunchPacketKind, SIGNAL_KIND, TraversalAddress, + TraversalAnswer, TraversalOffer, }; diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index 6dd4844..a08bf88 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -14,16 +14,19 @@ use tokio::sync::{Mutex, RwLock, Semaphore, mpsc, oneshot}; use tokio::task::JoinHandle; use tracing::{debug, trace, warn}; +use super::failure_state::FailureState; use super::signal::{ - SignalEnvelope, build_signal_event, create_traversal_answer, create_traversal_offer, - unwrap_signal_event, validate_offer_freshness, validate_traversal_answer_for_offer, + FreshnessOutcome, SignalEnvelope, build_signal_event, create_traversal_answer, + create_traversal_offer, estimate_clock_skew, unwrap_signal_event, validate_offer_freshness, + validate_traversal_answer_for_offer, }; use super::stun::observe_traversal_addresses; use super::traversal::{nonce, now_ms, planned_remote_endpoints, run_punch_attempt}; use super::types::{ ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent, - CachedOverlayAdvert, OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint, - SIGNAL_KIND, TraversalAnswer, TraversalOffer, + CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome, + OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint, SIGNAL_KIND, + TraversalAnswer, TraversalOffer, }; use crate::config::{NostrDiscoveryConfig, PeerConfig}; use crate::discovery::EstablishedTraversal; @@ -70,6 +73,7 @@ pub struct NostrDiscovery { event_rx: Mutex>, notify_task: Mutex>>, advertise_task: Mutex>>, + failure_state: FailureState, } impl NostrDiscovery { @@ -104,6 +108,13 @@ impl NostrDiscovery { let (event_tx, event_rx) = mpsc::unbounded_channel(); let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers)); + let failure_state = FailureState::new( + config.failure_streak_threshold, + config.extended_cooldown_secs, + config.warn_log_interval_secs, + config.failure_state_max_entries, + ); + let runtime = Arc::new(Self { client, keys, @@ -121,6 +132,7 @@ impl NostrDiscovery { event_rx: Mutex::new(event_rx), notify_task: Mutex::new(None), advertise_task: Mutex::new(None), + failure_state, }); runtime.subscribe().await?; @@ -154,6 +166,121 @@ impl NostrDiscovery { }); } + /// Record a NAT-traversal failure for `npub`, returning the + /// resulting decision (WARN suppression + extended cooldown + + /// threshold-crossing flag for the B6 re-fetch). + pub fn record_traversal_failure(&self, npub: &str, now_ms: u64) -> NostrFailureDecision { + let d = self.failure_state.record_failure(npub, now_ms); + NostrFailureDecision { + consecutive_failures: d.consecutive_failures, + should_warn: d.should_warn, + cooldown_until_ms: d.cooldown_until_ms, + crossed_threshold: d.crossed_threshold, + } + } + + /// Record a successful traversal — clears the streak/cooldown. + pub fn record_traversal_success(&self, npub: &str, now_ms: u64) { + self.failure_state.record_success(npub, now_ms); + } + + /// Cooldown wall-clock ms if the peer is currently suppressed, + /// else None. Used by the open-discovery sweep to skip enqueue. + pub fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option { + self.failure_state.cooldown_until(npub, now_ms) + } + + /// Snapshot of per-npub failure state for `show_peers` rendering. + pub fn failure_state_snapshot(&self) -> Vec { + self.failure_state + .snapshot() + .into_iter() + .map(|(npub, rec)| NostrPeerFailureView { + npub, + consecutive_failures: rec.consecutive_failures, + cooldown_until_ms: rec.cooldown_until_ms, + last_observed_skew_ms: rec.last_observed_skew_ms, + }) + .collect() + } + + /// Stale-advert re-check (B6). Called by lifecycle on the + /// streak-threshold transition. Actively re-queries the peer's + /// Kind 37195 advert from `advert_relays`; evicts the cache entry + /// if absent, refreshes if newer than the cached `created_at`, + /// otherwise leaves the cache untouched. + pub async fn refetch_advert_for_stale_check(&self, peer_npub: &str) -> NostrRefetchOutcome { + let target_pubkey = match PublicKey::parse(peer_npub) { + Ok(p) => p, + Err(_) => return NostrRefetchOutcome::Skipped, + }; + if self.config.advert_relays.is_empty() { + return NostrRefetchOutcome::Skipped; + } + let cached_created_at = self + .advert_cache + .read() + .await + .get(peer_npub) + .map(|c| c.created_at); + + let events = match self + .client + .fetch_events_from( + self.config.advert_relays.clone(), + Filter::new() + .author(target_pubkey) + .kind(Kind::Custom(ADVERT_KIND)) + .identifier(ADVERT_IDENTIFIER), + Duration::from_secs(2), + ) + .await + { + Ok(e) => e, + Err(_) => return NostrRefetchOutcome::Skipped, + }; + + let mut newest: Option<(u64, &Event)> = None; + for ev in events.iter() { + let ts = ev.created_at.as_secs(); + match newest { + Some((cur, _)) if ts <= cur => {} + _ => newest = Some((ts, ev)), + } + } + + let Some((relay_created_at, ev)) = newest else { + // Absent on relays. Evict any stale cache entry. + self.advert_cache.write().await.remove(peer_npub); + self.failure_state.reset_streak_after_refresh(peer_npub); + return NostrRefetchOutcome::Evicted; + }; + + match cached_created_at { + Some(cached) if relay_created_at <= cached => NostrRefetchOutcome::SameAdvert, + _ => { + let Some(valid_until_ms) = self.event_valid_until_ms(ev) else { + return NostrRefetchOutcome::Skipped; + }; + let Ok(advert) = Self::parse_overlay_advert_event(ev, &self.config.app) else { + return NostrRefetchOutcome::Skipped; + }; + let updated = CachedOverlayAdvert { + author_npub: peer_npub.to_string(), + advert, + created_at: relay_created_at, + valid_until_ms, + }; + self.advert_cache + .write() + .await + .insert(peer_npub.to_string(), updated); + self.failure_state.reset_streak_after_refresh(peer_npub); + NostrRefetchOutcome::Refreshed + } + } + } + pub async fn drain_events(&self) -> Vec { let mut out = Vec::new(); let mut rx = self.event_rx.lock().await; @@ -591,6 +718,7 @@ impl NostrDiscovery { } }; + let answer_received_at = now_ms(); debug!( peer = %peer_short, session = %short_id(&offer.session_id), @@ -599,14 +727,47 @@ impl NostrDiscovery { local = answer.payload.local_addresses.len(), "traversal: answer received" ); - validate_traversal_answer_for_offer( + if let Some(observed_skew_ms) = + estimate_clock_skew(&offer, &answer.payload, answer_received_at) + { + self.failure_state.note_observed_skew( + &peer_config.npub, + observed_skew_ms, + answer_received_at, + ); + let abs_skew = observed_skew_ms.unsigned_abs(); + // 30s threshold: well below the 60s SKEW_TOLERANCE wall but loud + // enough to surface a real clock problem on either side. + if abs_skew >= 30_000 { + debug!( + peer = %peer_short, + session = %short_id(&offer.session_id), + skew_ms = observed_skew_ms, + "traversal: significant peer clock skew observed" + ); + } else { + trace!( + peer = %peer_short, + skew_ms = observed_skew_ms, + "traversal: peer clock skew within nominal range" + ); + } + } + let outcome = validate_traversal_answer_for_offer( &offer, &answer.payload, - now_ms(), + answer_received_at, self.config.signal_ttl_secs * 1000, &answer.sender_npub, &self.npub, )?; + if outcome == FreshnessOutcome::FreshWithinSkewTolerance { + debug!( + peer = %peer_short, + session = %short_id(&offer.session_id), + "traversal: answer accepted within clock-skew tolerance" + ); + } if !answer.payload.accepted { return Err(BootstrapError::Protocol( answer @@ -643,6 +804,9 @@ impl NostrDiscovery { .publish_delete(&relays, [offer_event.id, answer.event_id]) .await; + self.failure_state + .record_success(&peer_config.npub, now_ms()); + Ok( EstablishedTraversal::new(session_id, peer_config.npub, remote_addr, base_socket) .with_transport_name("nostr-nat"), @@ -656,6 +820,7 @@ impl NostrDiscovery { sender_npub: String, ) -> Result<(), BootstrapError> { let peer_short = short_npub(&sender_npub); + let offer_received_at = now_ms(); debug!( peer = %peer_short, session = %short_id(&offer.session_id), @@ -663,13 +828,22 @@ impl NostrDiscovery { local = offer.local_addresses.len(), "traversal: offer received" ); - validate_offer_freshness( + let outcome = validate_offer_freshness( &offer, - now_ms(), + offer_received_at, self.config.signal_ttl_secs * 1000, &sender_npub, &self.npub, )?; + if outcome == FreshnessOutcome::FreshWithinSkewTolerance { + debug!( + peer = %peer_short, + session = %short_id(&offer.session_id), + offer_issued_at = offer.issued_at, + offer_received_at = offer_received_at, + "traversal: offer accepted within clock-skew tolerance" + ); + } self.mark_session_seen(&offer.session_id).await?; let base_socket = std::net::UdpSocket::bind(("0.0.0.0", 0))?; @@ -703,6 +877,7 @@ impl NostrDiscovery { stun_server, accepted.then(|| self.punch_hint()), (!accepted).then_some("no-usable-addresses".to_string()), + Some(offer_received_at), ); let relays = self.preferred_signal_relays(sender, None).await?; let answer_event = self.send_signal(&relays, sender, &answer).await?; @@ -1118,6 +1293,12 @@ impl NostrDiscovery { let config = NostrDiscoveryConfig::default(); let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers)); let (event_tx, event_rx) = mpsc::unbounded_channel(); + let failure_state = FailureState::new( + config.failure_streak_threshold, + config.extended_cooldown_secs, + config.warn_log_interval_secs, + config.failure_state_max_entries, + ); Self { client, keys, @@ -1135,6 +1316,7 @@ impl NostrDiscovery { event_rx: Mutex::new(event_rx), notify_task: Mutex::new(None), advertise_task: Mutex::new(None), + failure_state, } } diff --git a/src/discovery/nostr/signal.rs b/src/discovery/nostr/signal.rs index bf57053..b3d018b 100644 --- a/src/discovery/nostr/signal.rs +++ b/src/discovery/nostr/signal.rs @@ -6,6 +6,13 @@ use nostr::prelude::{ use super::types::{BootstrapError, PunchHint, SIGNAL_KIND, TraversalAnswer, TraversalOffer}; +/// Wall-clock skew tolerance applied to offer/answer freshness checks, in +/// milliseconds. Constant rather than configurable because loosening this +/// 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(super) struct SignalEnvelope { pub(super) payload: T, pub(super) event_id: EventId, @@ -75,23 +82,34 @@ pub(super) async fn unwrap_signal_event( }) } +/// Result of a freshness check. `Fresh` means the offer/answer is within the +/// strict TTL window; `FreshWithinSkewTolerance` means it was only accepted +/// after applying `FRESHNESS_SKEW_TOLERANCE_MS` grace, which is a useful +/// signal for operators to know clock skew is in play. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum FreshnessOutcome { + Fresh, + FreshWithinSkewTolerance, +} + pub(super) fn validate_offer_freshness( offer: &TraversalOffer, now: u64, signal_ttl_ms: u64, actual_sender_npub: &str, local_npub: &str, -) -> Result<(), BootstrapError> { +) -> Result { if offer.message_type != "offer" { return Err(BootstrapError::Protocol("invalid-offer".to_string())); } - if offer.expires_at <= now || now.saturating_sub(offer.issued_at) > signal_ttl_ms { - return Err(BootstrapError::Protocol("expired-offer".to_string())); - } + let outcome = match check_freshness(offer.issued_at, offer.expires_at, now, signal_ttl_ms) { + Some(o) => o, + None => return Err(BootstrapError::Protocol("expired-offer".to_string())), + }; if offer.sender_npub != actual_sender_npub || offer.recipient_npub != local_npub { return Err(BootstrapError::Protocol("identity-mismatch".to_string())); } - Ok(()) + Ok(outcome) } #[allow(clippy::too_many_arguments)] @@ -135,6 +153,7 @@ pub(super) fn create_traversal_answer( stun_server: Option, punch: Option, reason: Option, + offer_received_at: Option, ) -> TraversalAnswer { TraversalAnswer { message_type: "answer".to_string(), @@ -151,6 +170,7 @@ pub(super) fn create_traversal_answer( stun_server, punch, reason, + offer_received_at, } } @@ -161,16 +181,20 @@ pub(super) fn validate_traversal_answer_for_offer( signal_ttl_ms: u64, actual_sender_npub: &str, local_npub: &str, -) -> Result<(), BootstrapError> { +) -> Result { if answer.message_type != "answer" { return Err(BootstrapError::Protocol("invalid-answer".to_string())); } - if offer.expires_at <= now - || answer.expires_at <= now - || now.saturating_sub(answer.issued_at) > signal_ttl_ms + let offer_outcome = match check_freshness(offer.issued_at, offer.expires_at, now, signal_ttl_ms) { - return Err(BootstrapError::Protocol("expired-answer".to_string())); - } + Some(o) => o, + None => return Err(BootstrapError::Protocol("expired-answer".to_string())), + }; + let answer_outcome = + match check_freshness(answer.issued_at, answer.expires_at, now, signal_ttl_ms) { + Some(o) => o, + None => return Err(BootstrapError::Protocol("expired-answer".to_string())), + }; if offer.session_id != answer.session_id || answer.in_reply_to != offer.nonce { return Err(BootstrapError::Protocol("session-mismatch".to_string())); } @@ -189,5 +213,58 @@ pub(super) fn validate_traversal_answer_for_offer( "missing-rejection-reason".to_string(), )); } - Ok(()) + // Surface skew if either side was tolerated. The strict-Fresh case wins + // when both are strict; otherwise tolerance applied somewhere. + Ok( + if offer_outcome == FreshnessOutcome::Fresh && answer_outcome == FreshnessOutcome::Fresh { + FreshnessOutcome::Fresh + } else { + FreshnessOutcome::FreshWithinSkewTolerance + }, + ) +} + +/// NTP-style clock-skew estimate from a completed offer/answer round-trip. +/// Returns the responder's apparent offset relative to the initiator in +/// milliseconds (positive = responder clock ahead). Requires the responder +/// to have populated `answer.offer_received_at`; older responders won't, in +/// which case this returns `None`. +/// +/// Symmetric one-way-delay assumption (the standard NTP offset formula): +/// offset = ((T2 - T1) + (T3 - T4)) / 2 +/// where T1 = offer.issued_at, T2 = answer.offer_received_at, +/// T3 = answer.issued_at, T4 = answer_received_at. +pub(super) fn estimate_clock_skew( + offer: &TraversalOffer, + answer: &TraversalAnswer, + answer_received_at: u64, +) -> Option { + let t1 = offer.issued_at as i64; + let t2 = answer.offer_received_at? as i64; + let t3 = answer.issued_at as i64; + let t4 = answer_received_at as i64; + Some(((t2 - t1) + (t3 - t4)) / 2) +} + +/// Returns Some(outcome) if the (issued_at, expires_at) pair is acceptable +/// against `now` under the configured TTL plus `FRESHNESS_SKEW_TOLERANCE_MS` +/// of clock-skew grace on each side. Returns None if the message is +/// genuinely outside the tolerated window. +fn check_freshness( + issued_at: u64, + expires_at: u64, + now: u64, + signal_ttl_ms: u64, +) -> Option { + let strict_ok = expires_at > now && now.saturating_sub(issued_at) <= signal_ttl_ms; + if strict_ok { + return Some(FreshnessOutcome::Fresh); + } + let tolerated_ok = expires_at.saturating_add(FRESHNESS_SKEW_TOLERANCE_MS) > now + && now.saturating_sub(issued_at) <= signal_ttl_ms + FRESHNESS_SKEW_TOLERANCE_MS; + if tolerated_ok { + Some(FreshnessOutcome::FreshWithinSkewTolerance) + } else { + None + } } diff --git a/src/discovery/nostr/tests.rs b/src/discovery/nostr/tests.rs index ec7f77c..b412058 100644 --- a/src/discovery/nostr/tests.rs +++ b/src/discovery/nostr/tests.rs @@ -2,8 +2,8 @@ use nostr::prelude::{EventBuilder, Kind, Tag, Timestamp}; use super::runtime::NostrDiscovery; use super::signal::{ - build_signal_event, create_traversal_answer, create_traversal_offer, validate_offer_freshness, - validate_traversal_answer_for_offer, + FreshnessOutcome, build_signal_event, create_traversal_answer, create_traversal_offer, + estimate_clock_skew, validate_offer_freshness, validate_traversal_answer_for_offer, }; use super::stun::{parse_stun_binding_success, parse_stun_url}; use super::traversal::{ @@ -240,6 +240,7 @@ fn validates_offer_answer_pair() { duration_ms: 10_000, }), None, + Some(1_700_000_000_400), ); assert!( @@ -311,6 +312,7 @@ fn rejects_answer_with_mismatched_actual_sender() { duration_ms: 10_000, }), None, + Some(1_700_000_000_400), ); let result = validate_traversal_answer_for_offer( @@ -386,6 +388,172 @@ fn planned_remote_endpoints_include_private_and_reflexive_paths() { assert!(endpoints.contains(&"198.51.100.20:63000".parse().unwrap())); } +/// B4: strict-fresh path returns Fresh; the offer is well within TTL and +/// not expired. +#[test] +fn freshness_strict_returns_fresh_outcome() { + let offer = create_traversal_offer( + "sess-1".to_string(), + 1_700_000_000_000, + 60_000, + "offer-1".to_string(), + "npub1client".to_string(), + "npub1server".to_string(), + Some(addr("203.0.113.10", 62000)), + vec![addr("192.168.1.10", 62000)], + Some("stun:example.org:3478".to_string()), + ); + + let result = validate_offer_freshness( + &offer, + 1_700_000_000_500, + 60_000, + "npub1client", + "npub1server", + ) + .expect("strict-fresh offer should validate"); + assert_eq!(result, FreshnessOutcome::Fresh); +} + +/// B4: an offer whose `expires_at` has already passed by < SKEW_TOL is +/// accepted but flagged FreshWithinSkewTolerance — emulates the case where +/// the responder's clock is ahead of the initiator's. +#[test] +fn freshness_responder_clock_ahead_within_tolerance_is_tolerated() { + let offer = create_traversal_offer( + "sess-1".to_string(), + 1_700_000_000_000, + 60_000, // expires_at = 1_700_000_060_000 + "offer-1".to_string(), + "npub1client".to_string(), + "npub1server".to_string(), + Some(addr("203.0.113.10", 62000)), + vec![addr("192.168.1.10", 62000)], + None, + ); + + // now 90s past issued_at — 30s past strict expiry, but inside the 60s + // SKEW_TOL grace. + let result = validate_offer_freshness( + &offer, + 1_700_000_090_000, + 60_000, + "npub1client", + "npub1server", + ) + .expect("offer just past strict expiry should be tolerated"); + assert_eq!(result, FreshnessOutcome::FreshWithinSkewTolerance); +} + +/// B4: an offer beyond TTL + SKEW_TOL is rejected as expired. +#[test] +fn freshness_responder_clock_far_ahead_is_rejected() { + let offer = create_traversal_offer( + "sess-1".to_string(), + 1_700_000_000_000, + 60_000, + "offer-1".to_string(), + "npub1client".to_string(), + "npub1server".to_string(), + Some(addr("203.0.113.10", 62000)), + vec![addr("192.168.1.10", 62000)], + None, + ); + + // 130s past issued_at: 70s past strict expiry, 10s past tolerated expiry. + let err = validate_offer_freshness( + &offer, + 1_700_000_130_000, + 60_000, + "npub1client", + "npub1server", + ) + .expect_err("offer past tolerated expiry should be rejected"); + assert!(err.to_string().contains("expired-offer"), "{}", err); +} + +/// B5a: the NTP-style skew estimator returns the responder's apparent +/// clock offset relative to the initiator. Symmetric one-way delays of +/// 50ms each plus a +500ms responder skew should yield ≈+500ms. +#[test] +fn estimate_clock_skew_matches_responder_offset() { + // T1 (initiator sent) + let offer = create_traversal_offer( + "sess-1".to_string(), + 1_700_000_000_000, + 60_000, + "offer-1".to_string(), + "npub1client".to_string(), + "npub1server".to_string(), + None, + vec![addr("192.168.1.10", 62000)], + None, + ); + // Wire takes 50ms, responder clock is +500ms ahead, so: + // T2 = 1_700_000_000_000 + 50 + 500 = 1_700_000_000_550 + // T3 = 1_700_000_000_550 (no processing time for this synthetic case) + // T4 = T1 + 50 + (T3 - T2 + 500_skew_corrected) + 50 wire return + // For simplicity: T4 = T1 + 100ms wire + 0 responder processing + // = 1_700_000_000_100 (initiator wall clock) + let answer = create_traversal_answer( + "sess-1".to_string(), + 1_700_000_000_550, // T3 + 60_000, + "answer-1".to_string(), + "npub1server".to_string(), + "npub1client".to_string(), + "offer-1".to_string(), + true, + Some(addr("198.51.100.20", 63000)), + vec![], + None, + None, + None, + Some(1_700_000_000_550), // T2 + ); + let answer_received_at = 1_700_000_000_100; // T4 + + let skew = estimate_clock_skew(&offer, &answer, answer_received_at) + .expect("offer_received_at populated -> Some"); + // ((550 - 0) + (550 - 100)) / 2 = (550 + 450) / 2 = 500 + assert_eq!(skew, 500); +} + +/// B5a: backward-compat — when the responder did not populate +/// `offer_received_at` (older daemon), skew estimation returns None +/// and callers should silently skip logging it. +#[test] +fn estimate_clock_skew_returns_none_without_responder_timestamp() { + let offer = create_traversal_offer( + "sess-1".to_string(), + 1_700_000_000_000, + 60_000, + "offer-1".to_string(), + "npub1client".to_string(), + "npub1server".to_string(), + None, + vec![], + None, + ); + let answer = create_traversal_answer( + "sess-1".to_string(), + 1_700_000_000_500, + 60_000, + "answer-1".to_string(), + "npub1server".to_string(), + "npub1client".to_string(), + "offer-1".to_string(), + true, + Some(addr("198.51.100.20", 63000)), + vec![], + None, + None, + None, + None, // older responder + ); + assert!(estimate_clock_skew(&offer, &answer, 1_700_000_000_900).is_none()); +} + #[tokio::test] async fn signal_events_use_current_timestamps() { let sender = nostr::Keys::generate(); diff --git a/src/discovery/nostr/types.rs b/src/discovery/nostr/types.rs index 592444e..3d76419 100644 --- a/src/discovery/nostr/types.rs +++ b/src/discovery/nostr/types.rs @@ -164,6 +164,16 @@ pub struct TraversalAnswer { pub stun_server: Option, pub punch: Option, pub reason: Option, + /// Responder's local wall-clock (Unix ms) at the moment it received the + /// offer. Optional / non-breaking: the initiator uses this to derive an + /// NTP-style clock-skew estimate against the offer's `issued_at`. Older + /// responders that don't fill this in still produce valid answers. + #[serde( + rename = "offerReceivedAt", + default, + skip_serializing_if = "Option::is_none" + )] + pub offer_received_at: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -178,3 +188,35 @@ pub struct PunchPacket { pub sequence: u32, pub session_hash: [u8; 16], } + +/// Outcome of `NostrDiscovery::record_traversal_failure`. +#[derive(Debug, Clone, Copy)] +pub struct NostrFailureDecision { + pub consecutive_failures: u32, + /// True iff the lifecycle should log at WARN; false → DEBUG (rate-limit). + pub should_warn: bool, + /// Wall-clock ms before which retries should not fire, if cooldown + /// is in effect. + pub cooldown_until_ms: Option, + /// True only on the streak-threshold-crossing transition. Lifecycle + /// should run a one-shot stale-advert re-check (B6) when set. + pub crossed_threshold: bool, +} + +/// Snapshot row for `show_peers` rendering of per-npub Nostr-traversal state. +#[derive(Debug, Clone)] +pub struct NostrPeerFailureView { + pub npub: String, + pub consecutive_failures: u32, + pub cooldown_until_ms: Option, + pub last_observed_skew_ms: Option, +} + +/// Outcome of `NostrDiscovery::refetch_advert_for_stale_check` (B6). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NostrRefetchOutcome { + Evicted, + Refreshed, + SameAdvert, + Skipped, +} diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index b98f7fd..55b546d 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -407,7 +407,56 @@ impl Node { peer_config, reason, } => { - warn!(npub = %peer_config.npub, error = %reason, "NAT traversal failed"); + let now_ms = Self::now_ms(); + let decision = bootstrap.record_traversal_failure(&peer_config.npub, now_ms); + if decision.should_warn { + warn!( + npub = %peer_config.npub, + error = %reason, + consecutive_failures = decision.consecutive_failures, + cooldown_secs = decision + .cooldown_until_ms + .map(|t| t.saturating_sub(now_ms) / 1000), + "NAT traversal failed" + ); + } else { + debug!( + npub = %peer_config.npub, + error = %reason, + consecutive_failures = decision.consecutive_failures, + "NAT traversal failed (suppressed by warn-rate-limit)" + ); + } + + // B6: stale-advert eviction on the streak-threshold + // crossing. Fire-and-forget; the outcome is logged so + // operators can see when peers get cleaned up. + if decision.crossed_threshold { + let bootstrap = bootstrap.clone(); + let npub = peer_config.npub.clone(); + tokio::spawn(async move { + let outcome = bootstrap.refetch_advert_for_stale_check(&npub).await; + match outcome { + crate::discovery::nostr::NostrRefetchOutcome::Evicted => info!( + npub = %npub, + "stale-advert sweep: peer evicted from advert cache" + ), + crate::discovery::nostr::NostrRefetchOutcome::Refreshed => info!( + npub = %npub, + "stale-advert sweep: peer republished, cache refreshed and streak reset" + ), + crate::discovery::nostr::NostrRefetchOutcome::SameAdvert => debug!( + npub = %npub, + "stale-advert sweep: advert unchanged, cooldown stands" + ), + crate::discovery::nostr::NostrRefetchOutcome::Skipped => debug!( + npub = %npub, + "stale-advert sweep: skipped (relay error or no advert_relays)" + ), + } + }); + } + let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) { Ok(identity) => identity, Err(_) => continue, @@ -421,7 +470,16 @@ impl Node { continue; } - self.schedule_retry(*peer_identity.node_addr(), Self::now_ms()); + let node_addr = *peer_identity.node_addr(); + self.schedule_retry(node_addr, now_ms); + if let Some(cooldown_until_ms) = decision.cooldown_until_ms + && let Some(state) = self.retry_pending.get_mut(&node_addr) + { + // Push the next retry past the cooldown so the + // open-discovery sweep doesn't re-enqueue and the + // per-attempt backoff doesn't fire sooner. + state.retry_after_ms = state.retry_after_ms.max(cooldown_until_ms); + } } } } @@ -1204,6 +1262,7 @@ impl Node { let mut skipped_connecting = 0usize; let mut skipped_no_endpoints = 0usize; let mut skipped_invalid_npub = 0usize; + let mut skipped_cooldown = 0usize; for (npub, endpoints, created_at_secs) in candidates { if enqueue_budget == 0 { @@ -1242,6 +1301,10 @@ impl Node { skipped_retry_pending = skipped_retry_pending.saturating_add(1); continue; } + if bootstrap.cooldown_until(&npub, now_ms).is_some() { + skipped_cooldown = skipped_cooldown.saturating_add(1); + continue; + } let connecting = self.connections.values().any(|conn| { conn.expected_identity() .map(|id| id.node_addr() == &node_addr) @@ -1309,7 +1372,8 @@ impl Node { + skipped_retry_pending + skipped_connecting + skipped_no_endpoints - + skipped_invalid_npub; + + skipped_invalid_npub + + skipped_cooldown; let should_summarize = caller == "startup" || enqueued > 0; if should_summarize { info!( @@ -1324,6 +1388,7 @@ impl Node { skipped_connecting = skipped_connecting, skipped_no_endpoints = skipped_no_endpoints, skipped_invalid_npub = skipped_invalid_npub, + skipped_cooldown = skipped_cooldown, skipped_total = total_skipped, "open-discovery sweep complete" ); diff --git a/src/node/mod.rs b/src/node/mod.rs index 3d95d47..7744546 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1539,6 +1539,13 @@ impl Node { self.peers.values() } + /// Reference to the Nostr discovery handle if discovery is enabled. + /// Used by control queries (`show_peers` per-peer Nostr-traversal + /// state) to read failure-state without taking shared ownership. + pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> { + self.nostr_discovery.as_deref() + } + /// Iterate over all peer node IDs. pub fn peer_ids(&self) -> impl Iterator { self.peers.keys() From 2f95929862e4aed76715b68a6945d572a9654383 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 4 May 2026 13:57:22 +0000 Subject: [PATCH 2/2] nostr: public-IP discovery for UDP/TCP advert publication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a UDP transport had `advertise_on_nostr: true` + `public: true` + `bind_addr: 0.0.0.0:NNNN`, the advert builder previously read the kernel's `local_addr()`, found `0.0.0.0`, filtered it out (correctly — wildcard isn't a valid advertised endpoint), and silently emitted no UDP endpoint in the published advert. Operators on AWS EIP / GCP / Azure setups (where binding to the public IP directly is impossible because 1:1 NAT does the address translation off-host) had no way to advertise UDP without binding to a specific local IP — and no log explaining what was happening. TCP had the same shape, with no `public: true` precondition. Three pieces, layered. UDP gets zero-config autodiscovery via STUN; both UDP and TCP get an explicit operator-supplied override; the fall-through path now logs loudly instead of silently skipping. UDP public-IP autodiscovery (STUN) ---------------------------------- In the UDP `is_public()` + wildcard-bind branch, run a one-shot STUN observation against an ephemeral UDP socket on the daemon's configured `stun_servers`. Take the reflexive IPv4 (the STUN-reported port is the ephemeral source port and is discarded), combine with the configured listener port for the advert (`udp::`). Works on AWS EIP / GCP / Azure 1:1-NAT setups because STUN sees the public-Internet egress IP and the bind port is preserved through 1:1 NAT. Result is cached per-transport on a new `public_udp_addr_cache` field on `NostrDiscovery` (keyed by `TransportId.as_u32()`). Asymmetric cache TTL: a successful observation is cached for `advert_refresh_secs` (default 30 min) so we don't STUN every refresh tick. A failed observation is cached for only 60s (`PUBLIC_UDP_ADDR_FAILURE_TTL`) so a transient STUN flake at startup retries within ~a minute and the advert grows its UDP endpoint as soon as STUN starts working — rather than waiting the full 30-min cycle. The shared `observe_traversal_addresses` STUN helper had a hard-coded 2s per-server response wait, right for the per-traversal flow (latency-sensitive — 3 STUN servers worst-case = 6s) but too short for the one-shot advert-publish startup discovery. Parameterized `per_server_timeout` on the helper, with two named constants in `stun.rs`: `TRAVERSAL_STUN_TIMEOUT = 2s` (existing call sites) and `ADVERT_STUN_TIMEOUT = 5s` (new public-UDP discovery path). Both use `tokio::time::timeout_at` under the hood, so success returns immediately — the timeout is only the worst case. `external_addr` override (UDP + TCP) ------------------------------------ New `external_addr: Option` field on `transports.udp.*` and `transports.tcp.*` for explicit advertise-as override. Takes precedence over both the bound `local_addr` and (for UDP) the STUN-derived autodiscovery. Required for TCP on cloud-NAT setups (AWS EIP, GCP/Azure external IPs) where binding to the public IP directly fails with `EADDRNOTAVAIL` because the public IP isn't on a host interface — the network fabric does 1:1 NAT off-host. Without this field the operator's only TCP path was "leave advert off" or "find a way to make the public IP locally bindable." For UDP, `external_addr` is optional but useful as a deterministic alternative to STUN. Operators who want to skip STUN egress, whose STUN servers are blocked, or who want the daemon to not depend on external services for advert content can specify it explicitly. The accessor parses two shapes: - Bare IP (`"54.183.70.180"` or `"2001:db8::1"`): combines with the configured `bind_addr` port. - Full host:port (`"54.183.70.180:8443"` or `"[2001:db8::1]:443"`): used verbatim — useful for port-forward setups where the externally-visible port differs from the bind port. Final precedence in `Node::build_overlay_advert` (now async, only caller `refresh_overlay_advert` was already async): - UDP: `external_addr` → non-wildcard `local_addr` → STUN → loud warn - TCP: `external_addr` → non-wildcard `local_addr` → loud warn Loud warns instead of silent skips ---------------------------------- The wildcard-bind fall-through paths now log a `warn!` pointing at the operator-side fixes: - UDP: "set transports.udp.external_addr, bind to a specific public IP, or ensure node.discovery.nostr.stun_servers is reachable" - TCP: "Either set external_addr to the public IP (recommended for cloud 1:1-NAT setups) or bind explicitly to the public IP" Replaces the silent skip that previously cost operators a debugging session when the advert mysteriously contained only the Tor onion endpoint. Tests ----- 11 new unit tests in `src/config/transport.rs` covering the parser (IPv4/IPv6, bare/full, malformed) and the accessor (UDP with default bind, UDP with explicit port override, UDP unset, TCP without bind_addr, TCP with bind_addr, TCP with full socket-addr override, parse_bind_port for IPv4/IPv6/malformed). The 38-test nostr suite still passes. CHANGELOG entries under `[Unreleased]` Fixed. --- CHANGELOG.md | 41 +++++++++ src/config/transport.rs | 162 +++++++++++++++++++++++++++++++++ src/discovery/nostr/runtime.rs | 132 ++++++++++++++++++++++++++- src/discovery/nostr/stun.rs | 18 +++- src/node/lifecycle.rs | 84 +++++++++++++++-- 5 files changed, 424 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b27e449..4efeef0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -274,6 +274,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- UDP transport with `advertise_on_nostr: true` + `public: true` + + a wildcard `bind_addr` (e.g. `0.0.0.0:2121`) is now advertised + with its STUN-discovered public IPv4 instead of being silently + dropped from the published Kind 37195 advert. Previously the + advert builder filtered the wildcard out (since `0.0.0.0` is + not a valid endpoint), but emitted no log explaining what + happened — operators saw the daemon up, both flags set, and + no UDP endpoint in the advert. The fix runs a one-shot STUN + observation against an ephemeral socket on the daemon's + configured `stun_servers` and combines the reflexive IPv4 with + the configured listener port for the advert (`udp::`). + Successful STUN observations are cached per-transport for one + `advert_refresh_secs` cycle (default 30 min) so we don't re-STUN + every refresh. Failed observations are cached for only 60s, so + a transient STUN flake at startup retries within ~a minute and + grows the advert with UDP as soon as STUN starts working — + rather than waiting the full 30-min cycle. Per-server STUN + response timeout is 5s for the advert-publish path (vs. 2s for + the latency-sensitive per-traversal path), giving slow + first-call STUN time to complete without giving up. On STUN + failure, the wildcard-bind path still skips, but now logs a + loud `warn!` pointing at the operator-side fixes (set + `external_addr`, bind to a specific IP, or ensure `stun_servers` + reachable). Restores zero-config public-IP autodiscovery on + AWS EIP / GCP / Azure setups where binding to the public IP + directly is impossible (1:1 NAT) +- New `external_addr` field on `transports.udp.*` and + `transports.tcp.*` for explicit advertise-as override. Accepts + either a bare IP (`"54.183.70.180"` — the configured `bind_addr` + port is appended) or a full `host:port` + (`"54.183.70.180:8443"`). Takes precedence over both the bound + address and any STUN-derived autodiscovery. Required for TCP + on cloud-NAT setups (AWS EIP, GCP/Azure external IPs) where + binding to the public IP directly fails with `EADDRNOTAVAIL` + (the EIP isn't on a host interface). Optional but useful for + UDP as a deterministic alternative to STUN — operators who + want to skip STUN egress (or whose STUN is blocked) can + specify it explicitly. Without `external_addr`, TCP with a + wildcard `bind_addr` + `advertise_on_nostr: true` now logs a + loud `warn!` pointing at the two fixes instead of silently + skipping - Nostr-discovery now tolerates ±60s of clock skew on offer/answer freshness checks so a responder whose wall clock leads the initiator's by less than that no longer silently rejects every diff --git a/src/config/transport.rs b/src/config/transport.rs index af996f6..2281ee4 100644 --- a/src/config/transport.rs +++ b/src/config/transport.rs @@ -4,9 +4,31 @@ //! transport-specific configuration structs. use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr}; use serde::{Deserialize, Serialize}; +/// Parse an `external_addr` config string against a known bind port, +/// producing the absolute `SocketAddr` to advertise on Nostr. +/// +/// Accepts either a bare IP (`"54.183.70.180"` or `"[::1]"`) — in which +/// case the bind port is appended — or a full `host:port` form +/// (`"54.183.70.180:443"` or `"[::1]:443"`). Returns `None` on any parse +/// error. IPv6 must use bracket notation when supplying a port. +fn parse_external_advert_addr(raw: &str, bind_port: u16) -> Option { + if let Ok(sa) = raw.parse::() { + return Some(sa); + } + let ip: IpAddr = raw.parse().ok()?; + Some(SocketAddr::new(ip, bind_port)) +} + +/// Extract the port from a `bind_addr` string. Returns `None` if the +/// string can't be parsed (e.g. a bare hostname without port). +fn parse_bind_port(raw: &str) -> Option { + raw.parse::().ok().map(|sa| sa.port()) +} + /// Default UDP bind address. const DEFAULT_UDP_BIND_ADDR: &str = "0.0.0.0:2121"; @@ -54,6 +76,16 @@ pub struct UdpConfig { /// Default: false. #[serde(default, skip_serializing_if = "Option::is_none")] pub public: Option, + /// Optional explicit public address to advertise when `public: true` + /// is set. Takes precedence over both the bound address and any + /// STUN-derived autodiscovery. Accepts either a bare IP + /// (`"54.183.70.180"` — the configured `bind_addr` port is appended) + /// or a full `host:port` (`"54.183.70.180:443"`). Useful when the + /// public IP isn't on a local interface (e.g. AWS EIP / cloud 1:1 + /// NAT) and the operator wants to skip STUN autodiscovery for a + /// deterministic value. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub external_addr: Option, /// Outbound-only mode. When true, the transport binds to a kernel- /// assigned ephemeral port (`0.0.0.0:0`) instead of the configured /// `bind_addr`, refuses inbound handshake msg1, and is never @@ -119,6 +151,16 @@ impl UdpConfig { self.public.unwrap_or(false) } + /// Parse `external_addr` against the configured `bind_addr` port, + /// returning the absolute `SocketAddr` to advertise on Nostr. + /// Returns `None` if `external_addr` is unset or malformed, or if + /// the port cannot be inferred. + pub fn external_advert_addr(&self) -> Option { + let raw = self.external_addr.as_deref()?; + let bind_port = parse_bind_port(self.bind_addr())?; + parse_external_advert_addr(raw, bind_port) + } + /// Whether this transport runs in outbound-only mode. Default: false. pub fn outbound_only(&self) -> bool { self.outbound_only.unwrap_or(false) @@ -366,6 +408,16 @@ pub struct TcpConfig { /// Default: false. #[serde(default, skip_serializing_if = "Option::is_none")] pub advertise_on_nostr: Option, + + /// Optional explicit public address to advertise. Required when + /// `bind_addr` is wildcard (e.g. `"0.0.0.0:443"`) and + /// `advertise_on_nostr: true`, since TCP has no STUN equivalent + /// for autodiscovery. Accepts either a bare IP (`"54.183.70.180"` + /// — the configured `bind_addr` port is appended) or a full + /// `host:port`. Common pattern on AWS EIP / cloud 1:1 NAT setups + /// where the public IP isn't bindable on the host. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub external_addr: Option, } impl TcpConfig { @@ -410,6 +462,16 @@ impl TcpConfig { pub fn advertise_on_nostr(&self) -> bool { self.advertise_on_nostr.unwrap_or(false) } + + /// Parse `external_addr` against the configured `bind_addr` port, + /// returning the absolute `SocketAddr` to advertise on Nostr. + /// Returns `None` if `external_addr` is unset or malformed, or if + /// `bind_addr` is unset / unparseable so no port can be inferred. + pub fn external_advert_addr(&self) -> Option { + let raw = self.external_addr.as_deref()?; + let bind_port = parse_bind_port(self.bind_addr.as_deref()?)?; + parse_external_advert_addr(raw, bind_port) + } } // ============================================================================ @@ -804,3 +866,103 @@ impl TransportsConfig { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_external_addr_accepts_bare_ipv4_with_appended_bind_port() { + let sa = parse_external_advert_addr("54.183.70.180", 2121).unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:2121"); + } + + #[test] + fn parse_external_addr_accepts_full_ipv4_socket_addr() { + let sa = parse_external_advert_addr("54.183.70.180:443", 2121).unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:443"); + // Explicit port wins over the bind port we passed in. + } + + #[test] + fn parse_external_addr_accepts_bare_ipv6_with_appended_bind_port() { + let sa = parse_external_advert_addr("2001:db8::1", 443).unwrap(); + assert_eq!(sa.to_string(), "[2001:db8::1]:443"); + } + + #[test] + fn parse_external_addr_accepts_bracketed_ipv6_with_explicit_port() { + let sa = parse_external_advert_addr("[2001:db8::1]:8443", 443).unwrap(); + assert_eq!(sa.to_string(), "[2001:db8::1]:8443"); + } + + #[test] + fn parse_external_addr_rejects_garbage() { + assert!(parse_external_advert_addr("not-an-ip", 443).is_none()); + assert!(parse_external_advert_addr("", 443).is_none()); + } + + #[test] + fn udp_external_advert_addr_combines_with_bind_port_default() { + let cfg = UdpConfig { + external_addr: Some("54.183.70.180".to_string()), + ..UdpConfig::default() + }; + // bind_addr unset, so default DEFAULT_UDP_BIND_ADDR (0.0.0.0:2121) applies. + let sa = cfg.external_advert_addr().unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:2121"); + } + + #[test] + fn udp_external_advert_addr_with_explicit_full_socket_addr_overrides_bind_port() { + let cfg = UdpConfig { + bind_addr: Some("0.0.0.0:2121".to_string()), + external_addr: Some("54.183.70.180:9999".to_string()), + ..UdpConfig::default() + }; + let sa = cfg.external_advert_addr().unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:9999"); + } + + #[test] + fn udp_external_advert_addr_returns_none_when_unset() { + let cfg = UdpConfig::default(); + assert!(cfg.external_advert_addr().is_none()); + } + + #[test] + fn tcp_external_advert_addr_requires_bind_port() { + let cfg = TcpConfig { + external_addr: Some("54.183.70.180".to_string()), + ..TcpConfig::default() + }; + // bind_addr unset → no port to combine with → None. + assert!(cfg.external_advert_addr().is_none()); + + let cfg = TcpConfig { + bind_addr: Some("0.0.0.0:443".to_string()), + external_addr: Some("54.183.70.180".to_string()), + ..TcpConfig::default() + }; + let sa = cfg.external_advert_addr().unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:443"); + } + + #[test] + fn tcp_external_advert_addr_with_full_socket_addr_independent_of_bind() { + let cfg = TcpConfig { + bind_addr: Some("0.0.0.0:443".to_string()), + external_addr: Some("54.183.70.180:8443".to_string()), + ..TcpConfig::default() + }; + let sa = cfg.external_advert_addr().unwrap(); + assert_eq!(sa.to_string(), "54.183.70.180:8443"); + } + + #[test] + fn parse_bind_port_extracts_from_socket_addr_strings() { + assert_eq!(parse_bind_port("0.0.0.0:2121"), Some(2121)); + assert_eq!(parse_bind_port("[::]:443"), Some(443)); + assert_eq!(parse_bind_port("not-a-socket-addr"), None); + } +} diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index a08bf88..4a64300 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -1,6 +1,7 @@ use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use nostr::nips::nip17; use nostr::nips::nip19::ToBech32; @@ -56,6 +57,25 @@ fn endpoint_summary(endpoints: &[OverlayEndpointAdvert]) -> String { .join(",") } +/// Cached STUN-derived public address for an advert-eligible UDP transport +/// bound to a wildcard. Lives on `NostrDiscovery` so the freshness window +/// survives advert refresh cycles. +struct CachedPublicUdpAddr { + /// Most recent STUN observation. `None` means the last attempt failed + /// (recorded so we don't re-spam STUN every refresh tick on broken + /// network conditions). + addr: Option, + fetched_at: Instant, +} + +/// Cache lifetime for a *failed* STUN observation. Held briefly so that +/// transient flakes (slow startup network, momentary STUN-server +/// blip) get retried within ~a minute and the advert grows its UDP +/// endpoint as soon as STUN starts working — rather than waiting a +/// full `advert_refresh_secs` (30 min) for the success-path TTL to +/// expire. Successful results use the longer per-config TTL. +const PUBLIC_UDP_ADDR_FAILURE_TTL: Duration = Duration::from_secs(60); + pub struct NostrDiscovery { client: Client, keys: nostr::Keys, @@ -74,6 +94,10 @@ pub struct NostrDiscovery { notify_task: Mutex>>, advertise_task: Mutex>>, failure_state: FailureState, + /// STUN-derived public address per advert-eligible UDP transport + /// (keyed by `TransportId.as_u32()`). Populated on demand by + /// `learn_public_udp_addr()` and refreshed by TTL. + public_udp_addr_cache: RwLock>, } impl NostrDiscovery { @@ -133,6 +157,7 @@ impl NostrDiscovery { notify_task: Mutex::new(None), advertise_task: Mutex::new(None), failure_state, + public_udp_addr_cache: RwLock::new(HashMap::new()), }); runtime.subscribe().await?; @@ -204,6 +229,108 @@ impl NostrDiscovery { .collect() } + /// Discover (or return cached) the public-Internet address for an + /// advert-eligible UDP transport bound to a wildcard. Used by + /// `build_overlay_advert` to avoid emitting `udp:0.0.0.0:port`, + /// which is invalid as an advertised endpoint. Result is the + /// reflexive IP (from STUN against the daemon's first + /// `stun_servers` reachable) combined with the configured + /// `advertise_port`. + /// + /// Asymmetric cache TTL: a successful observation is cached for + /// `advert_refresh_secs` (default 1800 = same as advert refresh) + /// so we don't re-STUN every refresh tick. A failed observation + /// is cached for `PUBLIC_UDP_ADDR_FAILURE_TTL` (60s) so we retry + /// soon after a transient STUN flake at startup, instead of + /// blocking advertise-as-public for half an hour. Once a success + /// is cached, subsequent ticks are zero-overhead. + pub async fn learn_public_udp_addr( + &self, + transport_id_key: u32, + advertise_port: u16, + ) -> Option { + if let Some(entry) = self + .public_udp_addr_cache + .read() + .await + .get(&transport_id_key) + { + let ttl = if entry.addr.is_some() { + Duration::from_secs(self.config.advert_refresh_secs.max(60)) + } else { + PUBLIC_UDP_ADDR_FAILURE_TTL + }; + if entry.fetched_at.elapsed() < ttl { + return entry.addr; + } + } + let resolved = self.stun_observe_public_ip(advertise_port).await; + let mut cache = self.public_udp_addr_cache.write().await; + cache.insert( + transport_id_key, + CachedPublicUdpAddr { + addr: resolved, + fetched_at: Instant::now(), + }, + ); + resolved + } + + /// Run a one-shot STUN observation against an ephemeral UDP socket + /// to learn this host's public IPv4 (or IPv6, if the local STUN + /// server returns one). Returns `:`, + /// or `None` if STUN failed or no `stun_servers` are configured. + /// + /// The STUN-reported port is the ephemeral source port and is + /// discarded — what we want to advertise is the bound listener + /// port, which the kernel preserves through 1:1 NAT (AWS EIP, + /// GCP/Azure external IPs) and which the operator has explicitly + /// chosen via `bind_addr`. + async fn stun_observe_public_ip(&self, advertise_port: u16) -> Option { + if self.config.stun_servers.is_empty() { + return None; + } + let socket = match std::net::UdpSocket::bind("0.0.0.0:0") { + Ok(s) => s, + Err(err) => { + debug!(error = %err, "public-udp-addr: ephemeral bind failed"); + return None; + } + }; + if let Err(err) = socket.set_nonblocking(true) { + debug!(error = %err, "public-udp-addr: set_nonblocking failed"); + return None; + } + let observed = match super::stun::observe_traversal_addresses( + &socket, + &self.config.stun_servers, + false, + super::stun::ADVERT_STUN_TIMEOUT, + ) + .await + { + Ok((reflexive, _local, stun_server)) => { + debug!( + stun = %stun_server.as_deref().unwrap_or("-"), + reflexive = %reflexive + .as_ref() + .map(|a| format!("{}:{}", a.ip, a.port)) + .unwrap_or_else(|| "-".into()), + "public-udp-addr: STUN observation" + ); + reflexive + } + Err(err) => { + debug!(error = %err, "public-udp-addr: STUN failed"); + return None; + } + }; + observed.and_then(|addr| { + let parsed_ip: std::net::IpAddr = addr.ip.parse().ok()?; + Some(SocketAddr::new(parsed_ip, advertise_port)) + }) + } + /// Stale-advert re-check (B6). Called by lifecycle on the /// streak-threshold transition. Actively re-queries the peer's /// Kind 37195 advert from `advert_relays`; evicts the cache entry @@ -663,6 +790,7 @@ impl NostrDiscovery { &base_socket, &self.config.stun_servers, self.config.share_local_candidates, + super::stun::TRAVERSAL_STUN_TIMEOUT, ) .await?; debug!( @@ -852,6 +980,7 @@ impl NostrDiscovery { &base_socket, &self.config.stun_servers, self.config.share_local_candidates, + super::stun::TRAVERSAL_STUN_TIMEOUT, ) .await?; let accepted = reflexive_address.is_some() || !local_addresses.is_empty(); @@ -1317,6 +1446,7 @@ impl NostrDiscovery { notify_task: Mutex::new(None), advertise_task: Mutex::new(None), failure_state, + public_udp_addr_cache: RwLock::new(HashMap::new()), } } diff --git a/src/discovery/nostr/stun.rs b/src/discovery/nostr/stun.rs index c51743b..3aa7a14 100644 --- a/src/discovery/nostr/stun.rs +++ b/src/discovery/nostr/stun.rs @@ -11,10 +11,23 @@ use super::types::{BootstrapError, TraversalAddress}; // Local interface discovery remains best-effort and may still be incomplete // on dual-stack, NAT64, or heavily firewalled hosts. +/// Default per-server STUN response wait used by the per-traversal flow. +/// Latency-sensitive: keep tight so a misbehaving STUN server doesn't +/// stretch every traversal attempt. +pub(super) const TRAVERSAL_STUN_TIMEOUT: Duration = Duration::from_secs(2); + +/// Per-server STUN response wait used by the advert-publish path's +/// public-IP discovery. Longer than `TRAVERSAL_STUN_TIMEOUT` because +/// it's a one-shot at startup (cached afterward) and we'd rather block +/// the first advert build by a few seconds than skip UDP advertising +/// over a slow first response. Returns immediately on success. +pub(super) const ADVERT_STUN_TIMEOUT: Duration = Duration::from_secs(5); + pub(super) async fn observe_traversal_addresses( socket: &std::net::UdpSocket, stun_servers: &[String], share_local_candidates: bool, + per_server_timeout: Duration, ) -> Result< ( Option, @@ -39,7 +52,7 @@ pub(super) async fn observe_traversal_addresses( let mut last_error = None; for stun_server in stun_servers { - match perform_stun(socket, stun_server).await { + match perform_stun(socket, stun_server, per_server_timeout).await { Ok(mapped) => { debug!( stun_server = %stun_server, @@ -70,6 +83,7 @@ pub(super) async fn observe_traversal_addresses( async fn perform_stun( socket: &std::net::UdpSocket, stun_server: &str, + response_timeout: Duration, ) -> Result, BootstrapError> { let endpoint = parse_stun_url(stun_server)?; let txn_id = random_txn_id(); @@ -80,7 +94,7 @@ async fn perform_stun( let udp = UdpSocket::from_std(socket.try_clone()?)?; udp.send_to(&request, addr).await?; let mut buf = [0u8; 2048]; - let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let deadline = tokio::time::Instant::now() + response_timeout; loop { let result = tokio::time::timeout_at(deadline, udp.recv_from(&mut buf)).await; let Ok(Ok((len, _remote))) = result else { diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 55b546d..5d7105f 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -1487,7 +1487,10 @@ impl Node { ) } - fn build_overlay_advert(&self) -> Option { + async fn build_overlay_advert( + &self, + bootstrap: &std::sync::Arc, + ) -> Option { if !self.config.node.discovery.nostr.enabled { return None; } @@ -1509,13 +1512,49 @@ impl Node { continue; } if cfg.is_public() { - if let Some(addr) = handle.local_addr() - && !addr.ip().is_unspecified() - { + // Precedence: + // 1. operator-supplied `external_addr` (skips STUN) + // 2. non-wildcard `local_addr` (operator bound to + // a specific public IP directly) + // 3. STUN auto-discovery against ephemeral socket + // 4. loud warn + omit endpoint + if let Some(explicit) = cfg.external_advert_addr() { endpoints.push(OverlayEndpointAdvert { transport: OverlayTransportKind::Udp, - addr: addr.to_string(), + addr: explicit.to_string(), }); + } else { + match handle.local_addr() { + Some(addr) if !addr.ip().is_unspecified() => { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: addr.to_string(), + }); + } + Some(addr) => { + let key = handle.transport_id().as_u32(); + let port = addr.port(); + if let Some(public) = + bootstrap.learn_public_udp_addr(key, port).await + { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: public.to_string(), + }); + } else { + warn!( + transport_id = key, + bind_addr = %addr, + "advert: udp public=true bound to wildcard but \ + STUN observation failed; advertising no UDP \ + endpoint. Either set transports.udp.external_addr, \ + bind to a specific public IP, or ensure \ + node.discovery.nostr.stun_servers is reachable" + ); + } + } + None => {} + } } } else { endpoints.push(OverlayEndpointAdvert { @@ -1532,13 +1571,38 @@ impl Node { if !cfg.advertise_on_nostr() { continue; } - if let Some(addr) = handle.local_addr() - && !addr.ip().is_unspecified() - { + // Precedence: + // 1. operator-supplied `external_addr` (only path that + // works on cloud-NAT setups where the public IP is + // not on a host interface). + // 2. non-wildcard `local_addr` (operator bound to a + // specific public IP directly). + // 3. loud warn + omit endpoint (no TCP STUN equivalent). + if let Some(explicit) = cfg.external_advert_addr() { endpoints.push(OverlayEndpointAdvert { transport: OverlayTransportKind::Tcp, - addr: addr.to_string(), + addr: explicit.to_string(), }); + } else { + match handle.local_addr() { + Some(addr) if !addr.ip().is_unspecified() => { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Tcp, + addr: addr.to_string(), + }); + } + Some(addr) => { + warn!( + bind_addr = %addr, + "advert: tcp advertise_on_nostr=true bound to wildcard \ + and no transports.tcp.external_addr set; advertising no \ + TCP endpoint. Either set external_addr to the public \ + IP (recommended for cloud 1:1-NAT setups) or bind \ + explicitly to the public IP" + ); + } + None => {} + } } } "tor" => { @@ -1577,7 +1641,7 @@ impl Node { &self, bootstrap: &std::sync::Arc, ) -> Result<(), crate::discovery::nostr::BootstrapError> { - let advert = self.build_overlay_advert(); + let advert = self.build_overlay_advert(bootstrap).await; bootstrap.update_local_advert(advert).await }