diff --git a/CHANGELOG.md b/CHANGELOG.md index f0b09fd..63d1caa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -350,6 +350,72 @@ with v0.2.x peers. ### 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 + 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 6e3b0c9..dee7d79 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/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/control/queries.rs b/src/control/queries.rs index 092a34e..99b1ba7 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..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; @@ -14,16 +15,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; @@ -53,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, @@ -70,6 +93,11 @@ pub struct NostrDiscovery { event_rx: Mutex>, 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 { @@ -104,6 +132,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 +156,8 @@ impl NostrDiscovery { event_rx: Mutex::new(event_rx), notify_task: Mutex::new(None), advertise_task: Mutex::new(None), + failure_state, + public_udp_addr_cache: RwLock::new(HashMap::new()), }); runtime.subscribe().await?; @@ -154,6 +191,223 @@ 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() + } + + /// 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 + /// 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; @@ -536,6 +790,7 @@ impl NostrDiscovery { &base_socket, &self.config.stun_servers, self.config.share_local_candidates, + super::stun::TRAVERSAL_STUN_TIMEOUT, ) .await?; debug!( @@ -591,6 +846,7 @@ impl NostrDiscovery { } }; + let answer_received_at = now_ms(); debug!( peer = %peer_short, session = %short_id(&offer.session_id), @@ -599,14 +855,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 +932,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 +948,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 +956,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))?; @@ -678,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(); @@ -703,6 +1006,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 +1422,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 +1445,8 @@ impl NostrDiscovery { event_rx: Mutex::new(event_rx), 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/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/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/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 2192919..61bd9ca 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -448,7 +448,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, @@ -462,7 +511,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); + } } } } @@ -1247,6 +1305,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 { @@ -1285,6 +1344,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) @@ -1352,7 +1415,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!( @@ -1367,6 +1431,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" ); @@ -1465,7 +1530,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; } @@ -1487,13 +1555,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 { @@ -1510,13 +1614,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" => { @@ -1555,7 +1684,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 } diff --git a/src/node/mod.rs b/src/node/mod.rs index 1873cd4..bcf9ec0 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1567,6 +1567,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()