laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Nostr-mediated overlay discovery and UDP NAT traversal (#53)

Browse Source 
Optional peer discovery and NAT hole-punching path gated behind a new
`nostr-discovery` cargo feature. Nodes publish signed overlay endpoint
adverts to public Nostr relays, consume peer adverts to populate
fallback dial addresses, and use STUN-assisted UDP hole punching with
NIP-59 gift-wrap offer/answer signaling to establish direct UDP paths
between NATed peers. Once a punched socket is up, it is handed into
the existing FIPS UDP transport and the standard Noise/FMP session
stack takes over unchanged.

The cargo feature is in the default feature set
(`default = ["nostr-discovery"]`) so stock builds include it; a
build that explicitly disables default features (or selects a
feature set without `nostr-discovery`) does not link the nostr /
nostr-sdk crates and does not emit a no-op poll in the tick loop.
Runtime behavior is independently gated by
`node.discovery.nostr.enabled`, which defaults to false; if the
config enables Nostr on a non-feature build, startup logs a
warning and continues without it.

== Cargo feature and dependencies

- New cargo feature `nostr-discovery = ["dep:nostr", "dep:nostr-sdk"]`.
  Not in the default feature set.
- New optional Linux-only dependencies: `nostr 0.44` (features: std,
  nip59) and `nostr-sdk 0.44`. Gift-wrap unwrap is hand-rolled in
  `src/discovery/nostr/signal.rs` rather than relying on the SDK's
  rumor-author check, which FIPS sidesteps by trusting `seal.pubkey`
  exclusively.

== Wire format

Overlay advert event: `kind 37195`, parameterized replaceable
(NIP-01 application-defined replaceable range 30000-39999), with
`d = "fips-overlay-v1"`. The digits visually spell FIPS (7=F, 1=I,
9=P, 5=S); a relay survey confirmed the kind is unused.

Advert content carries the version tag, endpoint list
(`udp|tcp|tor` + addr), optional signal-relay and stun-server
metadata, and `issuedAt` / `expiresAt` timestamps. Endpoint
`addr: "nat"` is the sentinel that triggers traversal on the peer
side. NIP-40 `expiration` tag bounds staleness on permanent
shutdown. Lifecycle relies on parameterized-replaceable
supersession; the daemon does not emit NIP-09 kind-5 deletes —
strict relays (Damus, Primal) race delete-against-replace and can
silently drop the replacement.

Gift-wrapped signal event: `kind 21059`. Punch packets carry magic
values `PUNCH_MAGIC` / `PUNCH_ACK_MAGIC`, a sequence number, and a
16-byte session hash.

== Discovery surface

- `src/discovery.rs` (always compiled)
  - `EstablishedTraversal`: bound UDP socket + selected remote +
    peer npub + optional transport name/config tuning overrides.
  - `BootstrapHandoffResult`: returned on successful handoff —
    allocated transport id, local/remote addrs, peer NodeAddr,
    session id.
- `src/discovery/nostr/` (`#![cfg(feature = "nostr-discovery")]`)
  - `types.rs`: wire and control types described above. `ADVERT_KIND`
    constant. `BootstrapError` enumerates failure modes (disabled,
    missing advert, missing NAT endpoint, no usable relays, invalid
    advert, invalid npub, signal timeout, punch timeout, replay,
    STUN failure, protocol, nostr, io, serde, event-parse).
  - `runtime.rs`: `NostrDiscovery` coordinator. Owns the shared
    nostr-sdk `Client`, subscribes to advert + signal event kinds,
    maintains a bounded advert cache and a bounded seen-sessions
    replay set, drains `BootstrapEvent::{Established, Failed}` for
    the node to consume, exposes `update_local_advert`,
    `request_connect`, `advert_endpoints_for_peer`,
    `cached_open_discovery_candidates`, and `shutdown`.
  - `signal.rs`: NIP-59 gift-wrap encode/decode. Outbound wraps are
    built against per-attempt ephemeral keys; inbound events are
    unwrapped against the node identity.
  - `stun.rs`: RFC 5389/8489 Binding Request client with
    XOR-MAPPED-ADDRESS parsing for both IPv4 and IPv6; used only to
    observe the initiator's own reflexive address against its
    locally configured STUN list (peer-advertised STUN is
    informational, never an egress target).
  - `traversal.rs`: per-attempt candidate-pair punch planner.
    Allocates a fresh `0.0.0.0:0` UDP socket per attempt, enumerates
    LAN-private and ULA interface addresses alongside the STUN
    reflexive address, schedules probe/ack exchanges at the
    configured interval for the configured duration, and picks the
    first candidate pair that authenticates end-to-end.

Strategy ordering is Reflexive↔Reflexive first, then LAN, then
Mixed. The STUN-observed pair is the only candidate that's reliable
across arbitrary network topologies; trying it first prevents the
planner from latching onto a misleading host-candidate path before
the reflexive path gets a chance. There is no catch-all
Local↔Local strategy: a previous design that paired every local
host candidate from one side with every local host candidate from
the other could declare success on a one-way reachable asymmetric
L3 path (corporate VPN, Tailscale subnet route, overlapping private
address space), only for the FMP handshake to stall because the
return path didn't match. The legitimate `Lan` strategy still pairs
candidates that share a subnet.

== Configuration surface

`node.discovery.nostr.*` (`NostrDiscoveryConfig`), all `serde(default)`
with `deny_unknown_fields`:

- `enabled` (default false), `advertise` (default true)
- `advert_relays`, `dm_relays`, `stun_servers`: defaults are
  `wss://relay.damus.io`, `wss://nos.lol`, `wss://offchain.pub`
  for both relay lists, and Google / Cloudflare / Twilio for STUN.
  Operators are expected to override for production. Other
  verified-working public relays for reference:
  `nostr.bitcoiner.social`, `nostr-pub.wellorder.net`,
  `nostr.oxtr.dev`, `nostr.mom`.
- `app` (default `"fips-overlay-v1"`), `signal_ttl_secs` (120)
- `policy`: `NostrDiscoveryPolicy::{Disabled, ConfiguredOnly (default),
  Open}` — controls whether advert-derived endpoints are consumed
  only for peers carrying `via_nostr = true`, or also for
  non-configured peers within a budget cap.
- `share_local_candidates` (default false) — when false, the offer's
  `local_addresses` list is empty and peers see only the reflexive
  address. Enable per-node only for genuinely same-LAN deployments;
  off-by-default eliminates the misleading-path failure mode for
  the common case where peers are not on the same broadcast domain.
- `open_discovery_max_pending` (64) — caps queued open-discovery
  retries; bounded by available outbound slots.
- `max_concurrent_incoming_offers` (16) — semaphore against offer
  spam; excess offers are debug-logged and dropped.
- `advert_cache_max_entries` (2048) and `seen_sessions_max_entries`
  (2048) — bound memory under ambient relay volume; overflow
  evictions are debug-logged.
- `attempt_timeout_secs` (10), `replay_window_secs` (300)
- `punch_start_delay_ms` (2000), `punch_interval_ms` (200),
  `punch_duration_ms` (10000)
- `advert_ttl_secs` (3600), `advert_refresh_secs` (1800)

Per-peer and per-transport flags:

- `PeerConfig.via_nostr: bool` — when true (and Nostr is enabled),
  advert-derived addresses are appended as fallback dial candidates
  after static addresses for that peer.
- `PeerConfig.addresses` is now `serde(default)` and may be empty
  when `via_nostr: true`; validation requires at least one of the
  two to be present per peer, and the error message names the
  peer's npub.
- `UdpConfig.advertise_on_nostr: Option<bool>` and
  `UdpConfig.public: Option<bool>` — UDP transports can be
  advertised either as direct `host:port` (public = true) or as the
  `addr: "nat"` sentinel that triggers rendezvous on the peer side.
- `TcpConfig.advertise_on_nostr` and `TorConfig.advertise_on_nostr`
  — TCP and Tor onion endpoints can be advertised as directly
  reachable.
- A reserved peer address `transport: udp, addr: "nat"` parses without
  special-casing in YAML and routes through the bootstrap runtime.

Cross-field validation (`Config::validate`, called from `Node::new`
and `Node::with_identity`):

- Any transport with `advertise_on_nostr = true` requires
  `node.discovery.nostr.enabled = true`.
- Any peer with `via_nostr = true` requires
  `node.discovery.nostr.enabled = true`.
- A non-public UDP advert (`advertise_on_nostr = true`,
  `public = false` — i.e. `udp:nat`) additionally requires at least
  one `dm_relay` and at least one `stun_server`.
  Surfaced as `ConfigError::Validation`.

== Node integration

`src/node/lifecycle.rs` is the main integration point.

- At node start (after transports are up, before TUN), if Nostr is
  enabled and the feature is compiled in, `NostrDiscovery::start` is
  invoked, the initial local overlay advert is built from the live
  transport set and published, and the runtime handle is stored.
- The rx tick loop calls `poll_nostr_discovery` (feature-gated both
  at method definition and call site), which refreshes the local
  advert, drains bootstrap events, adopts established traversals,
  schedules retries for failed traversals, and — under `policy:
  open` — enqueues outbound retries for non-configured peers
  visible in the advert cache, bounded by
  `open_discovery_max_pending` and the remaining outbound slots.
- Outbound peer dialing is refactored to `try_peer_addresses`, which
  first exhausts the static address list in priority order and only
  then appends advert-derived fallback addresses; both lists run
  through the same `attempt_peer_address_list` code path. The
  `udp:nat` sentinel address triggers `NostrDiscovery::request_connect`
  for the peer instead of a direct dial and returns `Ok(())`.
- `build_overlay_advert` walks operational transports, consults
  per-instance `UdpConfig` / `TcpConfig` / `TorConfig` (matching by
  optional transport instance name), and emits an `OverlayAdvert`
  including `signalRelays` and `stunServers` when any UDP endpoint
  is advertised as NAT.
- `adopt_established_traversal` is the bootstrap handoff API:
  allocates a new `TransportId`, constructs a `UdpTransport` with
  the user-supplied (or default) `UdpConfig`, calls the new
  `adopt_socket_async` to reuse the punched socket verbatim,
  registers the transport in the normal transport map, records it
  in `bootstrap_transports`, and calls `initiate_connection` so the
  normal handshake path runs. On failure, the transport is stopped
  and removed cleanly and the set membership is rolled back.
- On clean shutdown, `NostrDiscovery::shutdown` is awaited so
  background tasks stop before transports are torn down. (The
  advert is not explicitly retracted; NIP-40 expiration plus the
  next refresh from any live publisher supersedes it.)

New `Node` fields:

- `nostr_discovery: Option<Arc<NostrDiscovery>>` (feature-gated).
- `bootstrap_transports: HashSet<TransportId>` — per-peer UDP
  transports adopted from NAT traversal, cleaned up via
  `cleanup_bootstrap_transport_if_unused` whenever the link,
  connection, peer, or pending-connect referencing them is removed.

Retry and error surface:

- `RetryState.expires_at_ms: Option<u64>` — optional absolute expiry
  for a retry entry. `pump_retries` drops expired entries with an
  info log. Used for open-discovery retries, which expire at two
  times the advert TTL.
- New `NodeError::BootstrapHandoff(String)` returned from
  `adopt_established_traversal` when the underlying transport
  adoption fails or local address discovery fails.
- New `ConfigError::Validation(String)`.
- A small refactor extracts `Node::now_ms()` and reuses it across
  lifecycle, rx-loop tick, and timeout bookkeeping.

== UDP transport

`src/transport/udp/`:

- `UdpRawSocket::adopt(std::net::UdpSocket, recv_buf, send_buf)`:
  adopts an externally bound socket, makes it non-blocking, applies
  the configured buffer sizes (warning if the kernel clamps), and
  reports the resulting local address. Preserves the NAT mapping —
  no rebind.
- `UdpTransport::adopt_socket_async(std::net::UdpSocket)`: the
  `start_async` analogue for an already-bound socket, wiring the
  async socket and recv task exactly as the fresh-bind path would.
- `Drop` impl for `UdpTransport`: if a transport is dropped while
  still holding a recv task or socket (for example on error
  teardown), aborts the task, clears the socket, and emits a debug
  log so the cleanup is visible in tracing rather than silent.

== Logging and observability

Default `EnvFilter` demotes third-party relay-pool DEBUG output to
TRACE-only: `nostr_relay_pool`, `nostr_sdk`, and `nostr` are pinned
at INFO when our level is anything below TRACE, and at TRACE when
our level is TRACE — so the raw frames are still reachable when
explicitly asked for. RUST_LOG continues to override completely.

Concise one-line DEBUG events are emitted at the meaningful points
in the discovery / hole-punch sequence:

- `advert: published` (event id, relay count, endpoints, ttl)
- `advert: peer cached` (notify-loop ingress for non-self)
- `advert: resolved` (cache hit / relay fetch outcome)
- `traversal: initiator starting`
- `traversal: initiator STUN observed` (reflexive, local count)
- `traversal: offer sent` (session id, relay count, event id)
- `traversal: answer received` (accepted, reflexive, local)
- `traversal: initiator punch succeeded` (remote addr)
- `traversal: offer received` (responder side)
- `traversal: responder STUN observed`
- `traversal: answer sent`
- `traversal: responder punch succeeded`

Npubs are shortened to `npub1<4>..<4>` and event/session ids to
their first 8 hex characters.

Other operator-facing logs:

- `UdpTransport` adoption and drop paths log at info / debug.
- `adopt_established_traversal` logs at debug on entry and info on
  successful return, tagged with peer npub, session id, transport
  id, and both socket endpoints, so the bootstrap handoff is
  traceable end-to-end alongside the `UdpTransport::drop` log.
- `cleanup_bootstrap_transport_if_unused` logs at debug when the
  reference-count check drops an adopted transport.
- `connect_peer` tags its entry `debug!` with `peer_npub` so
  downstream STUN, punch, and handshake logs for the same peer
  correlate for operators.
- Advert-cache and seen-sessions overflow evictions log at debug so
  mis-sized caps are visible under ambient relay volume.
- Gift-wrap unwrap failures on `SIGNAL_KIND` events log at trace
  (hot path: fires for every unrelated signal event on the same
  relay).
- Traversal-offer handler failures log at debug. Expected conditions
  such as punch timeout on symmetric NAT are covered there; real
  problems are reported upstream via `BootstrapEvent::Failed`.
- Inbound-offer rate-limit messages name the governing config field
  (`max_concurrent_incoming_offers`) and state that the offer was
  rate-limited rather than failing.

== Tests

- 18 new unit tests in `src/discovery/nostr/tests.rs` covering advert
  encoding, signal envelope round-trip, STUN parsing, punch-packet
  codec, and replay-window enforcement. Run under the
  `nostr-discovery` feature.
- Config-validation tests in `src/config/mod.rs` covering the three
  cross-field invariants and YAML parsing of the full
  `node.discovery.nostr` block plus `peers[].via_nostr`, empty
  `addresses` with `via_nostr: true`, and a `udp: nat` address.
- `src/node/tests/bootstrap.rs` integration tests that drive a
  synthetic traversal (bound UDP socket pair + synthetic peer
  identity) through `adopt_established_traversal` and assert the
  Noise handshake completes over the adopted socket.
- Punch-planner tests assert reflexive-before-LAN ordering and that
  same-LAN scenarios still include the LAN target in the plan.
- `testing/nat/` Docker NAT lab harness:
  - Local `strfry` relay, local STUN responder, and one or two
    router containers performing `iptables` NAT.
  - Node LAN interfaces are provisioned with explicit `veth` pairs
    injected into the node and router namespaces so every packet
    traverses the router namespace (plain Docker bridges are not
    used for the LAN).
  - `cone` scenario: both peers behind full-cone-emulation NAT
    (SNAT with source-port preservation, inbound DNAT back to the
    single LAN host regardless of remote source); asserts UDP
    traversal succeeds and link remote addresses are on the router
    WAN subnet.
  - `symmetric` scenario: `MASQUERADE --random-fully`; asserts UDP
    traversal fails and TCP fallback converges over router-
    published WAN addresses.
  - `lan` scenario: both peers share a LAN subnet; asserts LAN
    addresses are preferred over reflexive ones.
  - Cleanup tears down all profile-gated services
    (`--profile cone --profile symmetric --profile lan`) so no
    orphan containers survive a run.
- `testing/scripts/build.sh` builds the Docker test image with
  `--features "tui nostr-discovery"` by default so NAT-harness
  binaries include bootstrap support.

== CI

- Linux release build and nextest unit-test job both use
  `--features "gateway nostr-discovery"` so the feature-gated code
  and its unit tests compile and run in CI.
- Three new integration matrix entries (`nat-cone`, `nat-symmetric`,
  `nat-lan`) invoke `testing/nat/scripts/nat-test.sh`, collect
  `docker compose logs` on failure, and always stop containers.

== Packaging and operations

- `packaging/common/fips.yaml` ships a fully commented
  `node.discovery.nostr.*` block, plus documented
  `advertise_on_nostr` / `public` examples under the UDP transport,
  an `advertise_on_nostr` example under TCP, and a `via_nostr: true`
  example under the static peer section with both a direct
  `host:port` UDP address and a `udp: nat` fallback.
- `.github/workflows/package-openwrt.yml`: NIP-94 release event
  publishes target the new default relay set.

== Documentation

- `README.md`: overlay discovery + NAT traversal moved from
  "Near-term priorities" into "What works today".
- `docs/design/fips-intro.md`: rewrites the paragraphs that
  previously described Nostr discovery and NAT traversal as future
  work; describes the shipped mechanism and the feature gate.
- `docs/design/fips-transport-layer.md`: drops the "(future
  direction)" qualifier from the Nostr Relay Discovery section,
  expands with the `udp:nat` advertisement and bootstrap handoff
  description, and updates the Current State callout.
- `docs/design/fips-mesh-layer.md`: notes that mid-session NAT
  rebinding (roaming) and initial NAT traversal (Nostr path) are
  distinct mechanisms.
- `docs/design/fips-configuration.md`: documents the full
  `node.discovery.nostr.*` surface, including the three resource
  caps and `share_local_candidates`.
- `docs/design/fips-nostr-discovery.md`: design and configuration
  reference for the shipped mechanism, including the empty-
  `addresses`-with-`via_nostr` shorthand.
- `docs/proposals/nostr-udp-hole-punch-protocol.md`: adds an
  Implemented status callout, clarifies that the punch socket is
  per-peer and per-attempt rather than shared with the application
  listener, aligns field names with the shipped JSON
  (`sessionId`, `issuedAt` / `expiresAt`, `reflexiveAddress`,
  `localAddresses`, `stunServer`), sets the `d`-tag to
  `fips-overlay-v1`, names the kind as 37195, and notes that
  advertised STUN entries are informational.
- `docs/proposals/README.md`: adds a Status column and marks the
  hole-punching proposal Implemented.
- `CHANGELOG.md`: Unreleased > Added entry covering the discovery
  path, STUN/punch path, configuration surface, and Docker NAT lab.

Co-authored-by: Johnathan Corgan <johnathan@corganlabs.com>
This commit is contained in:
Tom
2026-04-27 08:15:58 -07:00
committed by GitHub
co-authored by Johnathan Corgan
parent 1e3b2c319e
commit 34e00b9f6e
55 changed files with 7448 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/dispatch.rs
+3
View File
@@ -175,6 +175,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);
src/node/handlers/rx_loop.rs
+3 -4
View File
@@ -112,12 +112,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;
src/node/handlers/timeout.rs
+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 if allocated
if let Some(idx) = conn.our_index() {
@@ -81,6 +79,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.
src/node/lifecycle.rs
+687 -93
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(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 {
// 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, 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.
@@ -296,10 +230,7 @@ impl Node {
let peer_node_addr = *peer_identity.node_addr();
// Create connection in handshake phase (outbound knows expected identity)
let current_time_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let current_time_ms = Self::now_ms();
let mut connection = PeerConnection::outbound(link_id, peer_identity, current_time_ms);
// Allocate a session index for this handshake
@@ -450,6 +381,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
@@ -537,11 +520,7 @@ impl Node {
// Clean up link and schedule retry
self.remove_link(&pending.link_id);
self.links.remove(&pending.link_id);
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(*pending.peer_identity.node_addr(), now_ms);
self.schedule_retry(*pending.peer_identity.node_addr(), Self::now_ms());
}
}
}
@@ -564,7 +543,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 {
@@ -590,6 +569,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;
@@ -856,6 +860,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 {
@@ -963,6 +975,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, 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.
@@ -976,12 +1482,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)
@@ -1034,4 +1541,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(), 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,
})
}
}
src/node/mod.rs
+60 -5
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.
@@ -339,7 +342,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.
@@ -428,6 +430,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>,
@@ -466,6 +474,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();
@@ -587,6 +596,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,
@@ -599,7 +611,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];
@@ -655,7 +671,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(),
@@ -708,6 +724,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,
@@ -716,7 +735,7 @@ impl Node {
peer_aliases: HashMap::new(),
peer_acl,
host_map,
}
})
}
/// Create a leaf-only node (simplified state).
@@ -1386,6 +1405,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()
src/node/retry.rs
+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);
}
src/node/tests/bootstrap.rs
+245
View File
@@ -0,0 +1,245 @@
//! 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};
#[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"
);
}
#[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, 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();
}
}
src/node/tests/mod.rs
+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;
src/node/tests/unit.rs
+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();
@@ -716,6 +779,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