Merge branch 'master' into next

# Conflicts:
#	src/node/lifecycle.rs
#	testing/static/scripts/rekey-test.sh
This commit is contained in:
Johnathan Corgan
2026-04-27 16:10:11 +00:00
74 changed files with 7646 additions and 328 deletions
+19
View File
@@ -3,21 +3,33 @@
//! Allows unmodified LAN hosts to reach FIPS mesh destinations via
//! DNS-allocated virtual IPs and kernel nftables NAT.
#[cfg(target_os = "linux")]
use clap::Parser;
#[cfg(target_os = "linux")]
use fips::Config;
#[cfg(target_os = "linux")]
use fips::gateway::{control, dns, nat, net, pool};
#[cfg(target_os = "linux")]
use fips::version;
#[cfg(target_os = "linux")]
use std::path::PathBuf;
#[cfg(target_os = "linux")]
use std::sync::Arc;
#[cfg(target_os = "linux")]
use std::sync::atomic::{AtomicUsize, Ordering};
#[cfg(target_os = "linux")]
use std::time::Instant;
#[cfg(target_os = "linux")]
use tokio::signal::unix::{SignalKind, signal};
#[cfg(target_os = "linux")]
use tokio::sync::{Mutex, mpsc, watch};
#[cfg(target_os = "linux")]
use tracing::{error, info, warn};
#[cfg(target_os = "linux")]
use tracing_subscriber::{EnvFilter, fmt};
/// FIPS outbound LAN gateway
#[cfg(target_os = "linux")]
#[derive(Parser, Debug)]
#[command(
name = "fips-gateway",
@@ -35,6 +47,13 @@ struct Args {
log_level: String,
}
#[cfg(not(target_os = "linux"))]
fn main() {
eprintln!("fips-gateway requires Linux (nftables unavailable on this platform)");
std::process::exit(1);
}
#[cfg(target_os = "linux")]
#[tokio::main(flavor = "current_thread")]
async fn main() {
let args = Args::parse();
+22 -2
View File
@@ -73,11 +73,31 @@ async fn run_daemon(
}
};
// Initialize logging: RUST_LOG env var overrides config if set
// Initialize logging: RUST_LOG env var overrides config if set.
//
// The nostr-sdk relay pool emits the full JSON of every event it
// sends and receives at DEBUG level. At our DEBUG level that drowns
// out everything else, so suppress it unless the operator has
// explicitly asked for TRACE — at which point the raw frames come
// back.
let log_level = config.node.log_level();
let nostr_directive = if log_level == tracing::Level::TRACE {
"trace"
} else {
"info"
};
let default_directive = format!(
"{log_level},nostr_relay_pool={nostr_directive},nostr_sdk={nostr_directive},nostr={nostr_directive}"
);
let filter = EnvFilter::builder()
.with_default_directive(log_level.into())
.from_env_lossy();
.parse_lossy(default_directive);
let filter = match std::env::var("RUST_LOG") {
Ok(env) if !env.is_empty() => EnvFilter::builder()
.with_default_directive(log_level.into())
.parse_lossy(env),
_ => filter,
};
fmt().with_env_filter(filter).with_target(true).init();
+194 -6
View File
@@ -18,7 +18,7 @@
//! nsec: "nsec1..."
//! ```
#[cfg(feature = "gateway")]
#[cfg(target_os = "linux")]
mod gateway;
mod node;
mod peer;
@@ -30,12 +30,12 @@ use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use thiserror::Error;
#[cfg(feature = "gateway")]
#[cfg(target_os = "linux")]
pub use gateway::{ConntrackConfig, GatewayConfig, GatewayDnsConfig, PortForward, Proto};
pub use node::{
BloomConfig, BuffersConfig, CacheConfig, ControlConfig, DiscoveryConfig, LimitsConfig,
NodeConfig, RateLimitConfig, RekeyConfig, RetryConfig, SessionConfig, SessionMmpConfig,
TreeConfig,
NodeConfig, NostrDiscoveryConfig, NostrDiscoveryPolicy, RateLimitConfig, RekeyConfig,
RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
};
pub use peer::{ConnectPolicy, PeerAddress, PeerConfig};
pub use transport::{
@@ -337,6 +337,9 @@ pub enum ConfigError {
#[error("identity error: {0}")]
Identity(#[from] IdentityError),
#[error("invalid configuration: {0}")]
Validation(String),
}
/// Identity configuration (`node.identity.*`).
@@ -378,7 +381,7 @@ pub struct Config {
pub peers: Vec<PeerConfig>,
/// Gateway configuration (`gateway`).
#[cfg(feature = "gateway")]
#[cfg(target_os = "linux")]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub gateway: Option<GatewayConfig>,
}
@@ -500,7 +503,7 @@ impl Config {
self.peers = other.peers;
}
// Merge gateway section — higher-priority config replaces entirely
#[cfg(feature = "gateway")]
#[cfg(target_os = "linux")]
if other.gateway.is_some() {
self.gateway = other.gateway;
}
@@ -552,6 +555,69 @@ impl Config {
self.peers.iter().filter(|p| p.is_auto_connect())
}
/// Validate cross-field configuration invariants.
pub fn validate(&self) -> Result<(), ConfigError> {
let nostr = &self.node.discovery.nostr;
let any_transport_advertises_on_nostr = self
.transports
.udp
.iter()
.any(|(_, cfg)| cfg.advertise_on_nostr())
|| self
.transports
.tcp
.iter()
.any(|(_, cfg)| cfg.advertise_on_nostr())
|| self
.transports
.tor
.iter()
.any(|(_, cfg)| cfg.advertise_on_nostr());
if any_transport_advertises_on_nostr && !nostr.enabled {
return Err(ConfigError::Validation(
"at least one transport has `advertise_on_nostr = true`, but `node.discovery.nostr.enabled` is false".to_string(),
));
}
if self.peers.iter().any(|peer| peer.via_nostr) && !nostr.enabled {
return Err(ConfigError::Validation(
"at least one peer has `via_nostr = true`, but `node.discovery.nostr.enabled` is false".to_string(),
));
}
for (i, peer) in self.peers.iter().enumerate() {
if peer.addresses.is_empty() && !peer.via_nostr {
return Err(ConfigError::Validation(format!(
"peers[{i}] ({}): must specify at least one address, or set `via_nostr = true` to resolve endpoints from the Nostr advert",
peer.npub
)));
}
}
let has_nat_udp_advert = self
.transports
.udp
.iter()
.any(|(_, cfg)| cfg.advertise_on_nostr() && !cfg.is_public());
if nostr.enabled && has_nat_udp_advert {
if nostr.dm_relays.is_empty() {
return Err(ConfigError::Validation(
"NAT UDP advert publishing requires `node.discovery.nostr.dm_relays` to be non-empty".to_string(),
));
}
if nostr.stun_servers.is_empty() {
return Err(ConfigError::Validation(
"NAT UDP advert publishing requires `node.discovery.nostr.stun_servers` to be non-empty".to_string(),
));
}
}
Ok(())
}
/// Serialize this configuration to YAML.
pub fn to_yaml(&self) -> Result<String, serde_yaml::Error> {
serde_yaml::to_string(self)
@@ -1122,4 +1188,126 @@ peers:
assert_eq!(peer.addresses.len(), 2);
assert!(peer.is_auto_connect());
}
#[test]
fn test_parse_nostr_discovery_config() {
let yaml = r#"
node:
discovery:
nostr:
enabled: true
advertise: false
policy: configured_only
open_discovery_max_pending: 12
app: "fips.nat.test.v1"
signal_ttl_secs: 45
advert_relays:
- "wss://relay-a.example"
dm_relays:
- "wss://relay-b.example"
stun_servers:
- "stun:stun.example.org:3478"
peers:
- npub: "npub1peer"
via_nostr: true
addresses:
- transport: udp
addr: "nat"
"#;
let config: Config = serde_yaml::from_str(yaml).unwrap();
assert!(config.node.discovery.nostr.enabled);
assert!(!config.node.discovery.nostr.advertise);
assert_eq!(config.node.discovery.nostr.app, "fips.nat.test.v1");
assert_eq!(config.node.discovery.nostr.signal_ttl_secs, 45);
assert_eq!(
config.node.discovery.nostr.policy,
NostrDiscoveryPolicy::ConfiguredOnly
);
assert_eq!(config.node.discovery.nostr.open_discovery_max_pending, 12);
assert_eq!(
config.node.discovery.nostr.advert_relays,
vec!["wss://relay-a.example".to_string()]
);
assert_eq!(
config.node.discovery.nostr.dm_relays,
vec!["wss://relay-b.example".to_string()]
);
assert_eq!(
config.node.discovery.nostr.stun_servers,
vec!["stun:stun.example.org:3478".to_string()]
);
assert_eq!(
config.peers[0].addresses[0].addr, "nat",
"udp:nat address should parse without special-casing in YAML"
);
assert!(config.peers[0].via_nostr);
}
#[test]
fn test_validate_transport_advert_requires_nostr_enabled() {
let mut config = Config::default();
config.transports.udp = TransportInstances::Single(UdpConfig {
advertise_on_nostr: Some(true),
..Default::default()
});
config.node.discovery.nostr.enabled = false;
let err = config.validate().expect_err("validation should fail");
assert!(err.to_string().contains("advertise_on_nostr"));
}
#[test]
#[allow(clippy::field_reassign_with_default)]
fn test_validate_peer_via_nostr_requires_nostr_enabled() {
let mut config = Config::default();
config.peers = vec![PeerConfig {
npub: "npub1peer".to_string(),
via_nostr: true,
..Default::default()
}];
config.node.discovery.nostr.enabled = false;
let err = config.validate().expect_err("validation should fail");
assert!(err.to_string().contains("via_nostr"));
}
#[test]
#[allow(clippy::field_reassign_with_default)]
fn test_validate_peer_addresses_required_unless_via_nostr() {
// Empty addresses + via_nostr=false → error.
let mut config = Config::default();
config.peers = vec![PeerConfig {
npub: "npub1peer".to_string(),
..Default::default()
}];
let err = config.validate().expect_err("validation should fail");
assert!(err.to_string().contains("at least one address"));
// Empty addresses + via_nostr=true + nostr.enabled=true → ok.
config.peers[0].via_nostr = true;
config.node.discovery.nostr.enabled = true;
config
.validate()
.expect("via_nostr should allow empty addresses");
}
#[test]
fn test_validate_nat_udp_advert_requires_relays_and_stun() {
let mut config = Config::default();
config.node.discovery.nostr.enabled = true;
config.node.discovery.nostr.dm_relays.clear();
config.transports.udp = TransportInstances::Single(UdpConfig {
advertise_on_nostr: Some(true),
public: Some(false),
..Default::default()
});
let err = config.validate().expect_err("validation should fail");
assert!(err.to_string().contains("dm_relays"));
config.node.discovery.nostr.dm_relays = vec!["wss://relay.example".to_string()];
config.node.discovery.nostr.stun_servers.clear();
let err = config.validate().expect_err("validation should fail");
assert!(err.to_string().contains("stun_servers"));
}
}
+228 -27
View File
@@ -192,14 +192,20 @@ pub struct DiscoveryConfig {
/// Hop limit for LookupRequest flood (`node.discovery.ttl`).
#[serde(default = "DiscoveryConfig::default_ttl")]
pub ttl: u8,
/// Lookup completion timeout in seconds (`node.discovery.timeout_secs`).
#[serde(default = "DiscoveryConfig::default_timeout_secs")]
pub timeout_secs: u64,
/// Per-attempt timeouts in seconds (`node.discovery.attempt_timeouts_secs`).
/// Each entry is the time to wait for a response before sending the next
/// LookupRequest (with a fresh request_id). Sequence length determines the
/// total number of attempts before declaring the destination unreachable.
/// Default `[1, 2, 4, 8]` gives 4 attempts and a 15s total budget.
#[serde(default = "DiscoveryConfig::default_attempt_timeouts_secs")]
pub attempt_timeouts_secs: Vec<u64>,
/// Dedup cache expiry in seconds (`node.discovery.recent_expiry_secs`).
#[serde(default = "DiscoveryConfig::default_recent_expiry_secs")]
pub recent_expiry_secs: u64,
/// Base backoff after first lookup failure in seconds (`node.discovery.backoff_base_secs`).
/// Doubles per consecutive failure up to `backoff_max_secs`.
/// Base backoff after lookup failure in seconds (`node.discovery.backoff_base_secs`).
/// Doubles per consecutive failure up to `backoff_max_secs`. Defaults to 0
/// (no post-failure suppression); the per-attempt sequence in
/// `attempt_timeouts_secs` provides the only retry pacing.
#[serde(default = "DiscoveryConfig::default_backoff_base_secs")]
pub backoff_base_secs: u64,
/// Maximum backoff cap in seconds (`node.discovery.backoff_max_secs`).
@@ -210,28 +216,21 @@ pub struct DiscoveryConfig {
/// Defense-in-depth against misbehaving nodes.
#[serde(default = "DiscoveryConfig::default_forward_min_interval_secs")]
pub forward_min_interval_secs: u64,
/// Retry interval within the timeout window in seconds
/// (`node.discovery.retry_interval_secs`).
/// After this interval without a response, resend the lookup.
#[serde(default = "DiscoveryConfig::default_retry_interval_secs")]
pub retry_interval_secs: u64,
/// Maximum attempts per lookup (`node.discovery.max_attempts`).
/// 1 = no retry, 2 = one retry, etc.
#[serde(default = "DiscoveryConfig::default_max_attempts")]
pub max_attempts: u8,
/// Nostr-mediated overlay endpoint discovery.
#[serde(default = "DiscoveryConfig::default_nostr")]
pub nostr: NostrDiscoveryConfig,
}
impl Default for DiscoveryConfig {
fn default() -> Self {
Self {
ttl: 64,
timeout_secs: 10,
attempt_timeouts_secs: vec![1, 2, 4, 8],
recent_expiry_secs: 10,
backoff_base_secs: 30,
backoff_max_secs: 300,
backoff_base_secs: 0,
backoff_max_secs: 0,
forward_min_interval_secs: 2,
retry_interval_secs: 5,
max_attempts: 2,
nostr: NostrDiscoveryConfig::default(),
}
}
}
@@ -240,26 +239,228 @@ impl DiscoveryConfig {
fn default_ttl() -> u8 {
64
}
fn default_timeout_secs() -> u64 {
10
fn default_attempt_timeouts_secs() -> Vec<u64> {
vec![1, 2, 4, 8]
}
fn default_recent_expiry_secs() -> u64 {
10
}
fn default_backoff_base_secs() -> u64 {
30
0
}
fn default_backoff_max_secs() -> u64 {
300
0
}
fn default_forward_min_interval_secs() -> u64 {
2
}
fn default_retry_interval_secs() -> u64 {
5
fn default_nostr() -> NostrDiscoveryConfig {
NostrDiscoveryConfig::default()
}
fn default_max_attempts() -> u8 {
2
}
/// Nostr advert discovery policy.
///
/// Controls how overlay endpoint adverts are consumed:
/// - `disabled`: ignore advert-derived endpoints for all peers
/// - `configured_only`: allow advert fallback only for configured peers with
/// `peers[].via_nostr = true`
/// - `open`: also consider adverts for non-configured peers
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NostrDiscoveryPolicy {
Disabled,
#[default]
ConfiguredOnly,
Open,
}
/// Nostr-mediated overlay endpoint discovery (`node.discovery.nostr.*`).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NostrDiscoveryConfig {
/// Enable Nostr-signaled traversal bootstrap.
#[serde(default)]
pub enabled: bool,
/// Publish service advertisements so remote peers can bootstrap inbound.
#[serde(default = "NostrDiscoveryConfig::default_advertise")]
pub advertise: bool,
/// Relay URLs used for service advertisements.
#[serde(default = "NostrDiscoveryConfig::default_advert_relays")]
pub advert_relays: Vec<String>,
/// Relay URLs used for encrypted signaling events.
#[serde(default = "NostrDiscoveryConfig::default_dm_relays")]
pub dm_relays: Vec<String>,
/// STUN servers used for local reflexive address discovery.
/// Outbound observation uses only this local list; peer-advertised STUN
/// values are informational and are not treated as egress targets.
#[serde(default = "NostrDiscoveryConfig::default_stun_servers")]
pub stun_servers: Vec<String>,
/// Whether to advertise local (RFC 1918 / ULA) interface addresses as
/// host candidates in the traversal offer.
///
/// Off by default: in most deployments the relevant peers are not on the
/// same broadcast domain, and sharing private host candidates causes
/// misleading punch successes when an asymmetric L3 path (corporate VPN,
/// Tailscale subnet route, overlapping address space, etc.) makes a
/// peer's private IP one-way reachable from this node. Enable only when
/// peers are on the same physical LAN and same-LAN punching is wanted.
#[serde(default)]
pub share_local_candidates: bool,
/// Traversal application namespace and advert identifier suffix.
#[serde(default = "NostrDiscoveryConfig::default_app")]
pub app: String,
/// Signaling TTL in seconds.
#[serde(default = "NostrDiscoveryConfig::default_signal_ttl_secs")]
pub signal_ttl_secs: u64,
/// Policy for advert-derived endpoint discovery.
#[serde(default)]
pub policy: NostrDiscoveryPolicy,
/// Max number of open-discovery peers queued for outbound retry/connection
/// at once. Prevents unbounded queue growth from ambient advert traffic.
#[serde(default = "NostrDiscoveryConfig::default_open_discovery_max_pending")]
pub open_discovery_max_pending: usize,
/// Max concurrent inbound traversal offers processed at once.
/// Acts as a rate limit against offer spam from relays.
#[serde(default = "NostrDiscoveryConfig::default_max_concurrent_incoming_offers")]
pub max_concurrent_incoming_offers: usize,
/// Max cached overlay adverts retained from relay traffic.
/// Bounds memory under ambient advert volume.
#[serde(default = "NostrDiscoveryConfig::default_advert_cache_max_entries")]
pub advert_cache_max_entries: usize,
/// Max seen-session IDs retained for replay detection.
/// Oldest entries are evicted when the cap is exceeded.
#[serde(default = "NostrDiscoveryConfig::default_seen_sessions_max_entries")]
pub seen_sessions_max_entries: usize,
/// Overall punch attempt timeout in seconds.
#[serde(default = "NostrDiscoveryConfig::default_attempt_timeout_secs")]
pub attempt_timeout_secs: u64,
/// Replay tracking retention window in seconds.
#[serde(default = "NostrDiscoveryConfig::default_replay_window_secs")]
pub replay_window_secs: u64,
/// Delay before punch traffic starts.
#[serde(default = "NostrDiscoveryConfig::default_punch_start_delay_ms")]
pub punch_start_delay_ms: u64,
/// Interval between punch packets.
#[serde(default = "NostrDiscoveryConfig::default_punch_interval_ms")]
pub punch_interval_ms: u64,
/// How long to keep punching before failure.
#[serde(default = "NostrDiscoveryConfig::default_punch_duration_ms")]
pub punch_duration_ms: u64,
/// Advert TTL in seconds.
#[serde(default = "NostrDiscoveryConfig::default_advert_ttl_secs")]
pub advert_ttl_secs: u64,
/// How often adverts are refreshed in seconds.
#[serde(default = "NostrDiscoveryConfig::default_advert_refresh_secs")]
pub advert_refresh_secs: u64,
}
impl Default for NostrDiscoveryConfig {
fn default() -> Self {
Self {
enabled: false,
advertise: Self::default_advertise(),
advert_relays: Self::default_advert_relays(),
dm_relays: Self::default_dm_relays(),
stun_servers: Self::default_stun_servers(),
share_local_candidates: false,
app: Self::default_app(),
signal_ttl_secs: Self::default_signal_ttl_secs(),
policy: NostrDiscoveryPolicy::default(),
open_discovery_max_pending: Self::default_open_discovery_max_pending(),
max_concurrent_incoming_offers: Self::default_max_concurrent_incoming_offers(),
advert_cache_max_entries: Self::default_advert_cache_max_entries(),
seen_sessions_max_entries: Self::default_seen_sessions_max_entries(),
attempt_timeout_secs: Self::default_attempt_timeout_secs(),
replay_window_secs: Self::default_replay_window_secs(),
punch_start_delay_ms: Self::default_punch_start_delay_ms(),
punch_interval_ms: Self::default_punch_interval_ms(),
punch_duration_ms: Self::default_punch_duration_ms(),
advert_ttl_secs: Self::default_advert_ttl_secs(),
advert_refresh_secs: Self::default_advert_refresh_secs(),
}
}
}
impl NostrDiscoveryConfig {
fn default_advertise() -> bool {
true
}
fn default_advert_relays() -> Vec<String> {
vec![
"wss://relay.damus.io".to_string(),
"wss://nos.lol".to_string(),
"wss://offchain.pub".to_string(),
]
}
fn default_dm_relays() -> Vec<String> {
vec![
"wss://relay.damus.io".to_string(),
"wss://nos.lol".to_string(),
"wss://offchain.pub".to_string(),
]
}
fn default_stun_servers() -> Vec<String> {
vec![
"stun:stun.l.google.com:19302".to_string(),
"stun:stun.cloudflare.com:3478".to_string(),
"stun:global.stun.twilio.com:3478".to_string(),
]
}
fn default_app() -> String {
"fips-overlay-v1".to_string()
}
fn default_signal_ttl_secs() -> u64 {
120
}
fn default_open_discovery_max_pending() -> usize {
64
}
fn default_max_concurrent_incoming_offers() -> usize {
16
}
fn default_advert_cache_max_entries() -> usize {
2048
}
fn default_seen_sessions_max_entries() -> usize {
2048
}
fn default_attempt_timeout_secs() -> u64 {
10
}
fn default_replay_window_secs() -> u64 {
300
}
fn default_punch_start_delay_ms() -> u64 {
2_000
}
fn default_punch_interval_ms() -> u64 {
200
}
fn default_punch_duration_ms() -> u64 {
10_000
}
fn default_advert_ttl_secs() -> u64 {
3_600
}
fn default_advert_refresh_secs() -> u64 {
1_800
}
}
+14 -1
View File
@@ -94,7 +94,11 @@ pub struct PeerConfig {
pub alias: Option<String>,
/// Transport addresses for reaching this peer.
/// At least one address is required.
///
/// At least one address is required unless `via_nostr` is `true`,
/// in which case the address list may be empty and endpoints are
/// resolved from the peer's Nostr advert at dial time.
#[serde(default)]
pub addresses: Vec<PeerAddress>,
/// Connection policy for this peer.
@@ -106,6 +110,13 @@ pub struct PeerConfig {
/// backoff after MMP removes this peer due to liveness timeout.
#[serde(default = "default_auto_reconnect")]
pub auto_reconnect: bool,
/// Whether to append Nostr-advertised endpoints when dialing this peer.
///
/// Static addresses are still attempted first; advert-derived endpoints are
/// appended as fallback candidates.
#[serde(default)]
pub via_nostr: bool,
}
impl Default for PeerConfig {
@@ -116,6 +127,7 @@ impl Default for PeerConfig {
addresses: Vec::new(),
connect_policy: ConnectPolicy::default(),
auto_reconnect: default_auto_reconnect(),
via_nostr: false,
}
}
}
@@ -133,6 +145,7 @@ impl PeerConfig {
addresses: vec![PeerAddress::new(transport, addr)],
connect_policy: ConnectPolicy::default(),
auto_reconnect: default_auto_reconnect(),
via_nostr: false,
}
}
+43
View File
@@ -38,6 +38,19 @@ pub struct UdpConfig {
/// UDP send buffer size in bytes (`send_buf_size`). Defaults to 2 MB.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub send_buf_size: Option<usize>,
/// Whether this transport should be advertised on Nostr overlay discovery.
/// Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub advertise_on_nostr: Option<bool>,
/// Whether UDP should be advertised as directly reachable (`host:port`) on
/// Nostr. When false and advertised, UDP is emitted as `addr: "nat"` to
/// trigger rendezvous traversal.
///
/// Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub public: Option<bool>,
}
impl UdpConfig {
@@ -60,6 +73,16 @@ impl UdpConfig {
pub fn send_buf_size(&self) -> usize {
self.send_buf_size.unwrap_or(DEFAULT_UDP_SEND_BUF)
}
/// Whether this UDP transport should be advertised on Nostr discovery.
pub fn advertise_on_nostr(&self) -> bool {
self.advertise_on_nostr.unwrap_or(false)
}
/// Whether this UDP transport should be advertised as directly reachable.
pub fn is_public(&self) -> bool {
self.public.unwrap_or(false)
}
}
/// Transport instances - either a single config or named instances.
@@ -293,6 +316,11 @@ pub struct TcpConfig {
/// Maximum simultaneous inbound connections. Defaults to 256.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_inbound_connections: Option<usize>,
/// Whether this transport should be advertised on Nostr overlay discovery.
/// Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub advertise_on_nostr: Option<bool>,
}
impl TcpConfig {
@@ -332,6 +360,11 @@ impl TcpConfig {
self.max_inbound_connections
.unwrap_or(DEFAULT_TCP_MAX_INBOUND)
}
/// Whether this TCP transport should be advertised on Nostr discovery.
pub fn advertise_on_nostr(&self) -> bool {
self.advertise_on_nostr.unwrap_or(false)
}
}
// ============================================================================
@@ -420,6 +453,11 @@ pub struct TorConfig {
/// in torrc; fips reads the .onion hostname from a file.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub directory_service: Option<DirectoryServiceConfig>,
/// Whether this transport should be advertised on Nostr overlay discovery.
/// Default: false.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub advertise_on_nostr: Option<bool>,
}
/// Directory-mode onion service configuration.
@@ -507,6 +545,11 @@ impl TorConfig {
self.max_inbound_connections
.unwrap_or(DEFAULT_TOR_MAX_INBOUND)
}
/// Whether this Tor transport should be advertised on Nostr discovery.
pub fn advertise_on_nostr(&self) -> bool {
self.advertise_on_nostr.unwrap_or(false)
}
}
// ============================================================================
+80
View File
@@ -0,0 +1,80 @@
//! Bootstrap handoff types.
//!
//! These types model the boundary between an external rendezvous/bootstrap
//! runtime and the core FIPS transport/handshake stack. The rendezvous side
//! owns Nostr/STUN/UDP hole punching; once a direct UDP path is established,
//! it hands the live socket and selected remote endpoint to FIPS so the
//! existing Noise/FMP transport path can take over.
#[cfg(feature = "nostr-discovery")]
pub mod nostr;
use crate::config::UdpConfig;
use crate::{NodeAddr, TransportId};
use std::net::{SocketAddr, UdpSocket};
/// Result of handing an established traversal session into FIPS.
#[derive(Debug, Clone)]
pub struct BootstrapHandoffResult {
/// Newly allocated transport ID used for the adopted UDP socket.
pub transport_id: TransportId,
/// Local socket address now owned by the FIPS UDP transport.
pub local_addr: SocketAddr,
/// Confirmed remote UDP endpoint selected by traversal.
pub remote_addr: SocketAddr,
/// Peer node address derived from the supplied peer identity.
pub peer_node_addr: NodeAddr,
/// Nostr session identifier used by the bootstrap runtime.
pub session_id: String,
}
/// Established UDP traversal ready to be handed into FIPS.
///
/// The socket must already be bound and must be the same socket used for the
/// traversal runtime's STUN and punch traffic so the NAT mapping is preserved.
#[derive(Debug)]
pub struct EstablishedTraversal {
/// Rendezvous session identifier for logging/correlation.
pub session_id: String,
/// Remote peer identity in `npub` form.
pub peer_npub: String,
/// The selected remote UDP endpoint to use for the FIPS handshake.
pub remote_addr: SocketAddr,
/// The live UDP socket carrying the established mapping.
pub socket: UdpSocket,
/// Optional name for the adopted UDP transport.
pub transport_name: Option<String>,
/// Optional UDP transport tuning overrides.
pub transport_config: Option<UdpConfig>,
}
impl EstablishedTraversal {
/// Construct an established traversal handoff.
pub fn new(
session_id: impl Into<String>,
peer_npub: impl Into<String>,
remote_addr: SocketAddr,
socket: UdpSocket,
) -> Self {
Self {
session_id: session_id.into(),
peer_npub: peer_npub.into(),
remote_addr,
socket,
transport_name: None,
transport_config: None,
}
}
/// Attach an explicit transport name to the adopted UDP transport.
pub fn with_transport_name(mut self, name: impl Into<String>) -> Self {
self.transport_name = Some(name.into());
self
}
/// Override UDP transport tuning for the adopted socket.
pub fn with_transport_config(mut self, config: UdpConfig) -> Self {
self.transport_config = Some(config);
self
}
}
+18
View File
@@ -0,0 +1,18 @@
#![cfg(feature = "nostr-discovery")]
mod runtime;
mod signal;
mod stun;
mod traversal;
mod types;
#[cfg(test)]
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,
};
File diff suppressed because it is too large Load Diff
+193
View File
@@ -0,0 +1,193 @@
use nostr::EventId;
use nostr::nips::{nip44, nip59};
use nostr::prelude::{
Event, EventBuilder, JsonUtil, Kind, NostrSigner, PublicKey, Tag, Timestamp, UnsignedEvent,
};
use super::types::{BootstrapError, PunchHint, SIGNAL_KIND, TraversalAnswer, TraversalOffer};
pub(super) struct SignalEnvelope<T> {
pub(super) payload: T,
pub(super) event_id: EventId,
pub(super) sender_npub: String,
}
pub(super) struct UnwrappedSignal {
pub(super) sender: PublicKey,
pub(super) rumor: UnsignedEvent,
}
pub(super) async fn build_signal_event(
signer: &nostr::Keys,
receiver: PublicKey,
rumor: UnsignedEvent,
expiration: Timestamp,
) -> Result<Event, BootstrapError> {
let seal = nip59::make_seal(signer, &receiver, rumor)
.await
.map_err(|e| BootstrapError::Nostr(e.to_string()))?
.sign(signer)
.await
.map_err(|e| BootstrapError::Nostr(e.to_string()))?;
let ephemeral = nostr::Keys::generate();
let content = nip44::encrypt(
ephemeral.secret_key(),
&receiver,
seal.as_json(),
nip44::Version::default(),
)
.map_err(|e| BootstrapError::Nostr(e.to_string()))?;
EventBuilder::new(Kind::Custom(SIGNAL_KIND), content)
.tags([Tag::public_key(receiver), Tag::expiration(expiration)])
.sign_with_keys(&ephemeral)
.map_err(|e| BootstrapError::Nostr(e.to_string()))
}
pub(super) async fn unwrap_signal_event(
signer: &nostr::Keys,
event: &Event,
) -> Result<UnwrappedSignal, BootstrapError> {
if event.kind != Kind::Custom(SIGNAL_KIND) {
return Err(BootstrapError::Protocol(
"not a traversal signal".to_string(),
));
}
let seal_json = signer
.nip44_decrypt(&event.pubkey, &event.content)
.await
.map_err(|e| BootstrapError::Nostr(e.to_string()))?;
let seal =
Event::from_json(seal_json).map_err(|e| BootstrapError::EventParse(e.to_string()))?;
seal.verify()
.map_err(|e| BootstrapError::Nostr(e.to_string()))?;
let rumor_json = signer
.nip44_decrypt(&seal.pubkey, &seal.content)
.await
.map_err(|e| BootstrapError::Nostr(e.to_string()))?;
let rumor = UnsignedEvent::from_json(rumor_json)
.map_err(|e| BootstrapError::EventParse(e.to_string()))?;
Ok(UnwrappedSignal {
sender: seal.pubkey,
rumor,
})
}
pub(super) fn validate_offer_freshness(
offer: &TraversalOffer,
now: u64,
signal_ttl_ms: u64,
actual_sender_npub: &str,
local_npub: &str,
) -> Result<(), 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()));
}
if offer.sender_npub != actual_sender_npub || offer.recipient_npub != local_npub {
return Err(BootstrapError::Protocol("identity-mismatch".to_string()));
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(super) fn create_traversal_offer(
session_id: String,
issued_at: u64,
ttl_ms: u64,
nonce: String,
sender_npub: String,
recipient_npub: String,
reflexive_address: Option<super::TraversalAddress>,
local_addresses: Vec<super::TraversalAddress>,
stun_server: Option<String>,
) -> TraversalOffer {
TraversalOffer {
message_type: "offer".to_string(),
session_id,
issued_at,
expires_at: issued_at + ttl_ms,
nonce,
sender_npub,
recipient_npub,
reflexive_address,
local_addresses,
stun_server,
}
}
#[allow(clippy::too_many_arguments)]
pub(super) fn create_traversal_answer(
session_id: String,
issued_at: u64,
ttl_ms: u64,
nonce: String,
sender_npub: String,
recipient_npub: String,
in_reply_to: String,
accepted: bool,
reflexive_address: Option<super::TraversalAddress>,
local_addresses: Vec<super::TraversalAddress>,
stun_server: Option<String>,
punch: Option<PunchHint>,
reason: Option<String>,
) -> TraversalAnswer {
TraversalAnswer {
message_type: "answer".to_string(),
session_id,
issued_at,
expires_at: issued_at + ttl_ms,
nonce,
sender_npub,
recipient_npub,
in_reply_to,
accepted,
reflexive_address,
local_addresses,
stun_server,
punch,
reason,
}
}
pub(super) fn validate_traversal_answer_for_offer(
offer: &TraversalOffer,
answer: &TraversalAnswer,
now: u64,
signal_ttl_ms: u64,
actual_sender_npub: &str,
local_npub: &str,
) -> Result<(), 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
{
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()));
}
if offer.sender_npub != local_npub
|| offer.recipient_npub != actual_sender_npub
|| answer.sender_npub != actual_sender_npub
|| answer.recipient_npub != local_npub
{
return Err(BootstrapError::Protocol("identity-mismatch".to_string()));
}
if answer.accepted && answer.reflexive_address.is_none() && answer.local_addresses.is_empty() {
return Err(BootstrapError::Protocol("missing-addresses".to_string()));
}
if !answer.accepted && answer.reason.as_deref().unwrap_or_default().is_empty() {
return Err(BootstrapError::Protocol(
"missing-rejection-reason".to_string(),
));
}
Ok(())
}
+391
View File
@@ -0,0 +1,391 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::time::Duration;
use tokio::net::{UdpSocket, lookup_host};
use tracing::debug;
use super::types::{BootstrapError, TraversalAddress};
// Current STUN parsing is intentionally minimal and only supports
// MAPPED-ADDRESS / XOR-MAPPED-ADDRESS for IPv4 and IPv6.
// Local interface discovery remains best-effort and may still be incomplete
// on dual-stack, NAT64, or heavily firewalled hosts.
pub(super) async fn observe_traversal_addresses(
socket: &std::net::UdpSocket,
stun_servers: &[String],
share_local_candidates: bool,
) -> Result<
(
Option<TraversalAddress>,
Vec<TraversalAddress>,
Option<String>,
),
BootstrapError,
> {
let local_port = socket.local_addr()?.port();
let local_addresses = if share_local_candidates {
local_addresses_from_port(local_port)
.into_iter()
.map(|ip| TraversalAddress {
protocol: "udp".to_string(),
ip,
port: local_port,
})
.collect::<Vec<_>>()
} else {
Vec::new()
};
let mut last_error = None;
for stun_server in stun_servers {
match perform_stun(socket, stun_server).await {
Ok(mapped) => {
debug!(
stun_server = %stun_server,
reflexive = ?mapped,
"STUN observation succeeded"
);
return Ok((
mapped.map(|addr| TraversalAddress {
protocol: "udp".to_string(),
ip: addr.ip().to_string(),
port: addr.port(),
}),
local_addresses.clone(),
Some(stun_server.clone()),
));
}
Err(err) => last_error = Some(err),
}
}
if let Some(err) = last_error {
debug!(error = %err, "stun observation failed, falling back to LAN-only addresses");
}
Ok((None, local_addresses, None))
}
async fn perform_stun(
socket: &std::net::UdpSocket,
stun_server: &str,
) -> Result<Option<SocketAddr>, BootstrapError> {
let endpoint = parse_stun_url(stun_server)?;
let txn_id = random_txn_id();
let request = create_stun_binding_request(txn_id);
let addr = resolve_udp_target(&endpoint.host, endpoint.port)
.await?
.ok_or_else(|| BootstrapError::Stun(format!("no address for {}", stun_server)))?;
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);
loop {
let result = tokio::time::timeout_at(deadline, udp.recv_from(&mut buf)).await;
let Ok(Ok((len, _remote))) = result else {
break;
};
if let Some(mapped) = parse_stun_binding_success(&buf[..len], &txn_id) {
return Ok(Some(mapped));
}
}
Err(BootstrapError::Stun(format!(
"timed out waiting for {}",
stun_server
)))
}
pub(super) fn parse_stun_url(input: &str) -> Result<StunEndpoint, BootstrapError> {
let raw = input.strip_prefix("stun:").unwrap_or(input);
let Some((host, port)) = raw.rsplit_once(':') else {
return Err(BootstrapError::Stun(format!("invalid STUN URL: {}", input)));
};
let port = port
.parse::<u16>()
.map_err(|_| BootstrapError::Stun(format!("invalid STUN URL: {}", input)))?;
if host.is_empty() {
return Err(BootstrapError::Stun(format!("invalid STUN URL: {}", input)));
}
Ok(StunEndpoint {
host: host.to_string(),
port,
})
}
pub(super) struct StunEndpoint {
pub(super) host: String,
pub(super) port: u16,
}
fn create_stun_binding_request(txn_id: [u8; 12]) -> [u8; 20] {
const STUN_BINDING_REQUEST: u16 = 0x0001;
const STUN_MAGIC_COOKIE: u32 = 0x2112_a442;
let mut packet = [0u8; 20];
packet[..2].copy_from_slice(&STUN_BINDING_REQUEST.to_be_bytes());
packet[2..4].copy_from_slice(&0u16.to_be_bytes());
packet[4..8].copy_from_slice(&STUN_MAGIC_COOKIE.to_be_bytes());
packet[8..20].copy_from_slice(&txn_id);
packet
}
pub(super) fn parse_stun_binding_success(packet: &[u8], txn_id: &[u8; 12]) -> Option<SocketAddr> {
const STUN_BINDING_SUCCESS: u16 = 0x0101;
const STUN_MAGIC_COOKIE: u32 = 0x2112_a442;
const STUN_ATTR_MAPPED_ADDRESS: u16 = 0x0001;
const STUN_ATTR_XOR_MAPPED_ADDRESS: u16 = 0x0020;
if packet.len() < 20 {
return None;
}
if u16::from_be_bytes(packet[..2].try_into().ok()?) != STUN_BINDING_SUCCESS {
return None;
}
if u32::from_be_bytes(packet[4..8].try_into().ok()?) != STUN_MAGIC_COOKIE {
return None;
}
if &packet[8..20] != txn_id {
return None;
}
let message_length = u16::from_be_bytes(packet[2..4].try_into().ok()?) as usize;
let mut offset = 20usize;
let max_offset = packet.len().min(20 + message_length);
while offset + 4 <= max_offset {
let attr_type = u16::from_be_bytes(packet[offset..offset + 2].try_into().ok()?);
let attr_len = u16::from_be_bytes(packet[offset + 2..offset + 4].try_into().ok()?) as usize;
let value_start = offset + 4;
let value_end = value_start + attr_len;
if value_end > packet.len() {
break;
}
let value = &packet[value_start..value_end];
let parsed = match attr_type {
STUN_ATTR_XOR_MAPPED_ADDRESS => parse_xor_mapped_address(value, txn_id),
STUN_ATTR_MAPPED_ADDRESS => parse_mapped_address(value),
_ => None,
};
if parsed.is_some() {
return parsed;
}
offset = value_end + ((4 - (attr_len % 4)) % 4);
}
None
}
fn parse_mapped_address(value: &[u8]) -> Option<SocketAddr> {
match value.get(1).copied()? {
0x01 if value.len() >= 8 => Some(SocketAddr::new(
IpAddr::V4(Ipv4Addr::new(value[4], value[5], value[6], value[7])),
u16::from_be_bytes([value[2], value[3]]),
)),
0x02 if value.len() >= 20 => {
let ip = Ipv6Addr::from(<[u8; 16]>::try_from(&value[4..20]).ok()?);
Some(SocketAddr::new(
IpAddr::V6(ip),
u16::from_be_bytes([value[2], value[3]]),
))
}
_ => None,
}
}
fn parse_xor_mapped_address(value: &[u8], txn_id: &[u8; 12]) -> Option<SocketAddr> {
const STUN_MAGIC_COOKIE: u32 = 0x2112_a442;
let xored_port = u16::from_be_bytes([value.get(2).copied()?, value.get(3).copied()?])
^ ((STUN_MAGIC_COOKIE >> 16) as u16);
let cookie = STUN_MAGIC_COOKIE.to_be_bytes();
match value.get(1).copied()? {
0x01 if value.len() >= 8 => {
let ip = Ipv4Addr::new(
value[4] ^ cookie[0],
value[5] ^ cookie[1],
value[6] ^ cookie[2],
value[7] ^ cookie[3],
);
Some(SocketAddr::new(IpAddr::V4(ip), xored_port))
}
0x02 if value.len() >= 20 => {
let mut ip = [0u8; 16];
for (index, byte) in ip.iter_mut().enumerate() {
let mask = if index < 4 {
cookie[index]
} else {
txn_id[index - 4]
};
*byte = value[4 + index] ^ mask;
}
Some(SocketAddr::new(IpAddr::V6(Ipv6Addr::from(ip)), xored_port))
}
_ => None,
}
}
async fn resolve_udp_target(host: &str, port: u16) -> Result<Option<SocketAddr>, BootstrapError> {
let normalized_host = host
.strip_prefix('[')
.and_then(|trimmed| trimmed.strip_suffix(']'))
.unwrap_or(host);
if let Ok(ip) = normalized_host.parse::<IpAddr>() {
return Ok(Some(SocketAddr::new(ip, port)));
}
let mut results = lookup_host((normalized_host, port)).await?;
Ok(results.next())
}
fn local_addresses_from_port(port: u16) -> Vec<String> {
let mut addresses = Vec::new();
push_private_interface_ips(&mut addresses);
push_local_probe(&mut addresses, "0.0.0.0:0", "8.8.8.8:80");
push_local_probe(&mut addresses, "[::]:0", "[2001:4860:4860::8888]:80");
push_bound_addr(&mut addresses, ("0.0.0.0", port));
push_bound_addr(&mut addresses, ("::", port));
addresses
}
fn push_private_interface_ips(addresses: &mut Vec<String>) {
for ip in private_interface_ips() {
push_ip(addresses, ip);
}
}
#[cfg(unix)]
fn private_interface_ips() -> Vec<IpAddr> {
let mut output = Vec::new();
let mut ifaddrs: *mut libc::ifaddrs = std::ptr::null_mut();
// SAFETY: `getifaddrs` initializes `ifaddrs` on success, and the linked
// list is valid until `freeifaddrs` is called.
let rc = unsafe { libc::getifaddrs(&mut ifaddrs) };
if rc != 0 || ifaddrs.is_null() {
return output;
}
let mut cursor = ifaddrs;
while !cursor.is_null() {
// SAFETY: `cursor` points at a valid node from the `getifaddrs` list.
let entry = unsafe { &*cursor };
let flags = entry.ifa_flags as i32;
let is_up = (flags & libc::IFF_UP) != 0;
let is_loopback = (flags & libc::IFF_LOOPBACK) != 0;
if is_up && !is_loopback && !entry.ifa_addr.is_null() {
// SAFETY: `ifa_addr` is non-null and its concrete type matches
// `sa_family` for this entry.
let maybe_ip = unsafe {
match (*entry.ifa_addr).sa_family as i32 {
libc::AF_INET => {
let sockaddr = &*(entry.ifa_addr as *const libc::sockaddr_in);
Some(IpAddr::V4(Ipv4Addr::from(
sockaddr.sin_addr.s_addr.to_ne_bytes(),
)))
}
libc::AF_INET6 => {
let sockaddr = &*(entry.ifa_addr as *const libc::sockaddr_in6);
Some(IpAddr::V6(Ipv6Addr::from(sockaddr.sin6_addr.s6_addr)))
}
_ => None,
}
};
if let Some(ip) = maybe_ip
&& is_private_overlay_candidate_ip(ip)
{
output.push(ip);
}
}
cursor = entry.ifa_next;
}
// SAFETY: `ifaddrs` came from `getifaddrs` and has not yet been freed.
unsafe { libc::freeifaddrs(ifaddrs) };
output
}
#[cfg(not(unix))]
fn private_interface_ips() -> Vec<IpAddr> {
Vec::new()
}
fn is_private_overlay_candidate_ip(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => v4.is_private(),
IpAddr::V6(v6) => v6.is_unique_local(),
}
}
fn push_local_probe(addresses: &mut Vec<String>, bind_addr: &str, connect_addr: &str) {
if let Ok(socket) = std::net::UdpSocket::bind(bind_addr)
&& socket.connect(connect_addr).is_ok()
&& let Ok(local_addr) = socket.local_addr()
{
push_ip(addresses, local_addr.ip());
}
}
fn push_bound_addr<A: std::net::ToSocketAddrs>(addresses: &mut Vec<String>, bind_addr: A) {
if let Ok(local_addr) =
std::net::UdpSocket::bind(bind_addr).and_then(|socket| socket.local_addr())
{
push_ip(addresses, local_addr.ip());
}
}
fn push_ip(addresses: &mut Vec<String>, ip: IpAddr) {
if ip.is_unspecified() {
return;
}
let ip = ip.to_string();
if !addresses.contains(&ip) {
addresses.push(ip);
}
}
fn random_txn_id() -> [u8; 12] {
let mut txn_id = [0u8; 12];
for byte in &mut txn_id {
*byte = rand::random::<u8>();
}
txn_id
}
#[cfg(test)]
mod tests {
use super::is_private_overlay_candidate_ip;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
#[test]
fn private_overlay_candidate_filter_includes_rfc1918_and_ula() {
assert!(is_private_overlay_candidate_ip(IpAddr::V4(Ipv4Addr::new(
192, 168, 1, 10
))));
assert!(is_private_overlay_candidate_ip(IpAddr::V4(Ipv4Addr::new(
10, 0, 0, 4
))));
assert!(is_private_overlay_candidate_ip(IpAddr::V4(Ipv4Addr::new(
172, 16, 5, 20
))));
assert!(is_private_overlay_candidate_ip(IpAddr::V6(
"fd00::1234".parse::<Ipv6Addr>().unwrap()
)));
}
#[test]
fn private_overlay_candidate_filter_excludes_public_and_link_local() {
assert!(!is_private_overlay_candidate_ip(IpAddr::V4(Ipv4Addr::new(
8, 8, 8, 8
))));
assert!(!is_private_overlay_candidate_ip(IpAddr::V4(Ipv4Addr::new(
169, 254, 1, 10
))));
assert!(!is_private_overlay_candidate_ip(IpAddr::V6(
"fe80::1".parse::<Ipv6Addr>().unwrap()
)));
assert!(!is_private_overlay_candidate_ip(IpAddr::V6(
"2001:db8::1".parse::<Ipv6Addr>().unwrap()
)));
}
}
+411
View File
@@ -0,0 +1,411 @@
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,
};
use super::stun::{parse_stun_binding_success, parse_stun_url};
use super::traversal::{
PunchStrategy, build_punch_packet, parse_punch_packet, plan_punch_targets,
planned_remote_endpoints, session_hash,
};
use super::{
ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, OverlayAdvert, OverlayEndpointAdvert,
OverlayTransportKind, PunchHint, PunchPacketKind, TraversalAddress,
};
#[derive(Clone, Copy, PartialEq, Eq)]
enum NatType {
RestrictedCone,
PortRestricted,
Symmetric,
}
fn addr(ip: &str, port: u16) -> TraversalAddress {
TraversalAddress {
protocol: "udp".to_string(),
ip: ip.to_string(),
port,
}
}
fn can_reach(local_nat: NatType, remote_nat: NatType) -> bool {
if local_nat == NatType::Symmetric || remote_nat == NatType::Symmetric {
return false;
}
!(local_nat == NatType::PortRestricted && remote_nat == NatType::PortRestricted)
}
fn signed_overlay_advert_event(created_at_secs: u64, expiration_secs: Option<u64>) -> nostr::Event {
let keys = nostr::Keys::generate();
let content = r#"{"identifier":"fips-overlay-v1","version":1,"endpoints":[{"transport":"tcp","addr":"203.0.113.10:443"}]}"#;
let mut builder = EventBuilder::new(Kind::Custom(ADVERT_KIND), content)
.custom_created_at(Timestamp::from(created_at_secs));
if let Some(expiration_secs) = expiration_secs {
builder = builder.tags([Tag::expiration(Timestamp::from(expiration_secs))]);
}
builder.sign_with_keys(&keys).unwrap()
}
#[test]
fn serializes_direct_overlay_advert_without_nat_metadata() {
let advert = OverlayAdvert {
identifier: ADVERT_IDENTIFIER.to_string(),
version: ADVERT_VERSION,
endpoints: vec![
OverlayEndpointAdvert {
transport: OverlayTransportKind::Tcp,
addr: "203.0.113.10:443".to_string(),
},
OverlayEndpointAdvert {
transport: OverlayTransportKind::Tor,
addr: "exampleonion.onion:1234".to_string(),
},
],
signal_relays: None,
stun_servers: None,
};
let json = serde_json::to_string(&advert).unwrap();
assert!(json.contains("\"endpoints\""));
assert!(!json.contains("\"signalRelays\""));
assert!(!json.contains("\"stunServers\""));
}
#[test]
fn serializes_nat_overlay_advert_with_metadata() {
let advert = OverlayAdvert {
identifier: ADVERT_IDENTIFIER.to_string(),
version: ADVERT_VERSION,
endpoints: vec![OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: "nat".to_string(),
}],
signal_relays: Some(vec!["wss://relay.example".to_string()]),
stun_servers: Some(vec!["stun:stun.example.org:3478".to_string()]),
};
let json = serde_json::to_string(&advert).unwrap();
assert!(json.contains("\"signalRelays\""));
assert!(json.contains("\"stunServers\""));
}
#[test]
fn rejects_invalid_overlay_adverts() {
let missing_nat_metadata = OverlayAdvert {
identifier: ADVERT_IDENTIFIER.to_string(),
version: ADVERT_VERSION,
endpoints: vec![OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: "nat".to_string(),
}],
signal_relays: None,
stun_servers: None,
};
assert!(NostrDiscovery::validate_overlay_advert(missing_nat_metadata).is_err());
let wrong_identifier = OverlayAdvert {
identifier: "not-fips-overlay".to_string(),
version: ADVERT_VERSION,
endpoints: vec![OverlayEndpointAdvert {
transport: OverlayTransportKind::Tcp,
addr: "203.0.113.10:443".to_string(),
}],
signal_relays: None,
stun_servers: None,
};
assert!(NostrDiscovery::validate_overlay_advert(wrong_identifier).is_err());
}
#[test]
fn advert_freshness_rejects_expired_events() {
let now_secs = Timestamp::now().as_secs();
let event = signed_overlay_advert_event(now_secs, Some(now_secs.saturating_sub(1)));
let valid_until =
NostrDiscovery::compute_advert_valid_until_ms(&event, 600_000, now_secs * 1000);
assert!(valid_until.is_none());
}
#[test]
fn advert_freshness_rejects_stale_created_at_without_expiration() {
let now_secs = Timestamp::now().as_secs();
let stale_created = now_secs.saturating_sub(10_000);
let event = signed_overlay_advert_event(stale_created, None);
let valid_until =
NostrDiscovery::compute_advert_valid_until_ms(&event, 600_000, now_secs * 1000);
assert!(valid_until.is_none());
}
#[test]
fn advert_freshness_uses_earliest_expiration_bound() {
let now_secs = Timestamp::now().as_secs();
let event = signed_overlay_advert_event(now_secs.saturating_sub(10), Some(now_secs + 30));
let valid_until =
NostrDiscovery::compute_advert_valid_until_ms(&event, 3_600_000, now_secs * 1000)
.expect("event should be fresh");
assert_eq!(valid_until, (now_secs + 30) * 1000);
}
#[test]
fn parses_stun_urls() {
let parsed = parse_stun_url("stun:stun.l.google.com:19302").unwrap();
assert_eq!(parsed.host, "stun.l.google.com");
assert_eq!(parsed.port, 19302);
}
#[test]
fn parses_ipv6_stun_urls() {
let parsed = parse_stun_url("stun:[2001:db8::10]:3478").unwrap();
assert_eq!(parsed.host, "[2001:db8::10]");
assert_eq!(parsed.port, 3478);
}
#[test]
fn parses_ipv6_xor_mapped_address() {
let txn_id = [
0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef, 0x10, 0x32, 0x54, 0x76,
];
let addr = std::net::SocketAddr::new("2001:db8::1234".parse().unwrap(), 3478);
let port = addr.port() ^ 0x2112;
let mut attr = Vec::with_capacity(24);
attr.extend_from_slice(&0x0020u16.to_be_bytes());
attr.extend_from_slice(&20u16.to_be_bytes());
attr.push(0);
attr.push(0x02);
attr.extend_from_slice(&port.to_be_bytes());
let ipv6 = match addr.ip() {
std::net::IpAddr::V6(ip) => ip.octets(),
std::net::IpAddr::V4(_) => panic!("expected IPv6 test address"),
};
let cookie = 0x2112_a442u32.to_be_bytes();
for index in 0..16 {
let mask = if index < 4 {
cookie[index]
} else {
txn_id[index - 4]
};
attr.push(ipv6[index] ^ mask);
}
let mut packet = Vec::with_capacity(44);
packet.extend_from_slice(&0x0101u16.to_be_bytes());
packet.extend_from_slice(&(attr.len() as u16).to_be_bytes());
packet.extend_from_slice(&0x2112_a442u32.to_be_bytes());
packet.extend_from_slice(&txn_id);
packet.extend_from_slice(&attr);
assert_eq!(parse_stun_binding_success(&packet, &txn_id), Some(addr));
}
#[test]
fn builds_and_parses_probe_packets() {
let packet = build_punch_packet(PunchPacketKind::Probe, 7, "sess-1");
let parsed = parse_punch_packet(&packet).unwrap();
assert_eq!(parsed.kind, PunchPacketKind::Probe);
assert_eq!(parsed.sequence, 7);
assert_eq!(parsed.session_hash, session_hash("sess-1"));
}
#[test]
fn validates_offer_answer_pair() {
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 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![addr("192.168.1.20", 63000)],
Some("stun:example.org:3478".to_string()),
Some(PunchHint {
start_at_ms: 1_700_000_002_000,
interval_ms: 200,
duration_ms: 10_000,
}),
None,
);
assert!(
validate_traversal_answer_for_offer(
&offer,
&answer,
1_700_000_000_900,
60_000,
"npub1server",
"npub1client",
)
.is_ok()
);
}
#[test]
fn rejects_offer_with_mismatched_actual_sender() {
let offer = create_traversal_offer(
"sess-1".to_string(),
1_700_000_000_000,
60_000,
"offer-1".to_string(),
"npub1claimed".to_string(),
"npub1server".to_string(),
None,
vec![addr("192.168.1.10", 62000)],
None,
);
let result = validate_offer_freshness(
&offer,
1_700_000_000_100,
60_000,
"npub1actual",
"npub1server",
);
assert!(result.is_err());
}
#[test]
fn rejects_answer_with_mismatched_actual_sender() {
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 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![addr("192.168.1.20", 63000)],
Some("stun:example.org:3478".to_string()),
Some(PunchHint {
start_at_ms: 1_700_000_002_000,
interval_ms: 200,
duration_ms: 10_000,
}),
None,
);
let result = validate_traversal_answer_for_offer(
&offer,
&answer,
1_700_000_000_900,
60_000,
"npub1spoofed",
"npub1client",
);
assert!(result.is_err());
}
#[test]
fn plans_reflexive_targets_before_lan() {
let planned = plan_punch_targets(
&[addr("192.168.1.10", 62000)],
Some(&addr("203.0.113.10", 62000)),
&[addr("192.168.1.20", 63000)],
Some(&addr("198.51.100.20", 63000)),
);
assert_eq!(planned[0].strategy, PunchStrategy::Reflexive);
assert_eq!(planned[1].strategy, PunchStrategy::Lan);
}
#[test]
fn simulated_lan_scenario_includes_lan_target_and_succeeds() {
let planned = plan_punch_targets(
&[addr("192.168.1.10", 62000)],
Some(&addr("203.0.113.10", 62000)),
&[addr("192.168.1.20", 63000)],
Some(&addr("198.51.100.20", 63000)),
);
assert!(
planned
.iter()
.any(|target| target.strategy == PunchStrategy::Lan)
);
assert!(can_reach(NatType::RestrictedCone, NatType::RestrictedCone));
}
#[test]
fn simulated_symmetric_nat_scenario_requires_fallback() {
let planned = plan_punch_targets(
&[addr("10.0.0.10", 62000)],
Some(&addr("203.0.113.10", 62000)),
&[addr("10.0.1.10", 63000)],
Some(&addr("198.51.100.20", 63000)),
);
assert!(
planned
.iter()
.any(|target| target.strategy == PunchStrategy::Reflexive)
);
assert!(!can_reach(NatType::Symmetric, NatType::RestrictedCone));
}
#[test]
fn planned_remote_endpoints_include_private_and_reflexive_paths() {
let endpoints = planned_remote_endpoints(
&[addr("192.168.1.10", 62000)],
Some(&addr("203.0.113.10", 62000)),
&[addr("192.168.1.20", 63000)],
Some(&addr("198.51.100.20", 63000)),
)
.expect("endpoint planning should succeed");
assert!(endpoints.contains(&"192.168.1.20:63000".parse().unwrap()));
assert!(endpoints.contains(&"198.51.100.20:63000".parse().unwrap()));
}
#[tokio::test]
async fn signal_events_use_current_timestamps() {
let sender = nostr::Keys::generate();
let receiver = nostr::Keys::generate();
let rumor = EventBuilder::private_msg_rumor(receiver.public_key(), "hello".to_string())
.build(sender.public_key());
let before = Timestamp::now().as_secs();
let event = build_signal_event(
&sender,
receiver.public_key(),
rumor,
Timestamp::from(before + 30),
)
.await
.expect("signal event should build");
let after = Timestamp::now().as_secs();
let created_at = event.created_at.as_secs();
assert!(created_at >= before);
assert!(created_at <= after);
}
+276
View File
@@ -0,0 +1,276 @@
use std::net::SocketAddr;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::net::UdpSocket;
use super::types::{
BootstrapError, PUNCH_ACK_MAGIC, PUNCH_MAGIC, PunchHint, PunchPacket, PunchPacketKind,
TraversalAddress,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum AddressSource {
Local,
Reflexive,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum PunchStrategy {
Lan,
Reflexive,
Mixed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct PlannedPunchTarget {
pub(super) strategy: PunchStrategy,
pub(super) local_source: AddressSource,
pub(super) remote_source: AddressSource,
pub(super) local: TraversalAddress,
pub(super) remote: TraversalAddress,
}
fn same_subnet_24(left: &TraversalAddress, right: &TraversalAddress) -> bool {
let left_parts = left.ip.split('.').collect::<Vec<_>>();
let right_parts = right.ip.split('.').collect::<Vec<_>>();
left_parts.len() == 4 && right_parts.len() == 4 && left_parts[..3] == right_parts[..3]
}
pub(super) fn plan_punch_targets(
local_addresses: &[TraversalAddress],
local_reflexive_address: Option<&TraversalAddress>,
remote_addresses: &[TraversalAddress],
remote_reflexive_address: Option<&TraversalAddress>,
) -> Vec<PlannedPunchTarget> {
let mut planned = Vec::new();
let mut push_unique = |target: PlannedPunchTarget| {
if !planned.iter().any(|existing| existing == &target) {
planned.push(target);
}
};
// Reflexive ↔ Reflexive first: the only path that's reliable across
// arbitrary network topologies. Try this before any host-candidate path
// so we don't latch onto a misleading asymmetric route (e.g. an offer's
// private host candidate that we can reach one-way via a routed VPN).
if let (Some(local), Some(remote)) = (local_reflexive_address, remote_reflexive_address) {
push_unique(PlannedPunchTarget {
strategy: PunchStrategy::Reflexive,
local_source: AddressSource::Reflexive,
remote_source: AddressSource::Reflexive,
local: local.clone(),
remote: remote.clone(),
});
}
// Same-LAN paths (matching /24 between local and remote host candidates).
// Only fires when both sides exposed local candidates AND they share a
// /24 prefix.
for local in local_addresses {
for remote in remote_addresses {
if same_subnet_24(local, remote) {
push_unique(PlannedPunchTarget {
strategy: PunchStrategy::Lan,
local_source: AddressSource::Local,
remote_source: AddressSource::Local,
local: local.clone(),
remote: remote.clone(),
});
}
}
}
// Mixed paths cover hairpin and one-side-public scenarios.
if let Some(remote) = remote_reflexive_address {
for local in local_addresses {
push_unique(PlannedPunchTarget {
strategy: PunchStrategy::Mixed,
local_source: AddressSource::Local,
remote_source: AddressSource::Reflexive,
local: local.clone(),
remote: remote.clone(),
});
}
}
if let Some(local) = local_reflexive_address {
for remote in remote_addresses {
push_unique(PlannedPunchTarget {
strategy: PunchStrategy::Mixed,
local_source: AddressSource::Reflexive,
remote_source: AddressSource::Local,
local: local.clone(),
remote: remote.clone(),
});
}
}
planned
}
pub(super) fn planned_remote_endpoints(
local_addresses: &[TraversalAddress],
local_reflexive_address: Option<&TraversalAddress>,
remote_addresses: &[TraversalAddress],
remote_reflexive_address: Option<&TraversalAddress>,
) -> Result<Vec<SocketAddr>, BootstrapError> {
let mut remotes = Vec::new();
for target in plan_punch_targets(
local_addresses,
local_reflexive_address,
remote_addresses,
remote_reflexive_address,
) {
let remote = SocketAddr::new(
target
.remote
.ip
.parse()
.map_err(|_| BootstrapError::Protocol("invalid-remote-ip".to_string()))?,
target.remote.port,
);
if !remotes.contains(&remote) {
remotes.push(remote);
}
}
Ok(remotes)
}
pub(super) async fn run_punch_attempt(
socket: &std::net::UdpSocket,
session_id: &str,
targets: &[SocketAddr],
punch: PunchHint,
timeout: Duration,
) -> Result<SocketAddr, BootstrapError> {
if targets.is_empty() {
return Err(BootstrapError::Protocol("no-punch-targets".to_string()));
}
let udp = Arc::new(UdpSocket::from_std(socket.try_clone()?)?);
let started_at = tokio::time::Instant::now();
let finish_at = started_at + timeout;
let delay = Duration::from_millis(punch.start_at_ms.saturating_sub(now_ms()));
let send_socket = Arc::clone(&udp);
let send_targets = targets.to_vec();
let send_session = session_id.to_string();
let send_handle = tokio::spawn(async move {
tokio::time::sleep(delay).await;
let end = Instant::now() + Duration::from_millis(punch.duration_ms.max(1));
let mut sequence = 0u32;
while Instant::now() < end {
let packet = build_punch_packet(PunchPacketKind::Probe, sequence, &send_session);
for target in &send_targets {
let _ = send_socket.send_to(&packet, target).await;
}
sequence = sequence.wrapping_add(1);
tokio::time::sleep(Duration::from_millis(punch.interval_ms.max(20))).await;
}
});
let expected_hash = session_hash(session_id);
let mut buf = [0u8; 2048];
let result = loop {
let recv = tokio::time::timeout_at(finish_at, udp.recv_from(&mut buf)).await;
let Ok(Ok((len, remote))) = recv else {
break Err(BootstrapError::PunchTimeout(session_id.to_string()));
};
let Ok(packet) = parse_punch_packet(&buf[..len]) else {
continue;
};
if packet.session_hash != expected_hash {
continue;
}
if packet.kind == PunchPacketKind::Probe {
let ack = build_punch_packet(PunchPacketKind::Ack, packet.sequence, session_id);
let _ = udp.send_to(&ack, remote).await;
}
break Ok(remote);
};
send_handle.abort();
result
}
pub(super) fn nonce() -> String {
format!("{}-{:016x}", now_ms(), rand::random::<u64>())
}
pub(super) fn now_ms() -> u64 {
struct ClockAnchor {
started_at: Instant,
started_unix_ms: u64,
}
static ANCHOR: OnceLock<ClockAnchor> = OnceLock::new();
let anchor = ANCHOR.get_or_init(|| ClockAnchor {
started_at: Instant::now(),
started_unix_ms: SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as u64)
.unwrap_or(0),
});
anchor
.started_unix_ms
.saturating_add(anchor.started_at.elapsed().as_millis() as u64)
}
pub(super) fn session_hash(session_id: &str) -> [u8; 16] {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(session_id.as_bytes());
let mut output = [0u8; 16];
output.copy_from_slice(&digest[..16]);
output
}
pub(super) fn build_punch_packet(
kind: PunchPacketKind,
sequence: u32,
session_id: &str,
) -> [u8; 24] {
let magic = match kind {
PunchPacketKind::Probe => PUNCH_MAGIC,
PunchPacketKind::Ack => PUNCH_ACK_MAGIC,
};
let mut packet = [0u8; 24];
packet[..4].copy_from_slice(&magic.to_be_bytes());
packet[4..8].copy_from_slice(&sequence.to_be_bytes());
packet[8..24].copy_from_slice(&session_hash(session_id));
packet
}
pub(super) fn parse_punch_packet(bytes: &[u8]) -> Result<PunchPacket, BootstrapError> {
if bytes.len() < 24 {
return Err(BootstrapError::Protocol(
"invalid-punch-packet-length".to_string(),
));
}
let magic = u32::from_be_bytes(
bytes[..4]
.try_into()
.map_err(|_| BootstrapError::Protocol("invalid-punch-magic".to_string()))?,
);
let kind = match magic {
PUNCH_MAGIC => PunchPacketKind::Probe,
PUNCH_ACK_MAGIC => PunchPacketKind::Ack,
_ => {
return Err(BootstrapError::Protocol("invalid-punch-magic".to_string()));
}
};
let sequence = u32::from_be_bytes(
bytes[4..8]
.try_into()
.map_err(|_| BootstrapError::Protocol("invalid-punch-seq".to_string()))?,
);
let mut hash = [0u8; 16];
hash.copy_from_slice(&bytes[8..24]);
Ok(PunchPacket {
kind,
sequence,
session_hash: hash,
})
}
+179
View File
@@ -0,0 +1,179 @@
use crate::config::PeerConfig;
use crate::discovery::EstablishedTraversal;
use serde::{Deserialize, Serialize};
pub const ADVERT_KIND: u16 = 37195;
pub const ADVERT_IDENTIFIER: &str = "fips-overlay-v1";
pub const ADVERT_VERSION: u32 = 1;
pub const SIGNAL_KIND: u16 = 21059;
pub const PUNCH_MAGIC: u32 = 0x4E505443;
pub const PUNCH_ACK_MAGIC: u32 = 0x4E505441;
pub const PROTOCOL_VERSION: &str = "1";
#[derive(Debug, thiserror::Error)]
pub enum BootstrapError {
#[error("bootstrap disabled")]
Disabled,
#[error("peer {0} has no overlay advert")]
MissingAdvert(String),
#[error("peer {0} advert does not contain udp:nat endpoint")]
MissingNatEndpoint(String),
#[error("peer {0} has no usable traversal relays")]
MissingRelays(String),
#[error("invalid overlay advert: {0}")]
InvalidAdvert(String),
#[error("invalid npub '{npub}': {reason}")]
InvalidPeerNpub { npub: String, reason: String },
#[error("signal timeout waiting for answer from {0}")]
SignalTimeout(String),
#[error("traversal attempt timed out for {0}")]
PunchTimeout(String),
#[error("replayed or duplicate session id: {0}")]
Replay(String),
#[error("stun failed: {0}")]
Stun(String),
#[error("protocol error: {0}")]
Protocol(String),
#[error("nostr error: {0}")]
Nostr(String),
#[error("io error: {0}")]
Io(#[from] std::io::Error),
#[error("serde error: {0}")]
Serde(#[from] serde_json::Error),
#[error("event parse error: {0}")]
EventParse(String),
}
#[derive(Debug)]
pub enum BootstrapEvent {
Established {
traversal: EstablishedTraversal,
},
Failed {
peer_config: PeerConfig,
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TraversalAddress {
pub protocol: String,
pub ip: String,
pub port: u16,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PunchHint {
#[serde(rename = "startAtMs")]
pub start_at_ms: u64,
#[serde(rename = "intervalMs")]
pub interval_ms: u64,
#[serde(rename = "durationMs")]
pub duration_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum OverlayTransportKind {
Udp,
Tcp,
Tor,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OverlayEndpointAdvert {
pub transport: OverlayTransportKind,
pub addr: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OverlayAdvert {
pub identifier: String,
pub version: u32,
pub endpoints: Vec<OverlayEndpointAdvert>,
#[serde(rename = "signalRelays", skip_serializing_if = "Option::is_none")]
pub signal_relays: Option<Vec<String>>,
#[serde(rename = "stunServers", skip_serializing_if = "Option::is_none")]
pub stun_servers: Option<Vec<String>>,
}
impl OverlayAdvert {
pub fn has_udp_nat_endpoint(&self) -> bool {
self.endpoints.iter().any(|endpoint| {
endpoint.transport == OverlayTransportKind::Udp
&& endpoint.addr.eq_ignore_ascii_case("nat")
})
}
}
#[derive(Debug, Clone)]
pub struct CachedOverlayAdvert {
pub author_npub: String,
pub advert: OverlayAdvert,
pub created_at: u64,
pub valid_until_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TraversalOffer {
#[serde(rename = "type")]
pub message_type: String,
#[serde(rename = "sessionId")]
pub session_id: String,
#[serde(rename = "issuedAt")]
pub issued_at: u64,
#[serde(rename = "expiresAt")]
pub expires_at: u64,
pub nonce: String,
#[serde(rename = "senderNpub")]
pub sender_npub: String,
#[serde(rename = "recipientNpub")]
pub recipient_npub: String,
#[serde(rename = "reflexiveAddress")]
pub reflexive_address: Option<TraversalAddress>,
#[serde(rename = "localAddresses")]
pub local_addresses: Vec<TraversalAddress>,
#[serde(rename = "stunServer")]
pub stun_server: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TraversalAnswer {
#[serde(rename = "type")]
pub message_type: String,
#[serde(rename = "sessionId")]
pub session_id: String,
#[serde(rename = "issuedAt")]
pub issued_at: u64,
#[serde(rename = "expiresAt")]
pub expires_at: u64,
pub nonce: String,
#[serde(rename = "senderNpub")]
pub sender_npub: String,
#[serde(rename = "recipientNpub")]
pub recipient_npub: String,
#[serde(rename = "inReplyTo")]
pub in_reply_to: String,
pub accepted: bool,
#[serde(rename = "reflexiveAddress")]
pub reflexive_address: Option<TraversalAddress>,
#[serde(rename = "localAddresses")]
pub local_addresses: Vec<TraversalAddress>,
#[serde(rename = "stunServer")]
pub stun_server: Option<String>,
pub punch: Option<PunchHint>,
pub reason: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PunchPacketKind {
Probe,
Ack,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PunchPacket {
pub kind: PunchPacketKind,
pub sequence: u32,
pub session_hash: [u8; 16],
}
+5 -1
View File
@@ -7,7 +7,8 @@ pub mod bloom;
pub mod cache;
pub mod config;
pub mod control;
#[cfg(feature = "gateway")]
pub mod discovery;
#[cfg(target_os = "linux")]
pub mod gateway;
pub mod identity;
pub mod mmp;
@@ -31,6 +32,9 @@ pub use identity::{
pub use config::{Config, ConfigError, IdentityConfig, TorConfig, UdpConfig};
pub use upper::config::{DnsConfig, TunConfig};
// Re-export discovery types
pub use discovery::{BootstrapHandoffResult, EstablishedTraversal};
// Re-export tree types
pub use tree::{CoordEntry, ParentDeclaration, TreeCoordinate, TreeError, TreeState};
+14 -11
View File
@@ -2,10 +2,12 @@
//!
//! Two complementary mechanisms:
//!
//! - **`DiscoveryBackoff`** (originator-side): Exponential backoff for failed
//! lookups. After a lookup times out, suppresses re-initiation with
//! increasing delays (30s → 60s → 300s cap). Reset on topology changes
//! (parent change, new peer, first RTT, reconnection).
//! - **`DiscoveryBackoff`** (originator-side, optional): Exponential
//! suppression of fresh lookups after the per-attempt sequence in
//! `node.discovery.attempt_timeouts_secs` has been exhausted.
//! **Disabled by default** (base/cap = 0); the per-attempt sequence
//! is the only retry pacing in the standard configuration. Reset on
//! topology changes (parent change, new peer, first RTT, reconnection).
//!
//! - **`DiscoveryForwardRateLimiter`** (transit-side): Per-target minimum
//! interval for forwarded requests. Defense-in-depth against misbehaving
@@ -19,11 +21,11 @@ use std::time::{Duration, Instant};
// Originator-side: Discovery Backoff
// ============================================================================
/// Default base backoff after first lookup failure.
const DEFAULT_BACKOFF_BASE_SECS: u64 = 30;
/// Default base backoff after first lookup failure. `0` = disabled.
const DEFAULT_BACKOFF_BASE_SECS: u64 = 0;
/// Default maximum backoff cap.
const DEFAULT_BACKOFF_MAX_SECS: u64 = 300;
/// Default maximum backoff cap. `0` = disabled.
const DEFAULT_BACKOFF_MAX_SECS: u64 = 0;
/// Backoff multiplier per consecutive failure.
const BACKOFF_MULTIPLIER: u64 = 2;
@@ -49,7 +51,7 @@ struct BackoffEntry {
}
impl DiscoveryBackoff {
/// Create with default parameters (30s base, 300s cap).
/// Create with default parameters (disabled — base/cap = 0).
pub fn new() -> Self {
Self::with_params(DEFAULT_BACKOFF_BASE_SECS, DEFAULT_BACKOFF_MAX_SECS)
}
@@ -245,7 +247,8 @@ mod tests {
#[test]
fn test_backoff_suppressed_after_failure() {
let mut backoff = DiscoveryBackoff::new();
// Backoff is opt-in; exercise the suppression path with explicit params.
let mut backoff = DiscoveryBackoff::with_params(30, 300);
backoff.record_failure(&addr(1));
assert!(backoff.is_suppressed(&addr(1)));
// Different target not affected
@@ -254,7 +257,7 @@ mod tests {
#[test]
fn test_backoff_cleared_on_success() {
let mut backoff = DiscoveryBackoff::new();
let mut backoff = DiscoveryBackoff::with_params(30, 300);
backoff.record_failure(&addr(1));
assert!(backoff.is_suppressed(&addr(1)));
+35 -34
View File
@@ -444,31 +444,26 @@ impl Node {
/// Initiate a discovery lookup if one is not already pending for this target.
///
/// Checks: pending dedup, backoff, bloom filter pre-check. If all pass,
/// initiates the lookup. If no tree peers have the target in their bloom
/// filter, the lookup is skipped (bloom miss) and recorded as a failure
/// for backoff purposes.
/// Checks: pending dedup, post-failure backoff (off by default), bloom
/// filter pre-check. If all pass, sends the first attempt's LookupRequest.
/// Subsequent attempts (with fresh request_ids) are scheduled by
/// [`Self::check_pending_lookups`] when each attempt's per-attempt timeout
/// expires, using the sequence in `node.discovery.attempt_timeouts_secs`.
pub(in crate::node) async fn maybe_initiate_lookup(&mut self, dest: &NodeAddr) {
let now_ms = Self::now_ms();
let lookup_timeout_ms = self.config.node.discovery.timeout_secs * 1000;
// Check pending lookup dedup (in-flight)
if let Some(entry) = self.pending_lookups.get(dest) {
let age_ms = now_ms.saturating_sub(entry.initiated_ms);
let attempt = entry.attempt;
if age_ms < lookup_timeout_ms {
self.stats_mut().discovery.req_deduplicated += 1;
debug!(
target_node = %self.peer_display_name(dest),
age_ms = age_ms,
attempt = attempt,
"Discovery lookup deduplicated, already pending"
);
return;
}
// Dedup: any pending lookup means we are already trying.
if self.pending_lookups.contains_key(dest) {
self.stats_mut().discovery.req_deduplicated += 1;
debug!(
target_node = %self.peer_display_name(dest),
"Discovery lookup deduplicated, already pending"
);
return;
}
// Check backoff from previous failures
// Optional post-failure suppression. Defaults are 0/0 (inert);
// operators can opt in by setting `node.discovery.backoff_*_secs`.
if self.discovery_backoff.is_suppressed(dest) {
self.stats_mut().discovery.req_backoff_suppressed += 1;
debug!(
@@ -508,27 +503,33 @@ impl Node {
}
}
/// Check pending lookups for retry or timeout.
/// Check pending lookups for next-attempt or final timeout.
///
/// Called periodically from the tick handler. For each pending lookup:
/// - If retry interval elapsed and attempts remain: resend
/// - If total timeout elapsed: fail, record backoff, send ICMP unreachable
/// Called periodically from the tick handler. The lookup state machine
/// runs through `node.discovery.attempt_timeouts_secs` (default
/// `[1, 2, 4, 8]`): each entry is the deadline for one attempt. When the
/// current attempt's deadline elapses:
/// - If more entries remain: send the next attempt with a fresh
/// `request_id`.
/// - Otherwise: declare the destination unreachable, drop queued packets,
/// and emit ICMPv6 destination-unreachable for each.
pub(in crate::node) async fn check_pending_lookups(&mut self, now_ms: u64) {
let timeout_ms = self.config.node.discovery.timeout_secs * 1000;
let retry_ms = self.config.node.discovery.retry_interval_secs * 1000;
let timeouts = self.config.node.discovery.attempt_timeouts_secs.clone();
let max_attempts = timeouts.len() as u8;
// Collect targets needing action
let mut to_retry: Vec<NodeAddr> = Vec::new();
let mut to_timeout: Vec<NodeAddr> = Vec::new();
for (&target, entry) in &self.pending_lookups {
let age = now_ms.saturating_sub(entry.initiated_ms);
if age >= timeout_ms {
to_timeout.push(target);
} else if entry.attempt < self.config.node.discovery.max_attempts
&& now_ms.saturating_sub(entry.last_sent_ms) >= retry_ms
{
to_retry.push(target);
let attempt_idx = (entry.attempt as usize).saturating_sub(1);
let attempt_timeout_ms = timeouts.get(attempt_idx).copied().unwrap_or(0) * 1000;
if now_ms.saturating_sub(entry.last_sent_ms) >= attempt_timeout_ms {
if entry.attempt >= max_attempts {
to_timeout.push(target);
} else {
to_retry.push(target);
}
}
}
@@ -556,7 +557,7 @@ impl Node {
self.stats_mut().discovery.resp_timed_out += 1;
self.pending_lookups.remove(&addr);
// Record failure for backoff
// Record failure for optional backoff
self.discovery_backoff.record_failure(&addr);
let failures = self.discovery_backoff.failure_count(&addr);
+3
View File
@@ -179,6 +179,9 @@ impl Node {
// Remove link and address mapping
self.remove_link(&link_id);
if let Some(transport_id) = transport_id {
self.cleanup_bootstrap_transport_if_unused(transport_id);
}
// Tree state cleanup
let tree_changed = self.handle_peer_removal_tree_cleanup(node_addr);
+3 -4
View File
@@ -114,12 +114,11 @@ impl Node {
}
_ = tick.tick() => {
self.check_timeouts();
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let now_ms = Self::now_ms();
self.reload_peer_acl();
self.poll_pending_connects().await;
#[cfg(feature = "nostr-discovery")]
self.poll_nostr_discovery().await;
self.resend_pending_handshakes(now_ms).await;
self.resend_pending_rekeys(now_ms).await;
self.resend_pending_session_handshakes(now_ms).await;
+5 -4
View File
@@ -16,10 +16,7 @@ impl Node {
return;
}
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let now_ms = Self::now_ms();
let timeout_ms = self.config.node.rate_limit.handshake_timeout_secs * 1000;
let stale: Vec<LinkId> = self
@@ -70,6 +67,7 @@ impl Node {
Some(c) => c,
None => return,
};
let transport_id = conn.transport_id();
// Free session index and pending_outbound/pending_inbound if allocated
if let Some(idx) = conn.our_index() {
@@ -82,6 +80,9 @@ impl Node {
// Remove link and addr_to_link
self.remove_link(&link_id);
if let Some(transport_id) = transport_id {
self.cleanup_bootstrap_transport_if_unused(transport_id);
}
}
/// Resend handshake messages for pending connections.
+692 -97
View File
@@ -1,6 +1,13 @@
//! Node lifecycle management: start, stop, and peer connection initiation.
use super::{Node, NodeError, NodeState};
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
#[cfg(feature = "nostr-discovery")]
use crate::discovery::nostr::{
ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapEvent, NostrDiscovery, OverlayAdvert,
OverlayEndpointAdvert, OverlayTransportKind,
};
use crate::discovery::{BootstrapHandoffResult, EstablishedTraversal};
use crate::node::acl::PeerAclContext;
use crate::node::wire::build_msg1;
use crate::peer::PeerConnection;
@@ -8,10 +15,15 @@ use crate::protocol::{Disconnect, DisconnectReason};
use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface};
use crate::{NodeAddr, PeerIdentity};
#[cfg(feature = "nostr-discovery")]
use std::collections::HashSet;
use std::thread;
use std::time::Duration;
use tracing::{debug, info, warn};
#[cfg(feature = "nostr-discovery")]
const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2;
impl Node {
/// Initiate connections to configured static peers.
///
@@ -107,86 +119,8 @@ impl Node {
return Ok(());
}
// Try addresses in priority order until one works
for addr in peer_config.addresses_by_priority() {
// For Ethernet addresses ("interface/mac"), find the transport
// instance matching the interface name and parse the MAC.
let (transport_id, remote_addr) = if addr.transport == "ethernet" {
match self.resolve_ethernet_addr(&addr.addr) {
Ok(result) => result,
Err(e) => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
error = %e,
"Failed to resolve Ethernet address"
);
continue;
}
}
} else if addr.transport == "ble" {
#[cfg(target_os = "linux")]
{
match self.resolve_ble_addr(&addr.addr) {
Ok(result) => result,
Err(e) => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
error = %e,
"Failed to resolve BLE address"
);
continue;
}
}
}
#[cfg(not(target_os = "linux"))]
{
debug!(
transport = %addr.transport,
"BLE transport not available on this platform"
);
continue;
}
} else {
// Find a transport matching this address type
let tid = match self.find_transport_for_type(&addr.transport) {
Some(id) => id,
None => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
"No operational transport for address type"
);
continue;
}
};
(tid, TransportAddr::from_string(&addr.addr))
};
match self
.initiate_connection(transport_id, remote_addr, Some(peer_identity))
.await
{
Ok(()) => return Ok(()),
Err(e @ NodeError::AccessDenied(_)) => return Err(e),
Err(e) => {
debug!(
npub = %peer_config.npub,
transport_id = %transport_id,
error = %e,
"Connection attempt failed, trying next address"
);
continue;
}
}
}
// No address worked
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
peer_config.npub
)))
self.try_peer_addresses(peer_config, peer_identity, true)
.await
}
/// Initiate a connection to a peer on a specific transport and address.
@@ -302,15 +236,12 @@ impl Node {
remote_addr: TransportAddr,
peer_identity: Option<PeerIdentity>,
) -> Result<(), NodeError> {
// Create connection in handshake phase
let current_time_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// Create connection in handshake phase. Anonymous discovery
// (no peer_identity) leaves identity to be learned from XX msg2.
let current_time_ms = Self::now_ms();
let mut connection = if let Some(identity) = peer_identity {
PeerConnection::outbound(link_id, identity, current_time_ms)
} else {
// Anonymous discovery connection — identity learned from XX msg2
PeerConnection::outbound_anonymous(link_id, current_time_ms)
};
@@ -491,6 +422,58 @@ impl Node {
}
}
#[cfg(feature = "nostr-discovery")]
pub(super) async fn poll_nostr_discovery(&mut self) {
let Some(bootstrap) = self.nostr_discovery.clone() else {
return;
};
if let Err(err) = self.refresh_overlay_advert(&bootstrap).await {
debug!(error = %err, "Failed to refresh local Nostr overlay advert");
}
for event in bootstrap.drain_events().await {
match event {
BootstrapEvent::Established { traversal } => {
let peer_npub = traversal.peer_npub.clone();
match self.adopt_established_traversal(traversal).await {
Ok(_) => {
info!(peer_npub = %peer_npub, "Adopted NAT traversal socket");
}
Err(err) => {
warn!(peer_npub = %peer_npub, error = %err, "Failed to adopt NAT traversal");
if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_npub) {
self.schedule_retry(*peer_identity.node_addr(), Self::now_ms());
}
}
}
}
BootstrapEvent::Failed {
peer_config,
reason,
} => {
warn!(npub = %peer_config.npub, error = %reason, "NAT traversal failed");
let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) {
Ok(identity) => identity,
Err(_) => continue,
};
if self
.try_peer_addresses(&peer_config, peer_identity, false)
.await
.is_ok()
{
continue;
}
self.schedule_retry(*peer_identity.node_addr(), Self::now_ms());
}
}
}
self.queue_open_discovery_retries(&bootstrap).await;
}
/// Poll pending transport connects and initiate handshakes for ready ones.
///
/// Called from the tick handler. For each pending connect, queries the
@@ -573,17 +556,14 @@ impl Node {
"Transport connect failed"
);
// Clean up link and schedule retry
// Clean up link and schedule retry. Anonymous discovery
// connections (no expected identity) don't retry —
// they'll be rediscovered via the shared-medium beacon.
self.remove_link(&pending.link_id);
self.links.remove(&pending.link_id);
if let Some(ref id) = pending.peer_identity {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
self.schedule_retry(*id.node_addr(), now_ms);
self.schedule_retry(*id.node_addr(), Self::now_ms());
}
// Anonymous connections don't retry — they'll be rediscovered
}
}
}
@@ -606,7 +586,7 @@ impl Node {
self.packet_tx = Some(packet_tx.clone());
self.packet_rx = Some(packet_rx);
// Initialize transports first (before TUN)
// Initialize transports first (before TUN, before Nostr discovery).
let transport_handles = self.create_transports(&packet_tx).await;
for mut handle in transport_handles {
@@ -632,6 +612,31 @@ impl Node {
info!(count = self.transports.len(), "Transports initialized");
}
#[cfg(feature = "nostr-discovery")]
if self.config.node.discovery.nostr.enabled {
match NostrDiscovery::start(&self.identity, self.config.node.discovery.nostr.clone())
.await
{
Ok(runtime) => {
if let Err(err) = self.refresh_overlay_advert(&runtime).await {
warn!(error = %err, "Failed to publish initial Nostr overlay advert");
}
self.nostr_discovery = Some(runtime);
info!("Nostr overlay discovery enabled");
}
Err(err) => {
warn!(error = %err, "Failed to start Nostr overlay discovery");
}
}
}
#[cfg(not(feature = "nostr-discovery"))]
if self.config.node.discovery.nostr.enabled {
warn!(
"Nostr overlay discovery configured but this build was compiled without the 'nostr-discovery' feature"
);
}
// Connect to static peers before TUN is active
// This allows handshake messages to be sent before we start accepting packets
self.initiate_peer_connections().await;
@@ -898,6 +903,14 @@ impl Node {
self.send_disconnect_to_all_peers(DisconnectReason::Shutdown)
.await;
// Stop Nostr overlay discovery background work and withdraw any advert.
#[cfg(feature = "nostr-discovery")]
if let Some(bootstrap) = self.nostr_discovery.take()
&& let Err(e) = bootstrap.shutdown().await
{
warn!(error = %e, "Failed to shutdown Nostr overlay discovery");
}
// Shutdown transports (they're packet producers)
let transport_ids: Vec<_> = self.transports.keys().cloned().collect();
for transport_id in transport_ids {
@@ -1005,6 +1018,500 @@ impl Node {
info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications");
}
fn static_peer_addresses(&self, peer_config: &PeerConfig) -> Vec<PeerAddress> {
peer_config
.addresses_by_priority()
.into_iter()
.cloned()
.collect()
}
#[cfg(feature = "nostr-discovery")]
async fn nostr_peer_fallback_addresses(
&self,
peer_config: &PeerConfig,
existing: &[PeerAddress],
) -> Vec<PeerAddress> {
if !self.config.node.discovery.nostr.enabled
|| !peer_config.via_nostr
|| self.config.node.discovery.nostr.policy
== crate::config::NostrDiscoveryPolicy::Disabled
{
return Vec::new();
}
let Some(bootstrap) = self.nostr_discovery.clone() else {
return Vec::new();
};
let endpoints = match bootstrap.advert_endpoints_for_peer(&peer_config.npub).await {
Ok(endpoints) => endpoints,
Err(err) => {
debug!(
npub = %peer_config.npub,
error = %err,
"Failed to resolve Nostr advert endpoints for configured peer"
);
return Vec::new();
}
};
let mut fallback = Vec::new();
let mut next_priority = existing
.iter()
.map(|addr| addr.priority)
.max()
.unwrap_or(100)
.saturating_add(1);
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, next_priority)
else {
continue;
};
if existing
.iter()
.any(|addr| addr.transport == candidate.transport && addr.addr == candidate.addr)
|| fallback.iter().any(|addr: &PeerAddress| {
addr.transport == candidate.transport && addr.addr == candidate.addr
})
{
continue;
}
fallback.push(candidate);
next_priority = next_priority.saturating_add(1);
}
fallback
}
#[cfg(feature = "nostr-discovery")]
fn overlay_endpoint_to_peer_address(
endpoint: &OverlayEndpointAdvert,
priority: u8,
) -> Option<PeerAddress> {
let transport = match endpoint.transport {
OverlayTransportKind::Udp => "udp",
OverlayTransportKind::Tcp => "tcp",
OverlayTransportKind::Tor => "tor",
};
Some(PeerAddress::with_priority(
transport,
endpoint.addr.clone(),
priority,
))
}
async fn attempt_peer_address_list(
&mut self,
peer_config: &PeerConfig,
peer_identity: PeerIdentity,
allow_bootstrap_nat: bool,
addresses: &[PeerAddress],
) -> Result<(), NodeError> {
for addr in addresses {
if addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat") {
if !allow_bootstrap_nat {
continue;
}
#[cfg(not(feature = "nostr-discovery"))]
{
debug!(npub = %peer_config.npub, "Skipping udp:nat address because this build does not include the nostr-discovery feature");
continue;
}
#[cfg(feature = "nostr-discovery")]
{
let Some(bootstrap) = self.nostr_discovery.clone() else {
debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address");
continue;
};
bootstrap.request_connect(peer_config.clone()).await;
info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt");
return Ok(());
}
}
let (transport_id, remote_addr) = if addr.transport == "ethernet" {
match self.resolve_ethernet_addr(&addr.addr) {
Ok(result) => result,
Err(e) => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
error = %e,
"Failed to resolve Ethernet address"
);
continue;
}
}
} else if addr.transport == "ble" {
#[cfg(bluer_available)]
{
match self.resolve_ble_addr(&addr.addr) {
Ok(result) => result,
Err(e) => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
error = %e,
"Failed to resolve BLE address"
);
continue;
}
}
}
#[cfg(not(bluer_available))]
{
debug!(transport = %addr.transport, "BLE transport not available on this build");
continue;
}
} else {
let tid = match self.find_transport_for_type(&addr.transport) {
Some(id) => id,
None => {
debug!(
transport = %addr.transport,
addr = %addr.addr,
"No operational transport for address type"
);
continue;
}
};
(tid, TransportAddr::from_string(&addr.addr))
};
match self
.initiate_connection(transport_id, remote_addr, Some(peer_identity))
.await
{
Ok(()) => return Ok(()),
Err(e @ NodeError::AccessDenied(_)) => return Err(e),
Err(e) => {
debug!(
npub = %peer_config.npub,
transport_id = %transport_id,
error = %e,
"Connection attempt failed, trying next address"
);
}
}
}
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
peer_config.npub
)))
}
#[cfg(feature = "nostr-discovery")]
async fn queue_open_discovery_retries(&mut self, bootstrap: &std::sync::Arc<NostrDiscovery>) {
if !self.config.node.discovery.nostr.enabled
|| self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open
{
return;
}
let configured_npubs = self
.config
.peers()
.iter()
.map(|peer| peer.npub.clone())
.collect::<HashSet<_>>();
let now_ms = Self::now_ms();
let mut enqueue_budget = self.open_discovery_enqueue_budget(&configured_npubs);
if enqueue_budget == 0 {
return;
}
for (npub, endpoints) in bootstrap.cached_open_discovery_candidates(64).await {
if enqueue_budget == 0 {
break;
}
if configured_npubs.contains(&npub) {
continue;
}
let peer_identity = match PeerIdentity::from_npub(&npub) {
Ok(identity) => identity,
Err(_) => continue,
};
let node_addr = *peer_identity.node_addr();
if node_addr == *self.identity.node_addr() || self.peers.contains_key(&node_addr) {
continue;
}
if self.retry_pending.contains_key(&node_addr) {
continue;
}
let connecting = self.connections.values().any(|conn| {
conn.expected_identity()
.map(|id| id.node_addr() == &node_addr)
.unwrap_or(false)
});
if connecting {
continue;
}
let mut addresses = Vec::new();
let mut priority = 120u8;
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, priority)
else {
continue;
};
if addresses.iter().any(|existing: &PeerAddress| {
existing.transport == candidate.transport && existing.addr == candidate.addr
}) {
continue;
}
addresses.push(candidate);
priority = priority.saturating_add(1);
}
if addresses.is_empty() {
continue;
}
self.peer_aliases
.entry(node_addr)
.or_insert_with(|| peer_identity.short_npub());
self.register_identity(node_addr, peer_identity.pubkey_full());
let mut state = super::retry::RetryState::new(PeerConfig {
npub: npub.clone(),
alias: None,
addresses,
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
});
state.reconnect = false;
state.retry_after_ms = now_ms;
state.expires_at_ms = Some(self.open_discovery_retry_expires_at_ms(now_ms));
self.retry_pending.insert(node_addr, state);
enqueue_budget = enqueue_budget.saturating_sub(1);
}
}
#[cfg(feature = "nostr-discovery")]
fn available_outbound_slots(&self) -> usize {
let connection_used = self
.connections
.len()
.saturating_add(self.pending_connects.len());
let connection_slots = if self.max_connections == 0 {
usize::MAX
} else {
self.max_connections.saturating_sub(connection_used)
};
let peer_slots = if self.max_peers == 0 {
usize::MAX
} else {
self.max_peers.saturating_sub(self.peers.len())
};
connection_slots.min(peer_slots)
}
#[cfg(feature = "nostr-discovery")]
fn open_discovery_enqueue_budget(&self, configured_npubs: &HashSet<String>) -> usize {
let current_open_discovery_pending = self
.retry_pending
.values()
.filter(|state| !configured_npubs.contains(&state.peer_config.npub))
.count();
let cap_remaining = self
.config
.node
.discovery
.nostr
.open_discovery_max_pending
.saturating_sub(current_open_discovery_pending);
cap_remaining.min(self.available_outbound_slots())
}
#[cfg(feature = "nostr-discovery")]
fn open_discovery_retry_expires_at_ms(&self, now_ms: u64) -> u64 {
now_ms.saturating_add(
self.config
.node
.discovery
.nostr
.advert_ttl_secs
.saturating_mul(1000)
.saturating_mul(OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER),
)
}
#[cfg(feature = "nostr-discovery")]
fn build_overlay_advert(&self) -> Option<OverlayAdvert> {
if !self.config.node.discovery.nostr.enabled {
return None;
}
let mut endpoints = Vec::new();
let mut has_udp_nat = false;
for handle in self.transports.values() {
if !handle.is_operational() {
continue;
}
match handle.transport_type().name {
"udp" => {
let Some(cfg) = self.lookup_udp_config(handle.name()) else {
continue;
};
if !cfg.advertise_on_nostr() {
continue;
}
if cfg.is_public() {
if let Some(addr) = handle.local_addr()
&& !addr.ip().is_unspecified()
{
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: addr.to_string(),
});
}
} else {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: "nat".to_string(),
});
has_udp_nat = true;
}
}
"tcp" => {
let Some(cfg) = self.lookup_tcp_config(handle.name()) else {
continue;
};
if !cfg.advertise_on_nostr() {
continue;
}
if let Some(addr) = handle.local_addr()
&& !addr.ip().is_unspecified()
{
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Tcp,
addr: addr.to_string(),
});
}
}
"tor" => {
let Some(cfg) = self.lookup_tor_config(handle.name()) else {
continue;
};
if !cfg.advertise_on_nostr() {
continue;
}
if let Some(addr) = handle.onion_address() {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Tor,
addr: addr.to_string(),
});
}
}
_ => {}
}
}
if endpoints.is_empty() {
return None;
}
Some(OverlayAdvert {
identifier: ADVERT_IDENTIFIER.to_string(),
version: ADVERT_VERSION,
endpoints,
signal_relays: has_udp_nat.then(|| self.config.node.discovery.nostr.dm_relays.clone()),
stun_servers: has_udp_nat
.then(|| self.config.node.discovery.nostr.stun_servers.clone()),
})
}
#[cfg(feature = "nostr-discovery")]
async fn refresh_overlay_advert(
&self,
bootstrap: &std::sync::Arc<NostrDiscovery>,
) -> Result<(), crate::discovery::nostr::BootstrapError> {
let advert = self.build_overlay_advert();
bootstrap.update_local_advert(advert).await
}
#[cfg(feature = "nostr-discovery")]
fn lookup_udp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::UdpConfig> {
match (&self.config.transports.udp, transport_name) {
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
_ => None,
}
}
#[cfg(feature = "nostr-discovery")]
fn lookup_tcp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TcpConfig> {
match (&self.config.transports.tcp, transport_name) {
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
_ => None,
}
}
#[cfg(feature = "nostr-discovery")]
fn lookup_tor_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TorConfig> {
match (&self.config.transports.tor, transport_name) {
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
_ => None,
}
}
pub(in crate::node) async fn try_peer_addresses(
&mut self,
peer_config: &PeerConfig,
peer_identity: PeerIdentity,
allow_bootstrap_nat: bool,
) -> Result<(), NodeError> {
// Static-first dialing: avoid delaying configured address attempts on
// advert fetch/network latency.
let static_addresses = self.static_peer_addresses(peer_config);
if self
.attempt_peer_address_list(
peer_config,
peer_identity,
allow_bootstrap_nat,
&static_addresses,
)
.await
.is_ok()
{
return Ok(());
}
#[cfg(feature = "nostr-discovery")]
{
let fallback = self
.nostr_peer_fallback_addresses(peer_config, &static_addresses)
.await;
if !fallback.is_empty()
&& self
.attempt_peer_address_list(
peer_config,
peer_identity,
allow_bootstrap_nat,
&fallback,
)
.await
.is_ok()
{
return Ok(());
}
}
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
peer_config.npub
)))
}
// === Control API methods ===
/// Connect to a peer via the control API.
@@ -1018,12 +1525,13 @@ impl Node {
address: &str,
transport: &str,
) -> Result<serde_json::Value, String> {
let peer_config = crate::config::PeerConfig {
let peer_config = PeerConfig {
npub: npub.to_string(),
alias: None,
addresses: vec![crate::config::PeerAddress::new(transport, address)],
connect_policy: crate::config::ConnectPolicy::Manual,
addresses: vec![PeerAddress::new(transport, address)],
connect_policy: ConnectPolicy::Manual,
auto_reconnect: false,
via_nostr: false,
};
// Pre-seed identity cache (same as initiate_peer_connections does)
@@ -1076,4 +1584,91 @@ impl Node {
"disconnected": true,
}))
}
/// Adopt an already-established UDP traversal and start the normal FIPS
/// Noise handshake over it.
///
/// This is intended for integration with an external rendezvous runtime
/// that has already completed relay signaling, STUN observation, and UDP
/// hole punching. After handoff, the adopted socket is owned by FIPS.
pub async fn adopt_established_traversal(
&mut self,
traversal: EstablishedTraversal,
) -> Result<BootstrapHandoffResult, NodeError> {
debug!(
peer_npub = %traversal.peer_npub,
session_id = %traversal.session_id,
remote_addr = %traversal.remote_addr,
"adopting established traversal socket"
);
if !self.state.is_operational() {
return Err(NodeError::NotStarted);
}
let packet_tx = self.packet_tx.clone().ok_or(NodeError::NotStarted)?;
let peer_identity = PeerIdentity::from_npub(&traversal.peer_npub).map_err(|e| {
NodeError::InvalidPeerNpub {
npub: traversal.peer_npub.clone(),
reason: e.to_string(),
}
})?;
let peer_node_addr = *peer_identity.node_addr();
self.peer_aliases
.insert(peer_node_addr, peer_identity.short_npub());
self.register_identity(peer_node_addr, peer_identity.pubkey_full());
let transport_id = self.allocate_transport_id();
let mut transport = crate::transport::udp::UdpTransport::new(
transport_id,
traversal.transport_name.clone(),
traversal.transport_config.clone().unwrap_or_default(),
packet_tx,
);
transport
.adopt_socket_async(traversal.socket)
.await
.map_err(|e| NodeError::BootstrapHandoff(e.to_string()))?;
let local_addr = transport.local_addr().ok_or_else(|| {
NodeError::BootstrapHandoff("adopted UDP transport has no local address".into())
})?;
self.transports.insert(
transport_id,
crate::transport::TransportHandle::Udp(transport),
);
self.bootstrap_transports.insert(transport_id);
let remote_addr = TransportAddr::from_string(&traversal.remote_addr.to_string());
if let Err(err) = self
.initiate_connection(transport_id, remote_addr.clone(), Some(peer_identity))
.await
{
self.bootstrap_transports.remove(&transport_id);
if let Some(mut handle) = self.transports.remove(&transport_id) {
let _ = handle.stop().await;
}
return Err(err);
}
info!(
peer = %self.peer_display_name(&peer_node_addr),
transport_id = %transport_id,
local_addr = %local_addr,
remote_addr = %traversal.remote_addr,
session_id = %traversal.session_id,
"adopted NAT traversal socket; handshake initiated"
);
Ok(BootstrapHandoffResult {
transport_id,
local_addr,
remote_addr: traversal.remote_addr,
peer_node_addr,
session_id: traversal.session_id,
})
}
}
+66 -13
View File
@@ -47,7 +47,7 @@ use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
use crate::utils::index::IndexAllocator;
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity};
use rand::Rng;
use std::collections::{HashMap, VecDeque};
use std::collections::{HashMap, HashSet, VecDeque};
use std::fmt;
use std::sync::Arc;
use std::thread::JoinHandle;
@@ -137,6 +137,9 @@ pub enum NodeError {
#[error("transport error: {0}")]
TransportError(String),
#[error("bootstrap handoff failed: {0}")]
BootstrapHandoff(String),
}
/// Node operational state.
@@ -262,7 +265,7 @@ struct PendingConnect {
///
/// The `addr_to_link` map enables dispatching incoming packets to the right
/// connection before authentication completes.
// Discovery lookup constants moved to config: node.discovery.timeout_secs, node.discovery.ttl
// Discovery lookup constants moved to config: node.discovery.attempt_timeouts_secs, node.discovery.ttl
pub struct Node {
// === Identity ===
/// This node's cryptographic identity.
@@ -344,7 +347,6 @@ pub struct Node {
/// Packets queued while waiting for session establishment.
/// Keyed by destination NodeAddr, bounded per-dest and total.
pending_tun_packets: HashMap<NodeAddr, VecDeque<Vec<u8>>>,
// === Pending Discovery Lookups ===
/// Tracks in-flight discovery lookups. Maps target NodeAddr to the
/// initiation timestamp (Unix ms). Prevents duplicate flood queries.
@@ -437,6 +439,12 @@ pub struct Node {
/// are exhausted.
retry_pending: HashMap<NodeAddr, retry::RetryState>,
/// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers.
#[cfg(feature = "nostr-discovery")]
nostr_discovery: Option<Arc<crate::discovery::nostr::NostrDiscovery>>,
/// Per-peer UDP transports adopted from NAT traversal handoff.
bootstrap_transports: HashSet<TransportId>,
// === Periodic Parent Re-evaluation ===
/// Timestamp of last periodic parent re-evaluation (for pacing).
last_parent_reeval: Option<std::time::Instant>,
@@ -475,6 +483,7 @@ pub struct Node {
impl Node {
/// Create a new node from configuration.
pub fn new(config: Config) -> Result<Self, NodeError> {
config.validate()?;
let identity = config.create_identity()?;
let node_addr = *identity.node_addr();
let is_leaf_only = config.is_leaf_only();
@@ -599,6 +608,9 @@ impl Node {
),
pending_connects: Vec::new(),
retry_pending: HashMap::new(),
#[cfg(feature = "nostr-discovery")]
nostr_discovery: None,
bootstrap_transports: HashSet::new(),
last_parent_reeval: None,
last_congestion_log: None,
estimated_mesh_size: None,
@@ -611,7 +623,11 @@ impl Node {
}
/// Create a node with a specific identity.
pub fn with_identity(identity: Identity, config: Config) -> Self {
///
/// This constructor validates cross-field config invariants before
/// constructing the node, same as [`Node::new`].
pub fn with_identity(identity: Identity, config: Config) -> Result<Self, NodeError> {
config.validate()?;
let node_addr = *identity.node_addr();
let mut startup_epoch = [0u8; 8];
@@ -667,7 +683,7 @@ impl Node {
std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH),
);
Self {
Ok(Self {
identity,
startup_epoch,
started_at: std::time::Instant::now(),
@@ -722,6 +738,9 @@ impl Node {
discovery_forward_limiter: DiscoveryForwardRateLimiter::new(),
pending_connects: Vec::new(),
retry_pending: HashMap::new(),
#[cfg(feature = "nostr-discovery")]
nostr_discovery: None,
bootstrap_transports: HashSet::new(),
last_parent_reeval: None,
last_congestion_log: None,
estimated_mesh_size: None,
@@ -730,7 +749,7 @@ impl Node {
peer_aliases: HashMap::new(),
peer_acl,
host_map,
}
})
}
/// Create a leaf-only node (simplified state).
@@ -813,7 +832,7 @@ impl Node {
}
// Create BLE transport instances
#[cfg(target_os = "linux")]
#[cfg(bluer_available)]
{
let ble_instances: Vec<_> = self
.config
@@ -823,7 +842,7 @@ impl Node {
.map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
.collect();
#[cfg(all(feature = "ble", not(test)))]
#[cfg(all(bluer_available, not(test)))]
for (name, ble_config) in ble_instances {
let transport_id = self.allocate_transport_id();
let adapter = ble_config.adapter().to_string();
@@ -845,12 +864,10 @@ impl Node {
}
}
#[cfg(any(not(feature = "ble"), test))]
#[cfg(any(not(bluer_available), test))]
if !ble_instances.is_empty() {
#[cfg(not(test))]
tracing::warn!(
"BLE transport configured but 'ble' feature not enabled at compile time"
);
tracing::warn!("BLE transport configured but this build lacks BlueZ support");
}
}
@@ -920,7 +937,7 @@ impl Node {
/// Resolve a BLE address string (`"adapter/AA:BB:CC:DD:EE:FF"`) to a
/// (TransportId, TransportAddr) pair by finding the BLE transport
/// instance matching the adapter name.
#[cfg(target_os = "linux")]
#[cfg(bluer_available)]
fn resolve_ble_addr(&self, addr_str: &str) -> Result<(TransportId, TransportAddr), NodeError> {
let ta = TransportAddr::from_string(addr_str);
let adapter = crate::transport::ble::addr::adapter_from_addr(&ta).ok_or_else(|| {
@@ -1416,6 +1433,42 @@ impl Node {
}
}
pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
if !self.bootstrap_transports.contains(&transport_id) {
return;
}
let transport_in_use = self
.links
.values()
.any(|link| link.transport_id() == transport_id)
|| self
.connections
.values()
.any(|conn| conn.transport_id() == Some(transport_id))
|| self
.peers
.values()
.any(|peer| peer.transport_id() == Some(transport_id))
|| self
.pending_connects
.iter()
.any(|pending| pending.transport_id == transport_id);
if transport_in_use {
return;
}
tracing::debug!(
transport_id = %transport_id,
"bootstrap transport has no remaining references; dropping"
);
self.bootstrap_transports.remove(&transport_id);
self.transport_drops.remove(&transport_id);
self.transports.remove(&transport_id);
}
/// Iterate over all links.
pub fn links(&self) -> impl Iterator<Item = &Link> {
self.links.values()
+31
View File
@@ -25,6 +25,12 @@ pub struct RetryState {
/// Whether this is an auto-reconnect (unlimited retries, ignores max_retries).
pub reconnect: bool,
/// Optional absolute expiry for this retry entry (Unix ms).
///
/// When set, retries are dropped after this point even if reconnect logic
/// would otherwise continue.
pub expires_at_ms: Option<u64>,
}
impl RetryState {
@@ -35,6 +41,7 @@ impl RetryState {
retry_count: 0,
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
}
}
@@ -203,6 +210,27 @@ impl Node {
return;
}
let expired: Vec<NodeAddr> = self
.retry_pending
.iter()
.filter_map(|(addr, state)| {
state
.expires_at_ms
.filter(|expires_at_ms| now_ms >= *expires_at_ms)
.map(|_| *addr)
})
.collect();
for node_addr in expired {
self.retry_pending.remove(&node_addr);
info!(
peer = %self.peer_display_name(&node_addr),
"Retry window expired, dropping pending retry state"
);
}
if self.retry_pending.is_empty() {
return;
}
// Collect retries that are due
let due: Vec<NodeAddr> = self
.retry_pending
@@ -277,6 +305,7 @@ mod tests {
retry_count: 0,
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
// base = 5000ms
assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 5000); // 5s * 2^0
@@ -313,6 +342,7 @@ mod tests {
retry_count: 20, // 2^20 * 5000 would be huge
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
assert_eq!(
state.backoff_ms(5000, TEST_MAX_BACKOFF_MS),
@@ -327,6 +357,7 @@ mod tests {
retry_count: 3,
retry_after_ms: 0,
reconnect: false,
expires_at_ms: None,
};
assert_eq!(state.backoff_ms(0, TEST_MAX_BACKOFF_MS), 0);
}
+247
View File
@@ -0,0 +1,247 @@
//! Integration tests for bootstrap handoff into the FIPS node.
use super::*;
use crate::EstablishedTraversal;
use crate::config::UdpConfig;
use crate::node::wire::{PHASE_MSG1, PHASE_MSG2};
use crate::transport::udp::UdpTransport;
use crate::utils::index::IndexAllocator;
use tokio::time::{Duration, timeout, timeout_at};
#[ignore = "needs XX-handshake adaptation: peer promotion now happens at msg3, not msg1; test runs only two rx_loop iterations"]
#[tokio::test]
async fn test_adopted_udp_traversal_completes_handshake() {
let mut node_a = make_node();
let mut node_b = make_node();
let transport_id_b = TransportId::new(1);
let udp_config = UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
mtu: Some(1280),
..Default::default()
};
let (packet_tx_a, packet_rx_a) = packet_channel(64);
let (packet_tx_b, packet_rx_b) = packet_channel(64);
node_a.packet_tx = Some(packet_tx_a.clone());
node_a.packet_rx = Some(packet_rx_a);
node_a.state = NodeState::Running;
let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b.clone());
transport_b.start_async().await.unwrap();
let addr_b = transport_b.local_addr().unwrap();
node_b.packet_tx = Some(packet_tx_b.clone());
node_b.packet_rx = Some(packet_rx_b);
node_b.state = NodeState::Running;
node_b
.transports
.insert(transport_id_b, TransportHandle::Udp(transport_b));
let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let handoff = EstablishedTraversal::new("sess-1", node_b.npub(), addr_b, adopted_socket)
.with_transport_name("nostr-punched");
let result = node_a.adopt_established_traversal(handoff).await.unwrap();
assert_eq!(result.remote_addr, addr_b);
assert!(node_a.get_transport(&result.transport_id).is_some());
tokio::select! {
result = node_b.run_rx_loop() => {
panic!("node_b rx loop exited unexpectedly: {:?}", result);
}
_ = tokio::time::sleep(Duration::from_millis(500)) => {}
}
tokio::select! {
result = node_a.run_rx_loop() => {
panic!("node_a rx loop exited unexpectedly: {:?}", result);
}
_ = tokio::time::sleep(Duration::from_millis(500)) => {}
}
let peer_a_node_addr =
*PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
let peer_b_node_addr =
*PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full()).node_addr();
assert_eq!(
node_a.peer_count(),
1,
"node_a should promote node_b after handoff"
);
assert_eq!(
node_b.peer_count(),
1,
"node_b should promote node_a after receiving msg1"
);
assert!(node_a.get_peer(&peer_b_node_addr).unwrap().has_session());
assert!(node_b.get_peer(&peer_a_node_addr).unwrap().has_session());
for (_, transport) in node_a.transports.iter_mut() {
transport.stop().await.ok();
}
for (_, transport) in node_b.transports.iter_mut() {
transport.stop().await.ok();
}
}
#[tokio::test]
async fn test_failed_adopted_traversal_cleans_up_transport() {
let mut node = make_node();
let (packet_tx, packet_rx) = packet_channel(64);
node.packet_tx = Some(packet_tx);
node.packet_rx = Some(packet_rx);
node.state = NodeState::Running;
node.index_allocator = IndexAllocator::with_max_attempts(0);
let peer = make_node();
let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let handoff = EstablishedTraversal::new(
"sess-fail",
peer.npub(),
"127.0.0.1:9".parse().unwrap(),
adopted_socket,
)
.with_transport_name("nostr-punched");
let result = node.adopt_established_traversal(handoff).await;
assert!(
result.is_err(),
"handoff should fail when handshake setup cannot allocate a session index"
);
assert!(
node.transports.is_empty(),
"failed handoff should remove the adopted transport"
);
}
#[ignore = "needs XX-handshake adaptation: peer promotion now happens at msg3, not msg1; test runs only two rx_loop iterations"]
#[tokio::test]
async fn test_third_peer_can_handshake_via_adopted_transport_socket() {
let mut node_a = make_node(); // Existing traversal peer (Alice)
let mut node_b = make_node(); // Node with adopted socket (Bob)
let mut node_c = make_node(); // New peer onboarding via Bob socket (Colin)
let transport_id_a = TransportId::new(1);
let transport_id_c = TransportId::new(1);
let udp_config = UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
mtu: Some(1280),
..Default::default()
};
let (packet_tx_a, packet_rx_a) = packet_channel(64);
let (packet_tx_b, packet_rx_b) = packet_channel(64);
let (packet_tx_c, packet_rx_c) = packet_channel(64);
node_a.packet_tx = Some(packet_tx_a.clone());
node_a.packet_rx = Some(packet_rx_a);
node_a.state = NodeState::Running;
node_b.packet_tx = Some(packet_tx_b.clone());
node_b.packet_rx = Some(packet_rx_b);
node_b.state = NodeState::Running;
node_c.packet_tx = Some(packet_tx_c.clone());
node_c.packet_rx = Some(packet_rx_c);
node_c.state = NodeState::Running;
let mut transport_a = UdpTransport::new(transport_id_a, None, udp_config.clone(), packet_tx_a);
transport_a.start_async().await.unwrap();
let addr_a = transport_a.local_addr().unwrap();
node_a
.transports
.insert(transport_id_a, TransportHandle::Udp(transport_a));
// Bob adopts a traversal socket already "established" to Alice.
let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let handoff = EstablishedTraversal::new("sess-existing", node_a.npub(), addr_a, adopted_socket)
.with_transport_name("nostr-nat");
let handoff_result = node_b.adopt_established_traversal(handoff).await.unwrap();
// Drive Alice/Bob handshake manually (msg1 -> msg2).
let mut rx_a = node_a.packet_rx.take().expect("node_a packet_rx");
let mut rx_b = node_b.packet_rx.take().expect("node_b packet_rx");
let pkt_at_a = timeout(Duration::from_secs(1), rx_a.recv())
.await
.expect("timeout waiting for Bob->Alice msg1")
.expect("node_a channel closed");
assert_eq!(pkt_at_a.data[0] & 0x0f, PHASE_MSG1);
node_a.handle_msg1(pkt_at_a).await;
let pkt_at_b = timeout(Duration::from_secs(1), rx_b.recv())
.await
.expect("timeout waiting for Alice->Bob msg2")
.expect("node_b channel closed");
assert_eq!(pkt_at_b.data[0] & 0x0f, PHASE_MSG2);
node_b.handle_msg2(pkt_at_b).await;
let node_a_addr = *PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
assert!(
node_b.get_peer(&node_a_addr).is_some(),
"node_b should first be connected to node_a via adopted transport"
);
// Start Colin UDP transport and connect to Bob's adopted socket address.
let mut transport_c = UdpTransport::new(transport_id_c, None, udp_config, packet_tx_c);
transport_c.start_async().await.unwrap();
let addr_c = transport_c.local_addr().unwrap();
node_c
.transports
.insert(transport_id_c, TransportHandle::Udp(transport_c));
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let adopted_addr = TransportAddr::from_string(&handoff_result.local_addr.to_string());
node_c
.initiate_connection(transport_id_c, adopted_addr, Some(peer_b_identity))
.await
.unwrap();
// Drive Bob/Colin handshake manually (msg1 -> msg2).
let mut rx_c = node_c.packet_rx.take().expect("node_c packet_rx");
let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
let pkt_at_b = loop {
let pkt = timeout_at(deadline, rx_b.recv())
.await
.expect("timeout waiting for Colin->Bob msg1")
.expect("node_b channel closed");
if pkt.remote_addr.as_str() == Some(&addr_c.to_string())
&& pkt.data.first().map(|b| b & 0x0f) == Some(PHASE_MSG1)
{
break pkt;
}
};
node_b.handle_msg1(pkt_at_b).await;
let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
let pkt_at_c = loop {
let pkt = timeout_at(deadline, rx_c.recv())
.await
.expect("timeout waiting for Bob->Colin msg2")
.expect("node_c channel closed");
if pkt.data.first().map(|b| b & 0x0f) == Some(PHASE_MSG2) {
break pkt;
}
};
node_c.handle_msg2(pkt_at_c).await;
let node_c_addr = *PeerIdentity::from_pubkey_full(node_c.identity.pubkey_full()).node_addr();
assert!(
node_b.get_peer(&node_c_addr).is_some(),
"node_b should promote node_c when node_c handshakes via adopted socket"
);
for (_, transport) in node_a.transports.iter_mut() {
transport.stop().await.ok();
}
for (_, transport) in node_b.transports.iter_mut() {
transport.stop().await.ok();
}
for (_, transport) in node_c.transports.iter_mut() {
transport.stop().await.ok();
}
}
+2 -1
View File
@@ -9,9 +9,10 @@ mod acl;
mod ble;
mod bloom;
mod bloom_poison;
mod bootstrap;
mod disconnect;
mod discovery;
#[cfg(unix)]
#[cfg(target_os = "linux")]
mod ethernet;
mod forwarding;
mod handshake;
+89 -1
View File
@@ -1,5 +1,7 @@
use super::*;
use crate::peer::PromotionResult;
use crate::transport::udp::UdpTransport;
use crate::transport::{TransportHandle, packet_channel};
#[test]
fn test_node_creation() {
@@ -18,11 +20,26 @@ fn test_node_with_identity() {
let expected_node_addr = *identity.node_addr();
let config = Config::new();
let node = Node::with_identity(identity, config);
let node = Node::with_identity(identity, config).unwrap();
assert_eq!(node.node_addr(), &expected_node_addr);
}
#[test]
fn test_node_with_identity_validates_config() {
let identity = Identity::generate();
let mut config = Config::new();
config.node.discovery.nostr.enabled = false;
config.peers = vec![crate::config::PeerConfig {
npub: "npub1peer".to_string(),
via_nostr: true,
..Default::default()
}];
let err = Node::with_identity(identity, config).expect_err("expected config validation error");
assert!(matches!(err, NodeError::Config(_)));
}
#[test]
fn test_node_leaf_only() {
let config = Config::new();
@@ -32,6 +49,52 @@ fn test_node_leaf_only() {
assert!(node.bloom_state().is_leaf_only());
}
#[tokio::test]
async fn test_nat_bootstrap_failure_falls_back_to_direct_udp_address() {
let peer_identity = Identity::generate();
let mut node = make_node();
let (packet_tx, packet_rx) = packet_channel(64);
node.packet_tx = Some(packet_tx.clone());
node.packet_rx = Some(packet_rx);
let transport_id = TransportId::new(1);
let mut udp = UdpTransport::new(
transport_id,
Some("main".to_string()),
crate::config::UdpConfig {
bind_addr: Some("127.0.0.1:0".to_string()),
..Default::default()
},
packet_tx,
);
udp.start_async().await.unwrap();
node.transports
.insert(transport_id, TransportHandle::Udp(udp));
let peer_config = crate::config::PeerConfig {
npub: peer_identity.npub(),
alias: None,
addresses: vec![
crate::config::PeerAddress::with_priority("udp", "nat", 1),
crate::config::PeerAddress::with_priority("udp", "127.0.0.1:9", 2),
],
connect_policy: crate::config::ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
};
let peer_identity = PeerIdentity::from_npub(&peer_config.npub).unwrap();
node.try_peer_addresses(&peer_config, peer_identity, false)
.await
.unwrap();
assert_eq!(node.connection_count(), 1);
for transport in node.transports.values_mut() {
transport.stop().await.ok();
}
}
#[tokio::test]
async fn test_node_state_transitions() {
let mut node = make_node();
@@ -719,6 +782,31 @@ fn test_schedule_retry_skips_connected_peer() {
);
}
#[tokio::test]
async fn test_process_pending_retries_drops_expired_entries() {
let mut node = make_node();
let peer_identity = Identity::generate();
let peer_npub = peer_identity.npub();
let peer_node_addr = *PeerIdentity::from_npub(&peer_npub).unwrap().node_addr();
let mut state = super::super::retry::RetryState::new(crate::config::PeerConfig::new(
peer_npub,
"udp",
"127.0.0.1:9",
));
state.retry_after_ms = 0;
state.expires_at_ms = Some(1_000);
state.reconnect = true;
node.retry_pending.insert(peer_node_addr, state);
node.process_pending_retries(1_000).await;
assert!(
!node.retry_pending.contains_key(&peer_node_addr),
"expired retry entries should be dropped before retry processing"
);
}
/// Test that schedule_reconnect preserves accumulated backoff across link-dead cycles.
///
/// Regression test for issue #5: previously `schedule_reconnect` always created a
+2 -2
View File
@@ -55,10 +55,10 @@ impl BleAddr {
}
// ============================================================================
// bluer type conversions (behind ble feature)
// bluer type conversions (glibc-linux only; see build.rs bluer_available)
// ============================================================================
#[cfg(feature = "ble")]
#[cfg(bluer_available)]
impl BleAddr {
/// Construct from a bluer `Address` and adapter name.
pub fn from_bluer(addr: bluer::Address, adapter: &str) -> Self {
+3 -3
View File
@@ -1,7 +1,7 @@
//! BLE I/O abstraction layer.
//!
//! Defines the `BleIo` trait that separates transport logic from the
//! BlueZ/bluer stack. `BluerIo` (behind `cfg(feature = "ble")`) provides
//! BlueZ/bluer stack. `BluerIo` (behind `cfg(bluer_available)`) provides
//! the real implementation; `MockBleIo` provides an in-memory test double.
use crate::transport::TransportError;
@@ -109,7 +109,7 @@ pub trait BleIo: Send + Sync + 'static {
// BluerIo — Production BLE I/O via BlueZ D-Bus
// ============================================================================
#[cfg(feature = "ble")]
#[cfg(bluer_available)]
mod bluer_impl {
use super::*;
use crate::transport::TransportError;
@@ -486,7 +486,7 @@ mod bluer_impl {
}
}
#[cfg(feature = "ble")]
#[cfg(bluer_available)]
pub use bluer_impl::{BluerAcceptor, BluerIo, BluerScanner, BluerStream, FIPS_SERVICE_UUID};
// ============================================================================
+5 -5
View File
@@ -9,7 +9,7 @@
//!
//! Transport logic (pool, discovery, lifecycle) is separated from the
//! BlueZ/bluer stack via the `BleIo` trait. `BluerIo` provides the real
//! implementation (behind `cfg(feature = "ble")`); `MockBleIo` provides
//! implementation (behind `cfg(bluer_available)`); `MockBleIo` provides
//! an in-memory test double for CI without hardware.
//!
//! ## Connection Pool
@@ -48,12 +48,12 @@ pub const DEFAULT_PSM: u16 = 0x0085;
/// Concrete BLE transport type for use in TransportHandle.
///
/// Production builds with the `ble` feature use `BluerIo` (real BlueZ stack).
/// Test builds and builds without `ble` use `MockBleIo`.
#[cfg(all(feature = "ble", not(test)))]
/// Production builds on glibc-linux use `BluerIo` (real BlueZ stack).
/// Test builds, musl-linux, and non-Linux platforms use `MockBleIo`.
#[cfg(all(bluer_available, not(test)))]
pub type DefaultBleTransport = BleTransport<io::BluerIo>;
#[cfg(any(not(feature = "ble"), test))]
#[cfg(any(not(bluer_available), test))]
pub type DefaultBleTransport = BleTransport<io::MockBleIo>;
// ============================================================================
+82
View File
@@ -198,6 +198,65 @@ impl UdpTransport {
Ok(())
}
/// Start the transport using an already-bound UDP socket.
///
/// This preserves an existing NAT mapping established by another
/// subsystem, such as STUN or UDP hole punching.
pub async fn adopt_socket_async(
&mut self,
socket: std::net::UdpSocket,
) -> Result<(), TransportError> {
if !self.state.can_start() {
return Err(TransportError::AlreadyStarted);
}
self.state = TransportState::Starting;
let raw_socket = UdpRawSocket::adopt(
socket,
self.config.recv_buf_size(),
self.config.send_buf_size(),
)?;
let actual_recv = raw_socket.recv_buffer_size()?;
let actual_send = raw_socket.send_buffer_size()?;
self.local_addr = Some(raw_socket.local_addr());
let async_socket = raw_socket.into_async()?;
self.socket = Some(async_socket.clone());
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
let mtu = self.config.mtu();
let stats = self.stats.clone();
let recv_task = tokio::spawn(async move {
udp_receive_loop(async_socket, transport_id, packet_tx, mtu, stats).await;
});
self.recv_task = Some(recv_task);
self.state = TransportState::Up;
if let Some(ref name) = self.name {
info!(
name = %name,
local_addr = %self.local_addr.unwrap(),
recv_buf = actual_recv,
send_buf = actual_send,
"UDP transport adopted existing socket"
);
} else {
info!(
local_addr = %self.local_addr.unwrap(),
recv_buf = actual_recv,
send_buf = actual_send,
"UDP transport adopted existing socket"
);
}
Ok(())
}
/// Stop the transport asynchronously.
pub async fn stop_async(&mut self) -> Result<(), TransportError> {
if !self.state.is_operational() {
@@ -309,6 +368,27 @@ impl Transport for UdpTransport {
}
}
impl Drop for UdpTransport {
fn drop(&mut self) {
let had_task = self.recv_task.is_some();
let had_socket = self.socket.is_some();
if had_task || had_socket {
debug!(
transport_id = %self.transport_id,
state = ?self.state,
had_recv_task = had_task,
had_socket = had_socket,
"UdpTransport dropped without stop_async(); cleaning up",
);
}
if let Some(task) = self.recv_task.take() {
task.abort();
}
self.socket.take();
self.local_addr = None;
}
}
/// UDP receive loop - runs as a spawned task.
async fn udp_receive_loop(
socket: AsyncUdpSocket,
@@ -379,6 +459,8 @@ mod tests {
mtu: Some(1280),
recv_buf_size: None,
send_buf_size: None,
advertise_on_nostr: None,
public: None,
}
}
+106
View File
@@ -125,6 +125,81 @@ mod platform {
})
}
/// Adopt an existing bound UDP socket.
///
/// This preserves socket identity/NAT mapping created by bootstrap code.
pub fn adopt(
socket: std::net::UdpSocket,
recv_buf_size: usize,
send_buf_size: usize,
) -> Result<Self, TransportError> {
let sock = Socket::from(socket);
sock.set_nonblocking(true).map_err(|e| {
TransportError::StartFailed(format!("set nonblocking failed: {}", e))
})?;
sock.set_recv_buffer_size(recv_buf_size)
.map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;
sock.set_send_buffer_size(send_buf_size)
.map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?;
let actual_recv = sock
.recv_buffer_size()
.map_err(|e| TransportError::StartFailed(format!("get recv buffer: {}", e)))?;
let actual_send = sock
.send_buffer_size()
.map_err(|e| TransportError::StartFailed(format!("get send buffer: {}", e)))?;
if actual_recv < recv_buf_size {
warn!(
requested = recv_buf_size,
actual = actual_recv,
"UDP recv buffer clamped by kernel (increase net.core.rmem_max)"
);
}
if actual_send < send_buf_size {
warn!(
requested = send_buf_size,
actual = actual_send,
"UDP send buffer clamped by kernel (increase net.core.wmem_max)"
);
}
#[cfg(target_os = "linux")]
{
let enable: libc::c_int = 1;
let ret = unsafe {
libc::setsockopt(
sock.as_raw_fd(),
libc::SOL_SOCKET,
libc::SO_RXQ_OVFL,
&enable as *const _ as *const libc::c_void,
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
)
};
if ret < 0 {
warn!(
"setsockopt(SO_RXQ_OVFL) failed: {}",
std::io::Error::last_os_error()
);
}
}
let local_addr = sock
.local_addr()
.map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?
.as_socket()
.ok_or_else(|| {
TransportError::StartFailed("local address is not an IP socket".into())
})?;
Ok(Self {
inner: sock,
local_addr,
})
}
/// Get the local bound address.
pub fn local_addr(&self) -> SocketAddr {
self.local_addr
@@ -371,6 +446,37 @@ mod platform {
})
}
/// Adopt an existing bound UDP socket.
pub fn adopt(
socket: std::net::UdpSocket,
recv_buf_size: usize,
send_buf_size: usize,
) -> Result<Self, TransportError> {
let sock = Socket::from(socket);
sock.set_nonblocking(true).map_err(|e| {
TransportError::StartFailed(format!("set nonblocking failed: {}", e))
})?;
sock.set_recv_buffer_size(recv_buf_size)
.map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;
sock.set_send_buffer_size(send_buf_size)
.map_err(|e| TransportError::StartFailed(format!("set send buffer: {}", e)))?;
let local_addr = sock
.local_addr()
.map_err(|e| TransportError::StartFailed(format!("get local addr: {}", e)))?
.as_socket()
.ok_or_else(|| {
TransportError::StartFailed("local address is not an IP socket".into())
})?;
Ok(Self {
inner: sock,
local_addr,
})
}
/// Get the local bound address.
pub fn local_addr(&self) -> SocketAddr {
self.local_addr