mirror of
https://github.com/jmcorgan/fips.git
synced 2026-08-10 08:37:02 +00:00
Merge branch 'master' into next
This commit is contained in:
@@ -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:<eip>:<port>`).
|
||||
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>.onion:<port>`) instead of
|
||||
just the bare onion hostname. The publisher previously emitted a
|
||||
|
||||
@@ -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.*`).
|
||||
|
||||
@@ -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<SocketAddr> {
|
||||
if let Ok(sa) = raw.parse::<SocketAddr>() {
|
||||
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<u16> {
|
||||
raw.parse::<SocketAddr>().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<bool>,
|
||||
/// 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<String>,
|
||||
/// 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<SocketAddr> {
|
||||
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<bool>,
|
||||
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
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<SocketAddr> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String, _> = 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<Value> = 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!({
|
||||
|
||||
@@ -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<u64>,
|
||||
/// When the extended cooldown was applied; while
|
||||
/// `cooldown_until_ms.is_some_and(|t| t > now)`, retries are
|
||||
/// suppressed.
|
||||
pub cooldown_until_ms: Option<u64>,
|
||||
/// 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<i64>,
|
||||
}
|
||||
|
||||
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<u64>,
|
||||
/// 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<HashMap<String, NpubFailureRecord>>,
|
||||
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<u64> {
|
||||
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<String, NpubFailureRecord>, 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"));
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
@@ -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<SocketAddr>,
|
||||
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<mpsc::UnboundedReceiver<BootstrapEvent>>,
|
||||
notify_task: Mutex<Option<JoinHandle<()>>>,
|
||||
advertise_task: Mutex<Option<JoinHandle<()>>>,
|
||||
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<HashMap<u32, CachedPublicUdpAddr>>,
|
||||
}
|
||||
|
||||
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<u64> {
|
||||
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<NostrPeerFailureView> {
|
||||
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<SocketAddr> {
|
||||
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 `<reflexive_ip>:<advertise_port>`,
|
||||
/// 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<SocketAddr> {
|
||||
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<BootstrapEvent> {
|
||||
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()),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<T> {
|
||||
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<FreshnessOutcome, BootstrapError> {
|
||||
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<String>,
|
||||
punch: Option<PunchHint>,
|
||||
reason: Option<String>,
|
||||
offer_received_at: Option<u64>,
|
||||
) -> 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<FreshnessOutcome, BootstrapError> {
|
||||
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<i64> {
|
||||
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<FreshnessOutcome> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<TraversalAddress>,
|
||||
@@ -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<Option<SocketAddr>, 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 {
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -164,6 +164,16 @@ pub struct TraversalAnswer {
|
||||
pub stun_server: Option<String>,
|
||||
pub punch: Option<PunchHint>,
|
||||
pub reason: Option<String>,
|
||||
/// 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<u64>,
|
||||
}
|
||||
|
||||
#[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<u64>,
|
||||
/// 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<u64>,
|
||||
pub last_observed_skew_ms: Option<i64>,
|
||||
}
|
||||
|
||||
/// Outcome of `NostrDiscovery::refetch_advert_for_stale_check` (B6).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum NostrRefetchOutcome {
|
||||
Evicted,
|
||||
Refreshed,
|
||||
SameAdvert,
|
||||
Skipped,
|
||||
}
|
||||
|
||||
+142
-13
@@ -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<OverlayAdvert> {
|
||||
async fn build_overlay_advert(
|
||||
&self,
|
||||
bootstrap: &std::sync::Arc<NostrDiscovery>,
|
||||
) -> Option<OverlayAdvert> {
|
||||
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<NostrDiscovery>,
|
||||
) -> 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
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Item = &NodeAddr> {
|
||||
self.peers.keys()
|
||||
|
||||
Reference in New Issue
Block a user