diff --git a/AGENTS.md b/AGENTS.md index d2cfc0a..9506c8b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,6 +43,11 @@ Keep these separate. Crossing them is the main thing to review for. - **A protocol never sees the network secret.** It proves who is at the other end of a tunnel; proving membership of a network stays in the core, which is why the authenticated transport does too. +- **A relay carries bytes, never packets.** A datagram passed on for two + other peers goes link in, link out: it is never written to the middle's + interface, never decrypted there, and never relayed twice. Reachability + that decides a route is first-hand and volatile — a live announcement + over the control plane, never a signed record and never second-hand. - **Plugins never learn reachability.** An `IpPlugin` is handed a `PacketLink` per peer and moves datagrams over it. Addresses, hole punching and relays belong to `crates/tsunagi/src/dataplane/transport/`. A plugin announcement says *who*, never diff --git a/README.md b/README.md index b86fa03..b751da0 100644 --- a/README.md +++ b/README.md @@ -536,6 +536,27 @@ async fn main() -> Result<()> { reaches the internet by accident. Opt into `DirectOnly` or `N0Defaults` explicitly. +### Through somebody in the middle + +Two members can both reach a third and not each other: a blocked path, a +relay that is unavailable, a network only reachable from inside somebody +else's building. When that happens the pair is routed through a member that +has both. + +Nothing is agreed and nothing is elected. Each member says only which peers +*it* has a live link with, first-hand, over the control plane and one hop +only; everyone picks their own way through from that, deterministically, and +drops it the moment a direct link exists. There is no routing protocol, no +second-hand claim to weigh, and a relayed datagram is never relayed again — +so a loop cannot form. + +The one in the middle carries **bytes it cannot read**: the tunnel stays end +to end between the two ends, and a relayed datagram never touches the middle +host's interface, so no forwarding, routing or firewall setting of that host +is involved. `status` says `via ` on a path that goes through +somebody, and counts what this device has carried for others — it is their +traffic on your uplink, and that should not be invisible. + ## How peers find each other Two different lookups are involved, and only one of them is this project's: diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 6933855..d343af7 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -2525,6 +2525,21 @@ fn network_section( )); } + // Only when something has: a relay that has carried nothing is not + // worth a line, and one that has is worth knowing about — it is + // somebody else's traffic on this device's uplink. + let relayed = network.relay_forwarded + network.relay_sent_via + network.relay_received_via; + if relayed > 0 { + section.push(Row::new( + Health::Info, + "relay", + format!( + "{} carried for others · {} sent through a peer, {} arrived through one", + network.relay_forwarded, network.relay_sent_via, network.relay_received_via + ), + )); + } + let rows = member_rows(network, own_id); let online = rows.iter().filter(|row| row.online()).count(); if rows.is_empty() { @@ -3530,6 +3545,9 @@ async fn build_report( network_id: network.network_id.to_string(), active: matches!(network.state, tsunagi::agent::NetworkState::Active), candidates: network.candidates.len() as u32, + relay_forwarded: network.relay.forwarded, + relay_sent_via: network.relay.sent_via, + relay_received_via: network.relay.received_via, peers: network .peers .iter() @@ -3825,6 +3843,9 @@ mod status_tests { network_id: "xa7gyz".into(), active: true, candidates: 1, + relay_forwarded: 0, + relay_sent_via: 0, + relay_received_via: 0, peers: vec![PeerReport { endpoint_id: ONLINE.into(), hostname: Some("music".into()), diff --git a/crates/tsunagi-wg-quic/tests/wireguard.rs b/crates/tsunagi-wg-quic/tests/wireguard.rs index 46380ae..f3d84eb 100644 --- a/crates/tsunagi-wg-quic/tests/wireguard.rs +++ b/crates/tsunagi-wg-quic/tests/wireguard.rs @@ -162,6 +162,43 @@ impl WgAgent { (agent, plugin, tuns) } + /// An agent that refuses a direct data path to some peers. + /// + /// The one arrangement a single host cannot produce by itself: two + /// agents that both reach a third and not each other. The control + /// plane is untouched — they are members and they talk — only the + /// direct data link is refused, which is the real-world case of a + /// blocked or unreachable data path. + async fn spawn_cut_off( + discovery: &SharedMemoryDiscovery, + tag: &str, + blocked: Arc>>, + ) -> Self { + let dir = TempDir::new().unwrap(); + let tuns = MemoryTunFactory::new(); + let plugin = WireguardPlugin::open( + WireguardConfig::new(dir.path().join("wireguard")) + .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), + ) + .await + .unwrap(); + let agent = Agent::spawn( + config_with(dir.path(), discovery) + .with_overlay_ipv4_range(Some(tsunagi::state::DEFAULT_IPV4_RANGE)) + .with_interface(Arc::new(tuns.clone()), tag, 1280) + .with_unreachable_data_peers(blocked) + .with_plugin(plugin.clone() as Arc), + ) + .await + .unwrap(); + Self { + dir, + agent, + plugin, + tuns, + } + } + /// An agent whose interface claims to be on the host but is not. /// /// Used only by the missing-address test: everywhere else the in-memory @@ -1306,3 +1343,78 @@ async fn an_mtu_below_what_ipv4_guarantees_is_refused() { .is_ok() ); } + +#[tokio::test] +async fn a_peer_with_no_direct_path_is_reached_through_one_that_has_both() { + // A and B can each reach C and not each other. Without a way through + // the middle they are lost to one another while sitting in the same + // mesh; with one, C carries their datagrams without being able to read + // a byte of them — the WireGuard tunnel is still end to end. + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-relay"); + + let a_blocks = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())); + let b_blocks = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())); + let a = WgAgent::spawn_cut_off(&discovery, "tra", Arc::clone(&a_blocks)).await; + let b = WgAgent::spawn_cut_off(&discovery, "trb", Arc::clone(&b_blocks)).await; + let middle = WgAgent::spawn(&discovery, "trc").await; + a_blocks.lock().unwrap().insert(b.endpoint_id()); + b_blocks.lock().unwrap().insert(a.endpoint_id()); + + let network_id = middle.agent.join_network(&name, &secret).await.unwrap(); + a.agent.join_network(&name, &secret).await.unwrap(); + b.agent.join_network(&name, &secret).await.unwrap(); + + // All three are members and all three talk: only the data path + // between A and B is missing. + wait_for_peers(&a.agent, network_id, 2).await; + wait_for_peers(&b.agent, network_id, 2).await; + middle.wait_for_tunnels(network_id, 2).await; + + let a_addr = a.overlay(network_id).await; + let b_addr = b.overlay(network_id).await; + assert_ne!(a_addr, b_addr); + + // A's tunnel to B comes up through C. + wait_until("a's tunnel to b is established", || async { + let view = a.plugin.overview(network_id)?; + view.peers + .iter() + .find(|peer| peer.endpoint_id == b.endpoint_id())? + .tunnel + .as_ref() + .map(|tunnel| tunnel.health.since_handshake)? + .map(|_| ()) + }) + .await; + let path = a + .plugin + .overview(network_id) + .unwrap() + .peers + .iter() + .find(|peer| peer.endpoint_id == b.endpoint_id()) + .and_then(|peer| peer.tunnel.as_ref().map(|tunnel| tunnel.path.clone())) + .unwrap_or_default(); + assert!( + path.contains("via"), + "the path should say it goes through somebody: {path}" + ); + + // And a real packet crosses: A's interface to B's interface, through C. + a.tun(network_id) + .await + .push_from_os(ipv4_packet(a_addr, b_addr, b"through the middle")); + let seen = tokio::time::timeout( + tsunagi::testing::DEADLINE, + b.tun(network_id).await.pop_to_os(), + ) + .await + .expect("the packet should arrive") + .unwrap(); + assert_eq!(&seen[20..], b"through the middle"); + + a.shutdown().await; + b.shutdown().await; + middle.shutdown().await; +} diff --git a/crates/tsunagi/src/agent/mod.rs b/crates/tsunagi/src/agent/mod.rs index 75bc251..d3abcff 100644 --- a/crates/tsunagi/src/agent/mod.rs +++ b/crates/tsunagi/src/agent/mod.rs @@ -567,6 +567,8 @@ impl Agent { hostname: self.inner.read_hostname(), transport: self.inner.transport.get().cloned(), device_secret: self.inner.identity.signing_key(), + #[cfg(feature = "testing")] + unreachable_data_peers: Arc::clone(&self.inner.config.unreachable_data_peers), ipv4_range: range.propose, ipv4_fallback: range.fallback, range_conflict: range.conflict, @@ -771,6 +773,7 @@ impl Agent { range: None, range_conflict: None, metrics: NetworkMetrics::default(), + relay: Default::default(), }); } diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 08d8f79..e5ff483 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -15,7 +15,7 @@ use tokio::sync::{broadcast, mpsc, oneshot}; use tokio::task::JoinHandle; use crate::config::{Limits, ReconnectPolicy}; -use crate::dataplane::transport::{InboundLink, PacketTransport, SharedLink}; +use crate::dataplane::transport::{InboundLink, PacketLink, PacketTransport, SharedLink}; use crate::dataplane::{PluginCapability, SharedPlugin}; use crate::discovery::{Candidate, CandidateSource, NetworkDiscovery}; use crate::error::{Error, Result}; @@ -137,6 +137,9 @@ pub(crate) struct RuntimeParams { pub(crate) hostname: String, /// Signing key for this agent's own records. pub(crate) device_secret: iroh::SecretKey, + /// Peers with no direct data path, for tests only. + #[cfg(feature = "testing")] + pub(crate) unreachable_data_peers: Arc>>, /// The IPv4 overlay range this agent would use, if the network has not /// already settled on another one. pub(crate) ipv4_range: Option, @@ -212,6 +215,14 @@ pub(crate) fn spawn(params: RuntimeParams) -> NetworkHandle { } } +/// How long a peer's account of what it can reach is believed. +/// +/// Soft state: it is repeated while it holds, and simply stops being +/// repeated when the peer goes. Nothing is revoked, so nothing can be +/// revoked wrongly — a relay that disappears stops being chosen by +/// itself, a few announcements later. +const REACH_EXPIRY: Duration = Duration::from_secs(90); + /// How long a network waits to be told its range before proposing the one /// derived from its id. /// @@ -236,7 +247,23 @@ struct Runtime { dial_results_tx: mpsc::Sender, dial_results_rx: mpsc::Receiver, /// Live data plane links, keyed by peer and plugin protocol. + /// + /// The direct ones. What a protocol holds is the hub's link for that + /// peer, which outlives any particular path. links: HashMap<(EndpointId, String), SharedLink>, + /// Every data link of this network, and what is reached through what. + hub: Arc, + /// Which peers each peer says it has a live link with, and when it + /// last said so. + /// + /// First-hand only: a peer speaks for itself and never for anybody + /// else, so there is no hearsay to weigh and nothing to poison. Soft + /// state — an entry that stops being refreshed expires, which is how a + /// relay that has gone stops being chosen without anybody revoking it. + reachable: HashMap, std::time::Instant)>, + /// What this agent last told its peers it could reach, so it is only + /// said again when it changed. + announced_reach: HashSet, /// Links currently being opened, so we do not start two. opening: HashSet<(EndpointId, String)>, link_results_tx: mpsc::Sender, @@ -283,6 +310,9 @@ impl Runtime { dial_results_tx, dial_results_rx, links: HashMap::new(), + hub: crate::dataplane::relay::RelayHub::new(network_id), + reachable: HashMap::new(), + announced_reach: HashSet::new(), opening: HashSet::new(), link_results_tx, link_results_rx, @@ -491,6 +521,10 @@ impl Runtime { // alone, so something has to look again; this runs on a timer and // the check is a comparison when nothing has changed. self.ensure_own_claim().await; + // The same timer repairs the relay picture: reachability is soft + // state, repeated while it holds and forgotten when it stops. + self.update_paths(); + self.announce_reach(true); let mut candidates: Vec = Vec::new(); @@ -1040,6 +1074,10 @@ impl Runtime { } for (peer, protocol) in dead { self.links.remove(&(peer, protocol.clone())); + // Only the direct path is gone. Whether the peer is still + // reachable through somebody is decided below; the protocol's + // link stays either way. + self.hub.clear_direct(peer, &protocol); self.emit(Event::DataLinkDown { network: self.network_id, peer, @@ -1076,6 +1114,9 @@ impl Runtime { if self.links.contains_key(&key) || self.opening.contains(&key) { continue; } + if self.no_direct_data(peer) { + continue; + } // The smaller endpoint id dials; the other side accepts. Both // compute this identically, so exactly one link is created. if self.local_id.as_bytes() >= peer.as_bytes() { @@ -1106,11 +1147,142 @@ impl Runtime { } } + /// Tells peers which peers this agent has a live link with, when that + /// has changed. + /// + /// Only what it can see for itself, and only to the peers it is + /// talking to: one hop, no flooding, nothing second-hand. A peer that + /// hears this is the only one that can use it, because it is also the + /// only one that could route through this agent. + fn announce_reach(&mut self, force: bool) { + let live: HashSet = self + .links + .iter() + .filter(|(_, link)| !link.is_closed()) + .map(|((peer, _), _)| *peer) + .collect(); + if !force && live == self.announced_reach { + return; + } + self.announced_reach = live.clone(); + let message = ControlMessage::Reachable { + peers: live.iter().map(|peer| *peer.as_bytes()).collect(), + }; + let peers: Vec = self.sessions.keys().copied().collect(); + for peer in peers { + if let Err(err) = self.send_to(peer, message.clone()) { + tracing::debug!(%err, "could not queue a reachability announcement"); + } + } + } + + /// Chooses, for every peer with no direct link, somebody to go through. + /// + /// Local and deterministic: the lowest endpoint id among the peers + /// that both this agent has a link with and that say they have a link + /// with the destination. Nothing is agreed with anybody — two agents + /// may well pick differently, and traffic each way finds its own path. + fn update_paths(&mut self) { + self.reachable + .retain(|_, (_, heard)| heard.elapsed() < REACH_EXPIRY); + + let served = self.served_protocols(); + let live: HashSet = self + .links + .iter() + .filter(|(_, link)| !link.is_closed()) + .map(|((peer, _), _)| *peer) + .collect(); + + // Every peer this agent would carry traffic with: a session says + // who it is and what it speaks, which is also what a protocol + // needs before it can have a link at all. + let wanted: Vec<(EndpointId, String)> = self + .sessions + .values() + .flat_map(|session| { + let peer = session.peer; + session + .capabilities + .iter() + .filter(|capability| capability.enabled) + .map(move |capability| (peer, capability.protocol.clone(), capability.version)) + }) + .filter(|(_, protocol, version)| { + served + .iter() + .any(|(name, ours)| name == protocol && ours == version) + }) + .map(|(peer, protocol, _)| (peer, protocol)) + .collect(); + + for (peer, protocol) in wanted { + if live.contains(&peer) { + // There is a direct link; a hop is only for when there is + // not, and holding a stale one would keep a peer that has + // gone looking reachable. + if self.hub.has_link(peer, &protocol) { + self.hub.set_hop(peer, &protocol, None); + } + continue; + } + let hop = self + .reachable + .iter() + .filter(|(middle, (reaches, _))| { + live.contains(*middle) && reaches.contains(&peer) && **middle != peer + }) + .map(|(middle, _)| *middle) + .min_by(|a, b| a.as_bytes().cmp(b.as_bytes())); + + let known = self.hub.has_link(peer, &protocol); + if !known && hop.is_none() { + // Nothing to reach it with and nothing to offer a + // protocol; inventing a link object here would only look + // like connectivity. + continue; + } + self.hub.set_hop(peer, &protocol, hop); + if known { + continue; + } + // A peer reachable only through somebody still gets a link, so + // its tunnel can be built: the protocol is not told how its + // datagrams travel, only that this is the way to that peer. + let Some(plugin) = self + .params + .plugins + .iter() + .find(|plugin| plugin.protocol_id() == protocol) + .cloned() + else { + continue; + }; + let link = self.hub.link(peer, &protocol); + let path = link.path_description(); + let max_datagram = link.max_datagram_size(); + plugin.on_peer_link(self.network_id, peer, link); + self.emit(Event::DataLinkUp { + network: self.network_id, + peer, + protocol, + path, + max_datagram, + }); + } + } + fn handle_link_result(&mut self, outcome: LinkOutcome) { let key = (outcome.peer, outcome.protocol.clone()); self.opening.remove(&key); match outcome.result { - Ok(link) => self.adopt_link(outcome.peer, outcome.protocol, link), + Ok(link) => { + self.adopt_link(outcome.peer, outcome.protocol, link); + // A new link is both a path for us and one we can offer + // others, so it changes what we route and what we say. + self.update_paths(); + self.announce_reach(false); + } Err(reason) => { // A data plane that cannot be set up is reported, never fatal. self.metrics.data_link_failures += 1; @@ -1124,8 +1296,32 @@ impl Runtime { } } + /// Whether this agent refuses a direct data path to a peer. + /// + /// Always false outside a test build: see + /// [`crate::config::AgentConfig::with_unreachable_data_peers`]. + fn no_direct_data(&self, peer: EndpointId) -> bool { + #[cfg(feature = "testing")] + { + match self.params.unreachable_data_peers.lock() { + Ok(guard) => guard.contains(&peer), + Err(poisoned) => poisoned.into_inner().contains(&peer), + } + } + #[cfg(not(feature = "testing"))] + { + let _ = peer; + false + } + } + fn install_link(&mut self, inbound: InboundLink) { + if self.no_direct_data(inbound.peer) { + return; + } self.adopt_link(inbound.peer, inbound.protocol, inbound.link); + self.update_paths(); + self.announce_reach(false); } /// Hands a link to the plugin that owns its protocol. @@ -1141,10 +1337,19 @@ impl Runtime { }; let path = link.path_description(); - let max_datagram = link.max_datagram_size(); self.links .insert((peer, protocol.clone()), Arc::clone(&link)); - plugin.on_peer_link(self.network_id, peer, link); + // The protocol is handed the hub's link, not this one: the same + // object for as long as the peer is a peer, whatever happens to + // the path under it. A tunnel is not torn down because a direct + // link came or went. + let first_time = !self.hub.has_link(peer, &protocol); + self.hub.set_direct(peer, &protocol, link); + let shared = self.hub.link(peer, &protocol); + let max_datagram = shared.max_datagram_size(); + if first_time { + plugin.on_peer_link(self.network_id, peer, shared); + } self.metrics.data_links_established += 1; self.emit(Event::DataLinkUp { network: self.network_id, @@ -1157,6 +1362,10 @@ impl Runtime { /// Drops every link to a peer. fn drop_links_for(&mut self, peer: EndpointId) { + // The peer itself is going, not just a path to it, so the + // protocol's link goes with it. + self.hub.remove_peer(peer); + self.reachable.remove(&peer); let keys: Vec<(EndpointId, String)> = self .links .keys() @@ -1386,6 +1595,17 @@ impl Runtime { ControlMessage::State { records } => { self.pending_state.push((peer, records.clone())); } + // What that peer can reach, from that peer. Kept with the time + // it was heard, so it can go stale on its own. + ControlMessage::Reachable { peers } => { + let heard: HashSet = peers + .iter() + .filter_map(|raw| EndpointId::from_bytes(raw).ok()) + .collect(); + self.reachable + .insert(peer, (heard, std::time::Instant::now())); + self.update_paths(); + } ControlMessage::Pong { .. } | ControlMessage::Bye { .. } => {} } @@ -1532,6 +1752,7 @@ impl Runtime { .range_conflict .filter(|_| self.effective_range().is_none()), metrics: self.metrics.clone(), + relay: self.hub.counters(), } } } diff --git a/crates/tsunagi/src/agent/status.rs b/crates/tsunagi/src/agent/status.rs index aedbef4..98ce247 100644 --- a/crates/tsunagi/src/agent/status.rs +++ b/crates/tsunagi/src/agent/status.rs @@ -176,6 +176,8 @@ pub struct NetworkStatus { pub range_conflict: Option, /// Per-network counters. pub metrics: NetworkMetrics, + /// What has gone through a peer in the middle, in both directions. + pub relay: crate::dataplane::relay::RelayCounters, } impl NetworkStatus { diff --git a/crates/tsunagi/src/config.rs b/crates/tsunagi/src/config.rs index 6f16339..4aff999 100644 --- a/crates/tsunagi/src/config.rs +++ b/crates/tsunagi/src/config.rs @@ -223,6 +223,10 @@ pub struct AgentConfig { pub bind_addrs: Vec, /// How much external connectivity machinery the endpoint may use. pub transport: TransportPolicy, + /// Peers with no direct data path, for tests. See + /// [`AgentConfig::with_unreachable_data_peers`]. + #[cfg(feature = "testing")] + pub unreachable_data_peers: Arc>>, /// Hostname announced to peers. `None` keeps whatever the state store holds, /// falling back to the OS hostname and finally to a short endpoint id. pub hostname: Option, @@ -264,6 +268,10 @@ impl AgentConfig { paths, bind_addrs: Vec::new(), transport: TransportPolicy::default(), + #[cfg(feature = "testing")] + unreachable_data_peers: Arc::new(std::sync::Mutex::new( + std::collections::HashSet::new(), + )), hostname: None, discovery: None, discovery_interval: Duration::from_secs(5), @@ -292,6 +300,21 @@ impl AgentConfig { self } + /// Peers this agent will not open or accept a **direct** data link with. + /// + /// Test-only, and behind the `testing` feature so a release build has + /// no such switch. The one arrangement that cannot be made on a single + /// host is two agents that both reach a third and not each other, + /// which is exactly the case a relay exists for. + #[cfg(feature = "testing")] + pub fn with_unreachable_data_peers( + mut self, + peers: Arc>>, + ) -> Self { + self.unreachable_data_peers = peers; + self + } + /// Sets the transport policy. pub fn with_transport(mut self, transport: TransportPolicy) -> Self { self.transport = transport; diff --git a/crates/tsunagi/src/dataplane/mod.rs b/crates/tsunagi/src/dataplane/mod.rs index 923e8e7..4953a3c 100644 --- a/crates/tsunagi/src/dataplane/mod.rs +++ b/crates/tsunagi/src/dataplane/mod.rs @@ -19,6 +19,7 @@ //! A data plane failure never stops the daemon: errors returned here are //! recorded and surfaced, the control plane keeps running. +pub mod relay; pub mod transport; use std::sync::Arc; diff --git a/crates/tsunagi/src/dataplane/relay.rs b/crates/tsunagi/src/dataplane/relay.rs new file mode 100644 index 0000000..38ff06a --- /dev/null +++ b/crates/tsunagi/src/dataplane/relay.rs @@ -0,0 +1,806 @@ +//! Carrying a peer's packets through another peer. +//! +//! Two members of a network can both reach a third and yet not reach each +//! other: a blocked path, a relay that is unavailable, a network that is +//! only reachable from inside somebody else's building. Without a way +//! through the middle, that pair is simply lost to each other while both +//! sit in the same mesh. +//! +//! # What goes through the middle, and what does not +//! +//! The relay moves **bytes it cannot read**. A datagram is wrapped with the +//! peer it is for, handed to the peer in the middle, and unwrapped on the +//! other side; the protocol's own encryption is end to end between the two +//! ends of the tunnel, so the one in the middle carries an opaque payload. +//! +//! It never reaches the middle's operating system either: a relayed +//! datagram goes link in, link out. Nothing is written to its interface, +//! so no routing, forwarding or firewall setting of its host is involved — +//! which is both the fast path and the only one allowed here, since an +//! agent touches no system object it did not create. +//! +//! # What is deliberately not here +//! +//! No routing protocol and nothing second-hand. A peer says only which +//! peers *it* has a live link with, and that is used to pick one hop — +//! never a claim about somebody else's reachability. One hop means loops +//! are impossible by construction rather than by a counter, and a relayed +//! datagram is never relayed again. +//! +//! Fairness between the peers a relay carries for is not addressed yet: the +//! queues are bounded and the counters say how much went through, which is +//! what a limit would be built on. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, RwLock, Weak}; + +use bytes::{BufMut, Bytes, BytesMut}; +use iroh::EndpointId; +use tokio::sync::mpsc; + +use crate::BoxFuture; +use crate::identity::NetworkId; + +use super::transport::{PacketLink, SharedLink, TransportError}; + +/// A datagram straight from the peer that sent it. +const TAG_DIRECT: u8 = 0; +/// A datagram for somebody else, to be passed on. +const TAG_TO_RELAY: u8 = 1; +/// A datagram that was passed on, naming who it came from. +const TAG_RELAYED: u8 = 2; + +/// The largest header any of the three shapes needs. +/// +/// Subtracted from every link's datagram size, relayed or not, so that the +/// size a protocol may use does not change when the path does. A tunnel +/// that had to renegotiate its packet size every time a path changed would +/// be worse than one that is a few bytes smaller than it could be. +pub const RELAY_OVERHEAD: usize = 1 + 32; + +/// How many datagrams may wait for a protocol to read them. +/// +/// Bounded, and the oldest is dropped rather than the newest kept waiting: +/// these are datagrams, loss is ordinary, and a queue that grows is worse +/// than one that spills. +const INBOX: usize = 256; + +/// What a datagram on a data link turned out to be. +#[derive(Debug, Clone, PartialEq, Eq)] +enum Frame { + /// From the peer at the other end of this link, for us. + Direct(Bytes), + /// From the peer at the other end, for somebody else. + ToRelay { to: EndpointId, payload: Bytes }, + /// Passed on by the peer at the other end, from somebody else. + Relayed { from: EndpointId, payload: Bytes }, +} + +fn wrap(tag: u8, id: Option, payload: &[u8]) -> Bytes { + let mut out = BytesMut::with_capacity(1 + 32 + payload.len()); + out.put_u8(tag); + if let Some(id) = id { + out.put_slice(id.as_bytes()); + } + out.put_slice(payload); + out.freeze() +} + +/// Reads a datagram, or `None` if it is not one of ours. +/// +/// Nothing here trusts a length: every field is checked before it is read, +/// because this is bytes off the network like any other. +fn unwrap(raw: Bytes) -> Option { + let tag = *raw.first()?; + match tag { + TAG_DIRECT => Some(Frame::Direct(raw.slice(1..))), + TAG_TO_RELAY | TAG_RELAYED => { + if raw.len() < 1 + 32 { + return None; + } + let mut id = [0u8; 32]; + id.copy_from_slice(&raw[1..33]); + let id = EndpointId::from_bytes(&id).ok()?; + let payload = raw.slice(33..); + Some(if tag == TAG_TO_RELAY { + Frame::ToRelay { to: id, payload } + } else { + Frame::Relayed { from: id, payload } + }) + } + _ => None, + } +} + +/// What has gone through the relay, from both sides of it. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct RelayCounters { + /// Datagrams this agent sent to a peer through somebody else. + pub sent_via: u64, + /// Datagrams this agent passed on for two other peers. + pub forwarded: u64, + /// Datagrams that arrived for this agent through somebody else. + pub received_via: u64, + /// Datagrams that could not be passed on: no link to the destination. + pub dropped_no_link: u64, + /// Datagrams that arrived relayed from a peer we have no link object + /// for, or that could not be read at all. + pub dropped_unknown: u64, +} + +#[derive(Debug, Default)] +struct Tally { + sent_via: AtomicU64, + forwarded: AtomicU64, + received_via: AtomicU64, + dropped_no_link: AtomicU64, + dropped_unknown: AtomicU64, +} + +impl Tally { + fn snapshot(&self) -> RelayCounters { + RelayCounters { + sent_via: self.sent_via.load(Ordering::Relaxed), + forwarded: self.forwarded.load(Ordering::Relaxed), + received_via: self.received_via.load(Ordering::Relaxed), + dropped_no_link: self.dropped_no_link.load(Ordering::Relaxed), + dropped_unknown: self.dropped_unknown.load(Ordering::Relaxed), + } + } +} + +/// Every data link of one network, and what it can reach through what. +/// +/// One per network runtime. It owns the raw links the transport produced +/// and hands protocols a [`PeerLink`] each, which is the thing that knows +/// whether a peer is reached directly or through somebody. +#[derive(Debug)] +pub struct RelayHub { + network: NetworkId, + /// What a protocol holds, one per peer and protocol. Kept alive here so + /// a datagram can be injected into a link the protocol is reading. + links: Mutex>>, + /// The raw links, which is what a hop is: a way to reach the peer in + /// the middle, never wrapped again. + raw: Mutex>, + tally: Arc, +} + +impl RelayHub { + /// A hub for one network. + pub fn new(network: NetworkId) -> Arc { + Arc::new(Self { + network, + links: Mutex::new(HashMap::new()), + raw: Mutex::new(HashMap::new()), + tally: Arc::new(Tally::default()), + }) + } + + /// What has gone through it. + pub fn counters(&self) -> RelayCounters { + self.tally.snapshot() + } + + fn links(&self) -> std::sync::MutexGuard<'_, HashMap<(EndpointId, String), Arc>> { + match self.links.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } + } + + fn raw(&self) -> std::sync::MutexGuard<'_, HashMap<(EndpointId, String), SharedLink>> { + match self.raw.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } + } + + /// The link a protocol uses for a peer, created on first ask. + /// + /// The same object for the life of the peer, whatever happens to the + /// path underneath it: a protocol's tunnel survives a direct link + /// coming and going, and a change of hop, without noticing. + pub fn link(self: &Arc, peer: EndpointId, protocol: &str) -> Arc { + let key = (peer, protocol.to_string()); + let mut links = self.links(); + if let Some(existing) = links.get(&key) { + return Arc::clone(existing); + } + let (inbox_tx, inbox_rx) = mpsc::channel(INBOX); + let link = Arc::new(PeerLink { + network: self.network, + peer, + protocol: protocol.to_string(), + direct: RwLock::new(None), + hop: RwLock::new(None), + inbox_tx, + inbox_rx: tokio::sync::Mutex::new(inbox_rx), + changed: tokio::sync::Notify::new(), + closed: AtomicBool::new(false), + gone: tokio::sync::Notify::new(), + hub: Arc::downgrade(self), + tally: Arc::clone(&self.tally), + }); + links.insert(key, Arc::clone(&link)); + link + } + + /// Whether a protocol already holds a link for this peer. + /// + /// Asked before handing one out, because a protocol is given its link + /// once and keeps it: handing the same peer a second link would leave + /// it with two and a tunnel it is no longer reading for. + pub fn has_link(&self, peer: EndpointId, protocol: &str) -> bool { + self.links().contains_key(&(peer, protocol.to_string())) + } + + /// Records the direct link to a peer, replacing any previous one. + pub fn set_direct(self: &Arc, peer: EndpointId, protocol: &str, raw: SharedLink) { + self.raw().insert((peer, protocol.to_string()), raw.clone()); + let link = self.link(peer, protocol); + link.set_direct(Some(raw)); + } + + /// Forgets the direct link to a peer. The protocol's link stays. + pub fn clear_direct(self: &Arc, peer: EndpointId, protocol: &str) { + self.raw().remove(&(peer, protocol.to_string())); + if let Some(link) = self.links().get(&(peer, protocol.to_string())) { + link.set_direct(None); + } + } + + /// Routes a peer through another peer, or stops doing so. + /// + /// The hop must be a peer with a direct link of its own; anything else + /// is a request to relay through somebody unreachable. + pub fn set_hop(self: &Arc, peer: EndpointId, protocol: &str, hop: Option) { + let resolved = hop.and_then(|hop| { + self.raw() + .get(&(hop, protocol.to_string())) + .map(|link| (hop, Arc::clone(link))) + }); + let link = self.link(peer, protocol); + link.set_hop(resolved); + } + + /// Closes and forgets everything held for a peer. + pub fn remove_peer(&self, peer: EndpointId) { + self.raw().retain(|(other, _), _| *other != peer); + let mut links = self.links(); + links.retain(|(other, _), link| { + if *other == peer { + link.close(); + false + } else { + true + } + }); + } + + /// Closes everything. The network is going away. + pub fn close(&self) { + self.raw().clear(); + for link in self.links().drain() { + link.1.close(); + } + } + + /// Passes a datagram on to the peer it is for. + /// + /// Only over a direct link: a relayed datagram is never relayed again, + /// which is what makes a loop impossible without counting hops. + fn forward(&self, to: EndpointId, protocol: &str, from: EndpointId, payload: Bytes) { + let link = self.raw().get(&(to, protocol.to_string())).cloned(); + let Some(link) = link else { + self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed); + return; + }; + match link.send(wrap(TAG_RELAYED, Some(from), &payload)) { + Ok(()) => { + self.tally.forwarded.fetch_add(1, Ordering::Relaxed); + } + Err(err) => { + tracing::debug!(%err, "cannot pass a datagram on"); + self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed); + } + } + } + + /// Hands a datagram that arrived through somebody to the peer's link. + fn inject(&self, from: EndpointId, protocol: &str, payload: Bytes) { + let link = self.links().get(&(from, protocol.to_string())).cloned(); + let Some(link) = link else { + // Nothing is reading for that peer: it is not one this agent + // carries traffic with, so the datagram has no owner. + self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed); + return; + }; + match link.inbox_tx.try_send(payload) { + Ok(()) => { + self.tally.received_via.fetch_add(1, Ordering::Relaxed); + } + Err(_) => { + self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed); + } + } + } +} + +/// A protocol's link to one peer, however that peer is reached. +/// +/// It outlives any particular path. A direct link that dies, a hop that +/// changes, a direct link that comes back: none of it is visible to the +/// protocol holding this, so its tunnel is not torn down and rebuilt every +/// time the way through changes. +#[derive(Debug)] +pub struct PeerLink { + network: NetworkId, + peer: EndpointId, + protocol: String, + direct: RwLock>, + hop: RwLock>, + inbox_tx: mpsc::Sender, + inbox_rx: tokio::sync::Mutex>, + /// Woken when the path changes, so a reader parked on the old one + /// starts again on the new one. + changed: tokio::sync::Notify, + closed: AtomicBool, + gone: tokio::sync::Notify, + hub: Weak, + tally: Arc, +} + +impl PeerLink { + fn direct(&self) -> Option { + match self.direct.read() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + } + + fn hop(&self) -> Option<(EndpointId, SharedLink)> { + match self.hop.read() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + } + + fn set_direct(&self, link: Option) { + match self.direct.write() { + Ok(mut guard) => *guard = link, + Err(poisoned) => *poisoned.into_inner() = link, + } + self.changed.notify_waiters(); + } + + fn set_hop(&self, hop: Option<(EndpointId, SharedLink)>) { + match self.hop.write() { + Ok(mut guard) => *guard = hop, + Err(poisoned) => *poisoned.into_inner() = hop, + } + self.changed.notify_waiters(); + } + + fn close(&self) { + self.closed.store(true, Ordering::Relaxed); + self.gone.notify_waiters(); + self.changed.notify_waiters(); + } + + /// The peer this link carries traffic for. + pub fn peer(&self) -> EndpointId { + self.peer + } + + /// Which peer it goes through, when it is not direct. + pub fn via(&self) -> Option { + if self.direct().is_some_and(|link| !link.is_closed()) { + return None; + } + self.hop().map(|(peer, _)| peer) + } +} + +impl PacketLink for PeerLink { + fn network(&self) -> NetworkId { + self.network + } + + fn peer(&self) -> EndpointId { + self.peer + } + + fn max_datagram_size(&self) -> usize { + // The same size whatever the path, so a protocol never has to + // resize because the way through changed. + let direct = self.direct().map(|link| link.max_datagram_size()); + let hop = self.hop().map(|(_, link)| link.max_datagram_size()); + let smallest = match (direct, hop) { + (Some(a), Some(b)) => a.min(b), + (Some(a), None) | (None, Some(a)) => a, + (None, None) => 0, + }; + smallest.saturating_sub(RELAY_OVERHEAD) + } + + fn send(&self, payload: Bytes) -> Result<(), TransportError> { + if self.closed.load(Ordering::Relaxed) { + return Err(TransportError::Closed); + } + // Direct while there is one: a hop is what you use when there is + // nothing better, never a preference. + if let Some(direct) = self.direct() + && !direct.is_closed() + { + return direct.send(wrap(TAG_DIRECT, None, &payload)); + } + if let Some((_, hop)) = self.hop() { + let out = hop.send(wrap(TAG_TO_RELAY, Some(self.peer), &payload)); + if out.is_ok() { + self.tally.sent_via.fetch_add(1, Ordering::Relaxed); + } + return out; + } + Err(TransportError::Unreachable(format!( + "no path to {}", + self.peer.fmt_short() + ))) + } + + fn recv(&self) -> BoxFuture<'_, Option> { + Box::pin(async move { + let mut inbox = self.inbox_rx.lock().await; + loop { + if self.closed.load(Ordering::Relaxed) { + return None; + } + let direct = self.direct(); + let changed = self.changed.notified(); + let gone = self.gone.notified(); + + let raw = tokio::select! { + // Whatever arrived through somebody else, already + // stripped of its wrapper by the link it came in on. + injected = inbox.recv() => return injected, + raw = async { + match &direct { + Some(link) => link.recv().await, + // Nothing direct: wait for a path or an injection. + None => std::future::pending().await, + } + } => raw, + // The path changed underneath: look again. + () = changed => continue, + () = gone => return None, + }; + + let Some(raw) = raw else { + // The direct link ended. The peer may still be + // reachable through somebody, so this link is not over. + self.set_direct(None); + continue; + }; + match unwrap(raw) { + Some(Frame::Direct(payload)) => return Some(payload), + // This agent is the one in the middle. + Some(Frame::ToRelay { to, payload }) => { + if let Some(hub) = self.hub.upgrade() { + hub.forward(to, &self.protocol, self.peer, payload); + } + } + // Somebody passed this on for a peer we talk to. + Some(Frame::Relayed { from, payload }) => { + if let Some(hub) = self.hub.upgrade() { + hub.inject(from, &self.protocol, payload); + } + } + None => { + self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed); + } + } + } + }) + } + + fn closed(&self) -> BoxFuture<'_, ()> { + Box::pin(async move { + loop { + if self.closed.load(Ordering::Relaxed) { + return; + } + self.gone.notified().await; + } + }) + } + + fn is_closed(&self) -> bool { + self.closed.load(Ordering::Relaxed) + } + + fn path_description(&self) -> String { + if let Some(direct) = self.direct() + && !direct.is_closed() + { + return direct.path_description(); + } + match self.hop() { + Some((peer, link)) => { + format!("via {} ({})", peer.fmt_short(), link.path_description()) + } + None => "no path".to_string(), + } + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use crate::identity::{NetworkKeys, NetworkName, NetworkSecret}; + + fn network() -> NetworkId { + NetworkKeys::derive( + &NetworkName::new("relay").unwrap(), + &NetworkSecret::from_bytes([3u8; 32]).unwrap(), + ) + .network_id() + } + + fn peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + /// A link that records what was sent and can be fed what arrives. + #[derive(Debug)] + struct Wire { + network: NetworkId, + peer: EndpointId, + sent: Mutex>, + inbound: tokio::sync::Mutex>, + feed: mpsc::Sender, + closed: AtomicBool, + } + + impl Wire { + fn new(peer: EndpointId) -> Arc { + let (feed, inbound) = mpsc::channel(32); + Arc::new(Self { + network: network(), + peer, + sent: Mutex::new(Vec::new()), + inbound: tokio::sync::Mutex::new(inbound), + feed, + closed: AtomicBool::new(false), + }) + } + + fn sent(&self) -> Vec { + self.sent.lock().unwrap().clone() + } + + async fn arrive(&self, raw: Bytes) { + self.feed.send(raw).await.unwrap(); + } + } + + impl PacketLink for Wire { + fn network(&self) -> NetworkId { + self.network + } + fn peer(&self) -> EndpointId { + self.peer + } + fn max_datagram_size(&self) -> usize { + 1200 + } + fn send(&self, payload: Bytes) -> Result<(), TransportError> { + if self.closed.load(Ordering::Relaxed) { + return Err(TransportError::Closed); + } + self.sent.lock().unwrap().push(payload); + Ok(()) + } + fn recv(&self) -> BoxFuture<'_, Option> { + Box::pin(async move { self.inbound.lock().await.recv().await }) + } + fn closed(&self) -> BoxFuture<'_, ()> { + Box::pin(async move { std::future::pending().await }) + } + fn is_closed(&self) -> bool { + self.closed.load(Ordering::Relaxed) + } + fn path_description(&self) -> String { + "test wire".into() + } + } + + #[test] + fn a_datagram_survives_the_wrapping_and_a_broken_one_is_refused() { + let id = peer(1); + let direct = wrap(TAG_DIRECT, None, b"hello"); + assert_eq!(unwrap(direct).unwrap(), Frame::Direct(Bytes::from("hello"))); + + let onward = wrap(TAG_TO_RELAY, Some(id), b"hello"); + assert_eq!( + unwrap(onward).unwrap(), + Frame::ToRelay { + to: id, + payload: Bytes::from("hello") + } + ); + + // Bytes off the network: nothing is read before it is checked. + assert!(unwrap(Bytes::new()).is_none()); + assert!(unwrap(Bytes::from_static(&[TAG_TO_RELAY, 1, 2, 3])).is_none()); + assert!(unwrap(Bytes::from_static(&[200, 1, 2, 3])).is_none()); + } + + #[tokio::test] + async fn a_direct_link_is_preferred_and_a_hop_is_used_when_there_is_none() { + let hub = RelayHub::new(network()); + let (them, middle) = (peer(1), peer(2)); + + let to_middle = Wire::new(middle); + hub.set_direct(middle, "test-ip", to_middle.clone()); + + // No direct link to them: nothing to send on, and nothing invented. + let link = hub.link(them, "test-ip"); + assert!(link.send(Bytes::from("x")).is_err()); + assert_eq!(link.via(), None); + + // Routed through the middle: the datagram goes out on that link, + // wrapped with who it is for. + hub.set_hop(them, "test-ip", Some(middle)); + assert_eq!(link.via(), Some(middle)); + link.send(Bytes::from("through you")).unwrap(); + let sent = to_middle.sent(); + assert_eq!(sent.len(), 1); + assert_eq!( + unwrap(sent[0].clone()).unwrap(), + Frame::ToRelay { + to: them, + payload: Bytes::from("through you") + } + ); + assert_eq!(hub.counters().sent_via, 1); + + // A direct link appears: it is used, and the hop is not. + let to_them = Wire::new(them); + hub.set_direct(them, "test-ip", to_them.clone()); + assert_eq!(link.via(), None); + link.send(Bytes::from("straight")).unwrap(); + assert_eq!( + unwrap(to_them.sent()[0].clone()).unwrap(), + Frame::Direct(Bytes::from("straight")) + ); + assert_eq!(to_middle.sent().len(), 1, "nothing more went the long way"); + } + + #[tokio::test] + async fn the_one_in_the_middle_passes_it_on_without_reading_it() { + // C between A and B. A's datagram arrives wrapped for B; C puts it + // on its link to B, saying who it came from, and never sees inside. + let hub = RelayHub::new(network()); + let (a, b) = (peer(1), peer(2)); + let from_a = Wire::new(a); + let to_b = Wire::new(b); + hub.set_direct(a, "test-ip", from_a.clone()); + hub.set_direct(b, "test-ip", to_b.clone()); + + let a_link = hub.link(a, "test-ip"); + let reading = tokio::spawn(async move { a_link.recv().await }); + + from_a.arrive(wrap(TAG_TO_RELAY, Some(b), b"opaque")).await; + // It is passed on rather than returned to the protocol here. + tokio::time::timeout(std::time::Duration::from_millis(200), async { + while to_b.sent().is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .expect("it was passed on"); + assert_eq!( + unwrap(to_b.sent()[0].clone()).unwrap(), + Frame::Relayed { + from: a, + payload: Bytes::from("opaque") + } + ); + assert_eq!(hub.counters().forwarded, 1); + assert!(!reading.is_finished(), "the protocol was not handed it"); + + // Nowhere to put it is a drop and a count, never an error upwards. + from_a + .arrive(wrap(TAG_TO_RELAY, Some(peer(9)), b"nobody")) + .await; + tokio::time::timeout(std::time::Duration::from_millis(200), async { + while hub.counters().dropped_no_link == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("counted"); + reading.abort(); + } + + #[tokio::test] + async fn what_arrives_through_somebody_is_handed_to_the_right_peer() { + // B's side: a datagram from A arrives on the link with C, and the + // protocol reading A's link is the one that gets it — otherwise + // the packet would be attributed to C and dropped as coming from + // an address C does not hold. + let hub = RelayHub::new(network()); + let (a, c) = (peer(1), peer(3)); + let from_c = Wire::new(c); + hub.set_direct(c, "test-ip", from_c.clone()); + let a_link = hub.link(a, "test-ip"); + let c_link = hub.link(c, "test-ip"); + + let reading_a = tokio::spawn(async move { a_link.recv().await }); + let driving_c = tokio::spawn(async move { c_link.recv().await }); + + from_c.arrive(wrap(TAG_RELAYED, Some(a), b"from a")).await; + let got = tokio::time::timeout(std::time::Duration::from_secs(2), reading_a) + .await + .expect("delivered") + .unwrap(); + assert_eq!(got, Some(Bytes::from("from a"))); + assert_eq!(hub.counters().received_via, 1); + driving_c.abort(); + } + + #[tokio::test] + async fn a_link_outlives_the_path_under_it() { + // The protocol's tunnel must not be torn down because a path + // changed: the link object is the peer, not the way to it. + let hub = RelayHub::new(network()); + let (them, middle) = (peer(1), peer(2)); + let direct = Wire::new(them); + hub.set_direct(them, "test-ip", direct.clone()); + let link = hub.link(them, "test-ip"); + assert!(!link.is_closed()); + + hub.clear_direct(them, "test-ip"); + assert!( + !link.is_closed(), + "still the same peer, still the same link" + ); + assert!( + link.send(Bytes::from("x")).is_err(), + "but nothing to send on" + ); + + let hop = Wire::new(middle); + hub.set_direct(middle, "test-ip", hop.clone()); + hub.set_hop(them, "test-ip", Some(middle)); + link.send(Bytes::from("x")).unwrap(); + assert_eq!(hop.sent().len(), 1); + + // The peer going away is what closes it. + hub.remove_peer(them); + assert!(link.is_closed()); + assert!(link.send(Bytes::from("x")).is_err()); + } + + #[tokio::test] + async fn the_size_a_protocol_may_use_does_not_change_with_the_path() { + let hub = RelayHub::new(network()); + let (them, middle) = (peer(1), peer(2)); + hub.set_direct(them, "test-ip", Wire::new(them)); + hub.set_direct(middle, "test-ip", Wire::new(middle)); + let link = hub.link(them, "test-ip"); + let direct_size = link.max_datagram_size(); + assert_eq!(direct_size, 1200 - RELAY_OVERHEAD); + + hub.set_hop(them, "test-ip", Some(middle)); + hub.clear_direct(them, "test-ip"); + assert_eq!( + link.max_datagram_size(), + direct_size, + "a tunnel that resized on every path change would be worse" + ); + } +} diff --git a/crates/tsunagi/src/ipc/mod.rs b/crates/tsunagi/src/ipc/mod.rs index 264c6e4..cfc8097 100644 --- a/crates/tsunagi/src/ipc/mod.rs +++ b/crates/tsunagi/src/ipc/mod.rs @@ -274,6 +274,12 @@ pub struct NetworkReport { /// has no way to reach anybody" — which look identical in a report /// that counts only members. pub candidates: u32, + /// Datagrams this agent passed on between two other peers. + pub relay_forwarded: u64, + /// Datagrams this agent sent to a peer through somebody else. + pub relay_sent_via: u64, + /// Datagrams that reached this agent through somebody else. + pub relay_received_via: u64, /// Outbound dials that failed. pub dial_failures: u64, /// Handshakes rejected in either direction. diff --git a/crates/tsunagi/src/ipc/unix.rs b/crates/tsunagi/src/ipc/unix.rs index de664e5..e385adf 100644 --- a/crates/tsunagi/src/ipc/unix.rs +++ b/crates/tsunagi/src/ipc/unix.rs @@ -367,7 +367,7 @@ pub async fn set_dns( /// /// Bump it whenever [`Request`], [`Response`] or anything they contain /// changes shape. -pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 12]); +pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 13]); async fn write_message(stream: &mut UnixStream, value: &T) -> Result<()> { let encoded = postcard::to_stdvec(value) diff --git a/crates/tsunagi/src/proto/message.rs b/crates/tsunagi/src/proto/message.rs index 15384fb..9e708bf 100644 --- a/crates/tsunagi/src/proto/message.rs +++ b/crates/tsunagi/src/proto/message.rs @@ -28,7 +28,12 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/1"; /// They carry one IP plugin's packets for one network and nothing else, so a /// saturated or broken data plane cannot disturb control traffic, and the /// transport underneath can be replaced without touching the control protocol. -pub const DATA_ALPN: &[u8] = b"tsunagi/data/1"; +/// +/// Version 2 puts a tag on every datagram, so one can say "this is for +/// somebody else" and be passed on by the peer in the middle. An agent +/// speaking version 1 simply does not form a data link with one speaking +/// version 2, which is what the version in an ALPN is for. +pub const DATA_ALPN: &[u8] = b"tsunagi/data/2"; /// Largest plugin protocol identifier accepted when opening a data channel. pub const MAX_DATA_PROTOCOL_LEN: usize = 32; @@ -124,6 +129,20 @@ pub enum ControlMessage { /// The records. Bounded by [`crate::config::Limits::max_state_records`]. records: Vec, }, + /// Which peers this sender has a live data link with, right now. + /// + /// First-hand and nothing else: a sender speaks only for itself, never + /// about what somebody else can reach. That is what makes it usable + /// without weighing hearsay — the claim is proved or disproved by + /// sending through it. + /// + /// A snapshot rather than a change, because a snapshot is idempotent + /// and repairs itself; and volatile, so it lives here and not in + /// signed state, which is for what has to survive a member being away. + Reachable { + /// The peers, as raw endpoint keys. + peers: Vec<[u8; 32]>, + }, /// Graceful goodbye. /// /// A peer going away is not a revocation of anything. @@ -191,6 +210,11 @@ pub fn validate_capability( /// network, the other networks or the agent. pub fn validate(message: &ControlMessage, limits: &Limits) -> Result<(), ProtocolError> { match message { + ControlMessage::Reachable { peers } => { + // One entry per member at most; a sender claiming more than a + // network can hold is not describing this network. + check_len("reachable.peers", peers.len(), limits.max_state_records)?; + } ControlMessage::Announce(announcement) => { check_len( "announce.hostname", @@ -229,6 +253,7 @@ pub fn kind(message: &ControlMessage) -> &'static str { ControlMessage::Ping { .. } => "ping", ControlMessage::Pong { .. } => "pong", ControlMessage::State { .. } => "state", + ControlMessage::Reachable { .. } => "reachable", ControlMessage::Bye { .. } => "bye", } } diff --git a/docs/testing.md b/docs/testing.md index dd36c7a..db2c92d 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -55,6 +55,16 @@ keeping the WireGuard identity, shutdown removing every interface, a forged overlay claim being rejected, and the core carrying the payload without interpreting it. +`crates/tsunagi/src/dataplane/relay.rs` has its own tests for the way +through a peer in the middle: the wrapping and what a malformed one does, a +direct path being preferred over a hop, the middle passing a datagram on +without being handed it, what arrives through somebody reaching the peer it +came from rather than the one that carried it, a link outliving the paths +under it, and the datagram size not changing when the path does. +`tests/wireguard.rs` proves it end to end — two agents that can each reach a +third and not each other, with a real WireGuard packet crossing through the +middle. + Unit tests in `crates/tsunagi/src/state/` cover the signed record model directly: tampering with any field breaks verification, a newer version wins while an older one never rolls back, two authors claiming one address resolve the same way no