diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 1570244..a80b34e 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -1884,17 +1884,24 @@ async fn up(args: UpArgs) -> Result<(), Box> { // The data plane is optional and never required for the control plane. let wireguard = if args.wireguard { + // The interface belongs to the agent, not to the protocol: one + // agent, one interface, and every protocol carries traffic for the + // same addresses on it. let tun_factory: Arc = if args.no_tun { Arc::new(MemoryTunFactory::new()) } else { system_tun_factory()? }; - let mut wg = WireguardConfig::new(paths.state_dir.join("wireguard")) - .with_interface_prefix(args.wg_prefix.clone()); + let mtu = args + .wg_mtu + .unwrap_or(tsunagi::dataplane::wireguard::DEFAULT_MTU); + config = config.with_interface(tun_factory, args.wg_prefix.clone(), mtu); + + let mut wg = WireguardConfig::new(paths.state_dir.join("wireguard")); if let Some(mtu) = args.wg_mtu { wg = wg.with_mtu(mtu); } - let plugin = WireguardPlugin::open(wg, tun_factory).await?; + let plugin = WireguardPlugin::open(wg).await?; config = config.with_plugin(plugin.clone() as Arc); Some(plugin) } else { @@ -2030,6 +2037,7 @@ async fn build_report( StatusReport, }; + let overlay = agent.overlay(); let dns = dns.map(|dns| DnsReport { zone: dns.zone, listening: dns.listening.map(|address| address.to_string()), @@ -2051,8 +2059,10 @@ async fn build_report( let overlay = wireguard .and_then(|plugin| plugin.overview(network.network_id)) .map(|view| OverlayReport { - interface: view.interface.clone(), - mtu: view.mtu, + interface: overlay + .as_ref() + .map_or_else(String::new, |overlay| overlay.interface.clone()), + mtu: overlay.as_ref().map_or(0, |overlay| overlay.mtu), address: view.overlay_address_v4.map(|addr| addr.to_string()), prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len), peers: view @@ -2089,9 +2099,21 @@ async fn build_report( .unwrap_or_else(|| "no data link".into()), }) .collect(), - unroutable_packets: view.unroutable_packets, - multicast_packets: view.multicast_packets, - unroutable_sample: view.unroutable_sample.map(|address| address.to_string()), + // The interface belongs to the agent, so the counters + // about it come from there and are the same for every + // network sharing it. + unroutable_packets: overlay + .as_ref() + .map_or(0, |overlay| overlay.counters.unroutable), + multicast_packets: overlay + .as_ref() + .map_or(0, |overlay| overlay.counters.multicast), + unroutable_sample: overlay.as_ref().and_then(|overlay| { + overlay + .counters + .unroutable_sample + .map(|address| address.to_string()) + }), }); NetworkReport { @@ -2309,12 +2331,18 @@ async fn print_status(agent: &Agent, network: NetworkId, wireguard: Option<&Wire ), } } - if view.unroutable_packets > 0 { - println!( - " {} packet(s) for unknown addresses", - view.unroutable_packets - ); - } + } + if let Some(overlay) = agent.overlay() + && overlay.counters.unroutable > 0 + { + println!( + " {} packet(s) for unknown addresses{}", + overlay.counters.unroutable, + match overlay.counters.unroutable_sample { + Some(sample) => format!(" (for example {sample})"), + None => String::new(), + } + ); } println!(); } diff --git a/crates/tsunagi/src/agent/mod.rs b/crates/tsunagi/src/agent/mod.rs index aecfb6f..e07acb8 100644 --- a/crates/tsunagi/src/agent/mod.rs +++ b/crates/tsunagi/src/agent/mod.rs @@ -27,7 +27,7 @@ mod status; pub use events::Event; pub use status::{ AgentStatus, CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus, - PeerStatus, + OverlayStatus, PeerStatus, }; use std::collections::HashMap; @@ -37,13 +37,15 @@ use iroh::{EndpointAddr, EndpointId}; use tokio::sync::{RwLock, broadcast, mpsc, oneshot}; use tokio::task::JoinHandle; +use crate::BoxFuture; use crate::config::{AgentConfig, Limits}; use crate::dataplane::transport::PacketTransport; use crate::dataplane::transport::iroh_link::{IrohTransport, TransportContext}; -use crate::dataplane::{PluginContext, PluginRequest}; +use crate::dataplane::{PacketSink, PluginContext, PluginRequest}; use crate::error::{Error, Result}; use crate::identity::{DeviceIdentity, NetworkId, NetworkKeys, NetworkName, NetworkSecret}; use crate::net::EndpointAdapter; +use crate::overlay::PacketCarrier; use crate::proto::handshake; use crate::proto::message::ControlMessage; use crate::storage::{CacheOutcome, Storage}; @@ -87,6 +89,10 @@ struct Inner { accept_task: std::sync::Mutex>>, plugin_task: std::sync::Mutex>>, transport: std::sync::OnceLock>, + /// Who holds which overlay address, across every network. + routes: Arc, + /// The one interface, when the agent was given a way to make one. + interface: std::sync::OnceLock>, } impl Inner { @@ -107,11 +113,69 @@ impl Inner { } } +/// Routes an outbound packet to whichever protocol can carry it. +/// +/// Holds a weak reference: the interface belongs to the agent, and a strong +/// one here would keep the agent alive for as long as its own interface. +struct Carrier(Weak); + +impl std::fmt::Debug for Carrier { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Carrier") + } +} + +impl std::fmt::Debug for Sink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Sink") + } +} + +impl PacketCarrier for Carrier { + fn carry(&self, route: crate::overlay::Route, packet: bytes::Bytes) -> bool { + let Some(inner) = self.0.upgrade() else { + return false; + }; + // Offered to each protocol in turn. With one configured this is a + // single call; the shape is what allows several to be live at once, + // each carrying the peers it has a link to. + inner + .config + .plugins + .iter() + .any(|plugin| plugin.carry(route.network, route.peer, packet.clone())) + } +} + +/// Writes a packet a protocol decrypted to the interface. +struct Sink(Weak); + +impl PacketSink for Sink { + fn deliver<'a>( + &'a self, + network: NetworkId, + peer: EndpointId, + packet: bytes::Bytes, + ) -> BoxFuture<'a, ()> { + Box::pin(async move { + let Some(inner) = self.0.upgrade() else { + return; + }; + let Some(interface) = inner.interface.get() else { + return; + }; + // The rejection is already counted on the interface; there is + // nothing useful to do with it here. + let _ = interface.deliver(network, peer, packet).await; + }) + } +} + /// Answers the data plane transport's questions about the agent. /// -/// Holds a weak reference on purpose: the transport lives inside the agent, so -/// a strong one would be a cycle and the agent — with its open databases and -/// its directory lock — would never be released. +/// Holds a weak reference for the same reason as [`Carrier`]: the transport +/// lives inside the agent, so a strong one would be a cycle and the agent — +/// with its open databases and its directory lock — would never be released. #[derive(Debug)] struct TransportCtx(Weak); @@ -178,9 +242,38 @@ impl Agent { accept_task: std::sync::Mutex::new(None), plugin_task: std::sync::Mutex::new(None), transport: std::sync::OnceLock::new(), + routes: Arc::new(crate::overlay::RoutingTable::new()), + interface: std::sync::OnceLock::new(), config, }); + // One interface for the agent, if it was given a way to make one. + // Failing to is reported and not fatal: the control plane works, and + // so do the protocols, they just have nowhere to put packets. + if let Some(factory) = inner.config.tun_factory.clone() { + let carrier = Arc::new(Carrier(Arc::downgrade(&inner))) as Arc; + match crate::overlay::Interface::start( + factory, + inner.config.interface_name.clone(), + inner.config.interface_mtu, + Arc::clone(&inner.routes), + carrier, + ) + .await + { + Ok(interface) => { + let _ = inner.interface.set(Arc::new(interface)); + } + Err(err) => { + let _ = inner.events.send(Event::PluginError { + network: NetworkId::from_bytes([0u8; 32]), + protocol: "overlay".into(), + reason: err.to_string(), + }); + } + } + } + // The data plane rides on iroh too, which is where it gets hole // punching and relay fallback from. It is a separate ALPN and a // separate connection, so the two planes stay independent. @@ -205,7 +298,11 @@ impl Agent { // bound; overflow drops the request rather than stalling the plugin. if !inner.config.plugins.is_empty() { let (plugin_tx, plugin_rx) = mpsc::channel(64); - let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id()); + let sink = inner + .interface + .get() + .map(|_| Arc::new(Sink(Arc::downgrade(&inner))) as Arc); + let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink); for plugin in &inner.config.plugins { plugin.attach(context.clone()); } @@ -245,6 +342,47 @@ impl Agent { self.inner.adapter.loopback_addr() } + /// Reserves the configured overlay range for a network, if it can. + /// + /// Done here, where activations are serialised, rather than inside the + /// runtime: whether a range is free depends on what the other networks + /// took, and deciding that in a task would make the answer depend on + /// which task ran first. + /// + /// `None` means this agent will not propose a range for that network and + /// waits to adopt whatever it settles on. One agent has one interface, so + /// proposing a range it could not route would be worse than having none: + /// the lowest author's range wins, and the collision would spread. + fn reserve_range(&self, network: NetworkId) -> Option { + let wanted = self.inner.config.overlay_ipv4_range?; + let reservation = crate::overlay::NetworkRoutes { + range: Some(wanted), + local: None, + peers: Vec::new(), + }; + match self.inner.routes.set_network(network, reservation) { + Ok(()) => Some(wanted), + Err(err) => { + tracing::info!(%err, "not proposing a range for this network"); + None + } + } + } + + /// The overlay interface this agent owns, when it has one. + /// + /// One agent, one interface, so this is not asked per network: a packet + /// on it may belong to any of them. + pub fn overlay(&self) -> Option { + let interface = self.inner.interface.get()?; + Some(OverlayStatus { + interface: interface.name().to_string(), + mtu: interface.mtu(), + addresses: interface.wanted_addresses(), + counters: interface.counters(), + }) + } + /// The hostname announced to peers. pub fn hostname(&self) -> String { self.inner.read_hostname() @@ -367,10 +505,12 @@ impl Agent { discovery: self.inner.config.discovery.clone(), discovery_interval: self.inner.config.discovery_interval, plugins: self.inner.config.plugins.clone(), + routes: Arc::clone(&self.inner.routes), + interface: self.inner.interface.get().cloned(), hostname: self.inner.read_hostname(), transport: self.inner.transport.get().cloned(), device_secret: self.inner.identity.signing_key(), - ipv4_range: self.inner.config.overlay_ipv4_range, + ipv4_range: self.reserve_range(network_id), }); networks.insert(network_id, handle); drop(networks); @@ -401,6 +541,15 @@ impl Agent { for plugin in &self.inner.config.plugins { plugin.on_network_deactivated(network_id); } + // The network's addresses come off the interface; the interface + // itself stays, because the agent owns it and other networks may + // still be using it. + self.inner.routes.remove_network(network_id); + if let Some(interface) = self.inner.interface.get() + && let Err(err) = interface.sync_addresses().await + { + tracing::warn!(%err, "cannot update the overlay addresses"); + } self.inner.storage.set_auto_start(network_id, false).await?; Ok(()) } @@ -583,6 +732,12 @@ impl Agent { } } + // The interface goes with the agent that owns it. Before the + // plugins, so nothing is still trying to write to it. + if let Some(interface) = self.inner.interface.get() { + interface.remove().await; + } + // Plugins remove whatever system objects they created. A plugin that // misbehaves here must not hold up the agent, so this is bounded. for plugin in &self.inner.config.plugins { diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 7646891..8e19222 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -120,6 +120,10 @@ pub(crate) struct RuntimeParams { pub(crate) discovery: Option>, pub(crate) discovery_interval: Duration, pub(crate) plugins: Vec, + /// Who holds which overlay address, shared with every other network. + pub(crate) routes: Arc, + /// The one interface, when the agent has one. + pub(crate) interface: Option>, pub(crate) hostname: String, /// Signing key for this agent's own records. pub(crate) device_secret: iroh::SecretKey, @@ -615,7 +619,7 @@ impl Runtime { .unwrap_or(0); self.ensure_own_claim().await; - self.publish_allocations(); + self.publish_allocations().await; } /// The range this network uses: whatever it has already settled on, else @@ -624,7 +628,20 @@ impl Runtime { /// Adopting the agreed one is what lets a participant join without being /// told the range out of band. fn effective_range(&self) -> Option { - self.state.agreed_range().or(self.params.ipv4_range) + if let Some(agreed) = self.state.agreed_range() { + return Some(agreed); + } + // Nobody has settled one yet, so this agent would be proposing its + // own. It must not propose a range it cannot route: one agent has + // one interface, and claiming an address in a range another network + // already owns would spread the collision rather than contain it — + // "the lowest author's range wins" would carry it to everybody. + // Better to hold off and adopt whatever the network settles on. + let wanted = self.params.ipv4_range?; + match self.params.routes.would_overlap(self.network_id, wanted) { + None => Some(wanted), + Some(_) => None, + } } /// Makes sure this agent holds an address, claiming one if it does not. @@ -810,14 +827,14 @@ impl Runtime { // Somebody may have taken the address we were using. self.ensure_own_claim().await; - self.publish_allocations(); + self.publish_allocations().await; if self.state.records() != before { self.broadcast_state(); } } /// Tells the plugins who holds which overlay address. - fn publish_allocations(&mut self) { + async fn publish_allocations(&mut self) { let Some(range) = self.effective_range() else { return; }; @@ -832,6 +849,40 @@ impl Runtime { for plugin in &self.params.plugins { plugin.on_address_allocation(self.network_id, range, &allocations); } + + // And the system level's own view, which is what decides whose + // packet is whose. The local address is kept out of the peer list: + // a packet for ourselves does not go over a tunnel. + let local = self.state.address_of(&self.local_id); + let routes = crate::overlay::NetworkRoutes { + range: Some(range), + local, + peers: allocations + .iter() + .filter(|(holder, _)| *holder != self.local_id) + .map(|(holder, address)| (*address, *holder)) + .collect(), + }; + if let Err(err) = self.params.routes.set_network(self.network_id, routes) { + // Reported once per change rather than swallowed: two networks + // wanting the same addresses is a thing the user has to settle. + self.metrics.plugin_errors += 1; + self.emit(Event::PluginError { + network: self.network_id, + protocol: "overlay".into(), + reason: err.to_string(), + }); + return; + } + if let Some(interface) = &self.params.interface + && let Err(err) = interface.sync_addresses().await + { + self.emit(Event::PluginError { + network: self.network_id, + protocol: "overlay".into(), + reason: err.to_string(), + }); + } } // ------------------------------------------------------------ data plane diff --git a/crates/tsunagi/src/agent/status.rs b/crates/tsunagi/src/agent/status.rs index 96b4e2a..cf0446b 100644 --- a/crates/tsunagi/src/agent/status.rs +++ b/crates/tsunagi/src/agent/status.rs @@ -36,6 +36,19 @@ pub struct CandidateStatus { pub consecutive_failures: u32, } +/// The overlay interface, as the agent sees it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OverlayStatus { + /// The interface name the operating system gave. + pub interface: String, + /// Its MTU. + pub mtu: u32, + /// The addresses it should be carrying. + pub addresses: Vec, + /// What has happened on it. + pub counters: crate::overlay::Counters, +} + /// A member the signed state knows about, connected or not. /// /// This is the durable roster: it comes from signed records, so a member that diff --git a/crates/tsunagi/src/config.rs b/crates/tsunagi/src/config.rs index a203fda..6f16339 100644 --- a/crates/tsunagi/src/config.rs +++ b/crates/tsunagi/src/config.rs @@ -237,6 +237,17 @@ pub struct AgentConfig { pub reconnect: ReconnectPolicy, /// IP plugins whose capabilities are announced and dispatched. pub plugins: Vec, + /// Where the overlay interface comes from, when the agent should have one. + /// + /// One agent, one interface: it belongs here rather than to a protocol, + /// because every protocol carries traffic for the same addresses on it. + /// `None` means no interface — the protocols still run and their packets + /// are discarded. + pub tun_factory: Option>, + /// The name to ask the operating system for. + pub interface_name: String, + /// The interface MTU. + pub interface_mtu: u32, /// The IPv4 overlay range this agent proposes. /// /// Addresses are allocated from it and recorded in signed state, so a @@ -259,6 +270,9 @@ impl AgentConfig { limits: Limits::default(), reconnect: ReconnectPolicy::default(), plugins: Vec::new(), + tun_factory: None, + interface_name: "tsun0".to_string(), + interface_mtu: 1280, overlay_ipv4_range: Some(crate::state::DEFAULT_IPV4_RANGE), } } @@ -314,6 +328,19 @@ impl AgentConfig { self } + /// Gives the agent an overlay interface, from this factory. + pub fn with_interface( + mut self, + factory: Arc, + name: impl Into, + mtu: u32, + ) -> Self { + self.tun_factory = Some(factory); + self.interface_name = name.into(); + self.interface_mtu = mtu; + self + } + /// Replaces the limits. pub fn with_limits(mut self, limits: Limits) -> Self { self.limits = limits; diff --git a/crates/tsunagi/src/dataplane/mod.rs b/crates/tsunagi/src/dataplane/mod.rs index 83d676e..7960ab3 100644 --- a/crates/tsunagi/src/dataplane/mod.rs +++ b/crates/tsunagi/src/dataplane/mod.rs @@ -52,6 +52,38 @@ pub struct PluginCapability { pub data: Vec, } +/// Where a protocol hands the packets it has decrypted. +/// +/// A protocol proves *who* sent a packet; it does not know what that member +/// is entitled to say, because entitlement is an address claim the system +/// level holds. So a decrypted packet goes here rather than straight to an +/// interface, and is checked on the way. +pub trait PacketSink: Send + Sync + 'static { + /// Hands over one packet, attributed to the peer whose tunnel decrypted + /// it. + fn deliver<'a>( + &'a self, + network: NetworkId, + peer: EndpointId, + packet: bytes::Bytes, + ) -> BoxFuture<'a, ()>; +} + +/// A sink that drops everything, for a protocol running without an interface. +#[derive(Debug, Clone, Copy, Default)] +pub struct DiscardPackets; + +impl PacketSink for DiscardPackets { + fn deliver<'a>( + &'a self, + _network: NetworkId, + _peer: EndpointId, + _packet: bytes::Bytes, + ) -> BoxFuture<'a, ()> { + Box::pin(async move {}) + } +} + /// Errors a plugin may return. They are recorded, never fatal for the agent. #[derive(Debug, thiserror::Error)] #[non_exhaustive] @@ -109,13 +141,32 @@ pub(crate) enum PluginRequest { pub struct PluginContext { sender: Option>, local: Option, + /// Where decrypted packets go. Absent when nothing is carrying traffic, + /// in which case a protocol still runs and its packets are discarded. + sink: Option>, } impl PluginContext { - pub(crate) fn new(sender: mpsc::Sender, local: EndpointId) -> Self { + pub(crate) fn new( + sender: mpsc::Sender, + local: EndpointId, + sink: Option>, + ) -> Self { Self { sender: Some(sender), local: Some(local), + sink, + } + } + + /// Where to hand a decrypted packet. + /// + /// Always answers: with no interface to write to, the packets are + /// discarded, which is what `--no-tun` means and is not an error. + pub fn packet_sink(&self) -> Arc { + match &self.sink { + Some(sink) => Arc::clone(sink), + None => Arc::new(DiscardPackets), } } @@ -124,6 +175,7 @@ impl PluginContext { Self { sender: None, local: None, + sink: None, } } @@ -238,6 +290,20 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static { let _ = (network, range, allocations); } + /// Carries one packet to a peer, encrypting it however this protocol + /// does. + /// + /// `false` when it cannot right now — no link, no tunnel, not this + /// protocol's peer — which the caller reports rather than treats as an + /// error. The packet came off the one interface the agent owns, and + /// which protocol takes it is settled by asking. + /// + /// The default carries nothing, which is right for a plugin that only + /// announces something. + fn carry(&self, _network: NetworkId, _peer: EndpointId, _packet: bytes::Bytes) -> bool { + false + } + /// A data plane link to a peer is available for this plugin's protocol. /// /// The plugin moves its packets over this link and never learns how the diff --git a/crates/tsunagi/src/dataplane/wireguard/device.rs b/crates/tsunagi/src/dataplane/wireguard/device.rs index a13e4b2..5703a7b 100644 --- a/crates/tsunagi/src/dataplane/wireguard/device.rs +++ b/crates/tsunagi/src/dataplane/wireguard/device.rs @@ -29,7 +29,7 @@ //! against. use std::collections::HashMap; -use std::net::{IpAddr, Ipv4Addr}; +use std::net::Ipv4Addr; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use std::time::Duration; @@ -39,13 +39,11 @@ use bytes::Bytes; use iroh::EndpointId; use tokio::task::JoinHandle; -use crate::dataplane::PluginError; use crate::dataplane::transport::{SharedLink, TransportError}; +use crate::dataplane::{PacketSink, PluginError}; use crate::identity::NetworkId; use super::keys::{WgPublicKey, WgSecretKey}; -use crate::overlay::packet::IpHeader; -use crate::overlay::tun::TunDevice; /// How often WireGuard's own timers are driven. /// @@ -65,6 +63,8 @@ struct PeerCounters { rx_bytes: AtomicU64, dropped_wrong_source: AtomicU64, dropped_oversize: AtomicU64, + /// Packets there was no session to encrypt with yet. + dropped_no_session: AtomicU64, protocol_errors: AtomicU64, } @@ -85,6 +85,11 @@ pub struct PeerStats { pub dropped_wrong_source: u64, /// Packets dropped because they did not fit in one link datagram. pub dropped_oversize: u64, + /// Packets dropped because there was no session to encrypt with yet. + /// + /// A handful while a tunnel comes up is normal; a number that keeps + /// climbing means the handshake is not completing. + pub dropped_no_session: u64, /// WireGuard protocol errors, including packets that failed to decrypt. pub protocol_errors: u64, } @@ -149,6 +154,7 @@ impl Peer { rx_bytes: self.counters.rx_bytes.load(Ordering::Relaxed), dropped_wrong_source: self.counters.dropped_wrong_source.load(Ordering::Relaxed), dropped_oversize: self.counters.dropped_oversize.load(Ordering::Relaxed), + dropped_no_session: self.counters.dropped_no_session.load(Ordering::Relaxed), protocol_errors: self.counters.protocol_errors.load(Ordering::Relaxed), } } @@ -186,29 +192,25 @@ pub struct PeerSummary { struct Inner { network: NetworkId, private_key: WgSecretKey, - tun: Arc, - /// The IPv4 overlay range, when the overlay is dual stack. peers: RwLock>>, - /// Both families, so one lookup routes any packet. - routes: RwLock>, + /// Where decrypted packets go. + /// + /// The interface belongs to the system level, so this hands a packet up + /// rather than writing it out: only that level knows which addresses the + /// sending member is entitled to use. + sink: Arc, next_index: AtomicU32, - unroutable: AtomicU64, - /// One destination nobody owned, kept so the counter can be acted on. - unroutable_sample: Mutex>, - multicast: AtomicU64, - ipv4_conflicts: AtomicU64, } impl std::fmt::Debug for Inner { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Inner") .field("network", &self.network.fmt_short()) - .field("tun", &self.tun.name()) .finish() } } -/// A userspace WireGuard interface for one network. +/// The userspace WireGuard tunnels of one network. #[derive(Debug)] pub struct WireguardDevice { inner: Arc, @@ -216,38 +218,73 @@ pub struct WireguardDevice { } impl WireguardDevice { - /// Starts a device on top of `tun`. - pub fn start(network: NetworkId, private_key: WgSecretKey, tun: Arc) -> Self { + /// Starts the tunnels for one network. + /// + /// No interface is involved: packets arrive through [`Self::carry`] and + /// leave through the sink. Which address belongs to whom, and therefore + /// where a packet should go, is decided above this. + pub fn start(network: NetworkId, private_key: WgSecretKey, sink: Arc) -> Self { let inner = Arc::new(Inner { network, private_key, - tun, peers: RwLock::new(HashMap::new()), - routes: RwLock::new(HashMap::new()), + sink, next_index: AtomicU32::new(1), - unroutable: AtomicU64::new(0), - unroutable_sample: Mutex::new(None), - multicast: AtomicU64::new(0), - ipv4_conflicts: AtomicU64::new(0), }); - let reader = tokio::spawn(read_from_os(Arc::clone(&inner))); let timers = tokio::spawn(drive_timers(Arc::clone(&inner))); Self { inner, - tasks: vec![reader, timers], + tasks: vec![timers], } } - /// The interface name in use. - pub fn interface(&self) -> &str { - self.inner.tun.name() - } + /// Encrypts a packet and sends it to a peer. + /// + /// `false` when there is no tunnel for that peer, which is a state the + /// caller reports rather than an error: a peer whose link has not come + /// up yet is normal. + pub fn carry(&self, peer: iroh::EndpointId, packet: &[u8]) -> bool { + let found = read_lock(&self.inner.peers) + .values() + .find(|candidate| candidate.endpoint_id == peer) + .map(Arc::clone); + let Some(peer) = found else { + return false; + }; - /// The interface MTU. - pub fn mtu(&self) -> u32 { - self.inner.tun.mtu() + let mut scratch = vec![0u8; SCRATCH]; + // The encryption is the whole of what this protocol contributes, so + // it happens here rather than anywhere the packet passes through. + // The lock is released before the send: a slow link must not hold up + // the tunnel's timers. + let len = { + let mut tunn = match peer.tunn.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + match tunn.encapsulate(packet, &mut scratch) { + TunnResult::WriteToNetwork(out) => Some(out.len()), + // No session yet, so nothing to send. Counted as dropped + // rather than reported: the handshake is in flight and the + // next packet will go. + _ => None, + } + }; + let Some(len) = len else { + peer.counters + .dropped_no_session + .fetch_add(1, Ordering::Relaxed); + return false; + }; + + send_to_peer(&peer, &scratch[..len]); + peer.counters.tx_packets.fetch_add(1, Ordering::Relaxed); + peer.counters + .tx_bytes + .fetch_add(packet.len() as u64, Ordering::Relaxed); + true } /// Adds or replaces a peer and starts its tunnel. @@ -295,9 +332,6 @@ impl WireguardDevice { } write_lock(&self.inner.peers).insert(public_key, Arc::clone(&peer)); - if let Some(v4) = overlay_v4 { - write_lock(&self.inner.routes).insert(IpAddr::V4(v4), public_key); - } // Start the handshake now instead of waiting for the next timer tick, // so the tunnel is usable as soon as the link exists. @@ -307,21 +341,9 @@ impl WireguardDevice { /// Removes a peer and stops its tunnel. pub fn remove_peer(&self, public_key: &WgPublicKey) { - if let Some(peer) = write_lock(&self.inner.peers).remove(public_key) { - let mut routes = write_lock(&self.inner.routes); - let v4 = match peer.overlay_v4.lock() { - Ok(guard) => *guard, - Err(poisoned) => *poisoned.into_inner(), - }; - if let Some(v4) = v4 { - routes.remove(&IpAddr::V4(v4)); - } - } - } - - /// How many IPv4 derivation collisions have been resolved. - pub fn ipv4_conflicts(&self) -> u64 { - self.inner.ipv4_conflicts.load(Ordering::Relaxed) + // Dropping it stops the task and closes the link; there is no route + // to withdraw, because routes are not kept here. + write_lock(&self.inner.peers).remove(public_key); } /// Removes every peer whose key is not in `keep`. @@ -364,34 +386,6 @@ impl WireguardDevice { peers.sort_by_key(|peer| peer.public_key); peers } - - /// Unicast packets the operating system sent to an address no peer owns. - /// - /// A non-zero value means something tried to reach a host that is not in - /// the overlay. - pub fn unroutable_packets(&self) -> u64 { - self.inner.unroutable.load(Ordering::Relaxed) - } - - /// One destination that nobody owned, if there was one. - /// - /// A bare count says something is wrong but not what; the address usually - /// says it outright. - pub fn unroutable_sample(&self) -> Option { - match self.inner.unroutable_sample.lock() { - Ok(guard) => *guard, - Err(poisoned) => *poisoned.into_inner(), - } - } - - /// Multicast packets dropped. - /// - /// Expected and harmless: Linux emits multicast listener and router - /// solicitation traffic on any IPv6 interface, and this overlay is - /// unicast only. Counted separately so it does not look like a fault. - pub fn multicast_packets(&self) -> u64 { - self.inner.multicast.load(Ordering::Relaxed) - } } impl Drop for WireguardDevice { @@ -454,77 +448,6 @@ fn send_to_peer(peer: &Peer, payload: &[u8]) { } } -/// Operating system -> peer. -async fn read_from_os(inner: Arc) { - loop { - let Some(packet) = inner.tun.recv().await else { - return; - }; - - // Route by destination: only the peer that owns that overlay address - // may receive it. Both families go through the same table. - let Some(destination) = IpHeader::parse(&packet).map(|header| header.destination()) else { - inner.unroutable.fetch_add(1, Ordering::Relaxed); - continue; - }; - // The kernel emits multicast on every IPv6 interface. The overlay is - // unicast only, so this is dropped, but it is not a fault. - if destination.is_multicast() { - inner.multicast.fetch_add(1, Ordering::Relaxed); - continue; - } - let target = read_lock(&inner.routes).get(&destination).copied(); - let Some(target) = target else { - note_unroutable(&inner, destination); - continue; - }; - let peer = read_lock(&inner.peers).get(&target).cloned(); - let Some(peer) = peer else { - note_unroutable(&inner, destination); - continue; - }; - - let mut scratch = vec![0u8; SCRATCH]; - let outcome = { - let mut tunn = match peer.tunn.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - }; - match tunn.encapsulate(&packet, &mut scratch) { - TunnResult::WriteToNetwork(out) => Some(out.len()), - TunnResult::Done => None, - TunnResult::Err(err) => { - tracing::trace!(?err, "wireguard encapsulation failed"); - peer.counters - .protocol_errors - .fetch_add(1, Ordering::Relaxed); - None - } - _ => None, - } - }; - - if let Some(len) = outcome { - send_to_peer(&peer, &scratch[..len]); - peer.counters.tx_packets.fetch_add(1, Ordering::Relaxed); - peer.counters - .tx_bytes - .fetch_add(packet.len() as u64, Ordering::Relaxed); - } - } -} - -/// Counts a packet nobody owned the destination of, keeping one example. -fn note_unroutable(inner: &Inner, destination: IpAddr) { - inner.unroutable.fetch_add(1, Ordering::Relaxed); - let mut sample = match inner.unroutable_sample.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - }; - *sample = Some(destination); -} - -/// Peer -> operating system. async fn read_from_link(inner: Arc, peer: Arc) { loop { let Some(datagram) = peer.link.recv().await else { @@ -543,12 +466,11 @@ async fn read_from_link(inner: Arc, peer: Arc) { }; match tunn.decapsulate(None, input.unwrap_or(&[]), &mut scratch) { TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len()), - TunnResult::WriteToTunnelV6(out, source) => { - Outcome::ToTunnel(out.len(), IpAddr::V6(source)) - } - TunnResult::WriteToTunnelV4(out, source) => { - Outcome::ToTunnel(out.len(), IpAddr::V4(source)) - } + // The source boringtun reports is not consulted here: + // whether the peer may use it is checked where the + // claims live. + TunnResult::WriteToTunnelV6(out, _) => Outcome::ToTunnel(out.len()), + TunnResult::WriteToTunnelV4(out, _) => Outcome::ToTunnel(out.len()), TunnResult::Done => Outcome::Done, TunnResult::Err(err) => { tracing::trace!(?err, "wireguard decapsulation failed"); @@ -564,37 +486,20 @@ async fn read_from_link(inner: Arc, peer: Arc) { input = None; continue; } - Outcome::ToTunnel(len, source) => { + Outcome::ToTunnel(len) => { + // Handed up, not written out. This end has proved *who* + // sent the packet; whether that member may use the source + // address it chose is a question about a signed claim, + // and only the system level holds those. let payload = Bytes::copy_from_slice(&scratch[..len]); - // Enforce address ownership: a peer may only send from - // the address the control plane agreed it holds. An - // overlay address is signed by its holder, so this is - // checked against the agreement and never against - // anything the peer said here. - let owned = match source { - IpAddr::V4(addr) => { - let held = match peer.overlay_v4.lock() { - Ok(guard) => *guard, - Err(poisoned) => *poisoned.into_inner(), - }; - held == Some(addr) - } - // The overlay is IPv4. Anything else has no owner - // here and is dropped rather than guessed at. - IpAddr::V6(_) => false, - }; - if !owned { - peer.counters - .dropped_wrong_source - .fetch_add(1, Ordering::Relaxed); - break; - } - if inner.tun.send(payload).await.is_ok() { - peer.counters.rx_packets.fetch_add(1, Ordering::Relaxed); - peer.counters - .rx_bytes - .fetch_add(len as u64, Ordering::Relaxed); - } + inner + .sink + .deliver(inner.network, peer.endpoint_id, payload) + .await; + peer.counters.rx_packets.fetch_add(1, Ordering::Relaxed); + peer.counters + .rx_bytes + .fetch_add(len as u64, Ordering::Relaxed); break; } Outcome::Failed => { @@ -611,7 +516,7 @@ async fn read_from_link(inner: Arc, peer: Arc) { enum Outcome { ToNetwork(usize), - ToTunnel(usize, IpAddr), + ToTunnel(usize), Done, Failed, } diff --git a/crates/tsunagi/src/dataplane/wireguard/plugin.rs b/crates/tsunagi/src/dataplane/wireguard/plugin.rs index de09e0a..14a4c82 100644 --- a/crates/tsunagi/src/dataplane/wireguard/plugin.rs +++ b/crates/tsunagi/src/dataplane/wireguard/plugin.rs @@ -1,4 +1,4 @@ -//! The WireGuard IP plugin. +//! The WireGuard protocol plugin. //! //! Each agent builds its own view of the overlay from the set of participants //! the control plane agreed on. For a full mesh of `N` members that is `N - 1` @@ -11,16 +11,16 @@ //! [`PacketLink`](crate::dataplane::transport::PacketLink) per peer and runs //! a WireGuard tunnel over it. Reachability, hole punching and relaying are //! the transport's problem. +//! * It does **not** know which addresses anybody holds, and owns no +//! interface. One agent has one interface, at the system level, and every +//! protocol carries traffic for the same addresses on it. A packet arrives +//! here already routed and leaves here already decrypted. //! * It owns one WireGuard key per network, in its own store, unrelated to the //! iroh device key and to the network secret. -//! * It owns one packet interface per network, named deterministically. -//! * It never touches an interface it did not create, and never changes -//! routing, DNS or firewall settings beyond its own device. //! //! WireGuard runs in userspace via [`boringtun`], so there is no kernel module -//! and no `wg` tool to depend on. The only privileged step is creating the -//! packet interface, and even that is behind [`TunFactory`] so the whole data -//! plane can run unprivileged in tests. +//! and no `wg` tool to depend on, and nothing this plugin does needs +//! privileges: creating the interface is somebody else's job now. //! //! A failure here is reported and retried. It never stops the control plane. @@ -35,6 +35,7 @@ use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; use crate::BoxFuture; +use crate::dataplane::PacketSink; use crate::dataplane::transport::SharedLink; use crate::dataplane::{IpPlugin, PluginCapability, PluginContext, PluginError}; use crate::identity::NetworkId; @@ -44,7 +45,6 @@ use super::device::{PeerSummary, WireguardDevice}; use super::keys::{WgPublicKey, WgSecretKey}; use super::store::WgKeyStore; use crate::overlay::config::{DEFAULT_INTERFACE_PREFIX, interface_name}; -use crate::overlay::tun::{TunFactory, TunRequest}; use crate::state::Ipv4Range; /// The protocol identifier this plugin announces. @@ -150,12 +150,6 @@ pub struct NetworkOverview { pub ipv4_range: Option, /// Peers this agent knows about. pub peers: Vec, - /// Unicast packets the operating system sent to an address no peer owns. - pub unroutable_packets: u64, - /// Multicast packets dropped. Expected, not a fault. - pub multicast_packets: u64, - /// One destination nobody owned, if there was one. - pub unroutable_sample: Option, } impl NetworkOverview { @@ -202,12 +196,6 @@ struct NetworkState { ipv4_range: Option, /// The address last reported as missing, so it is said once, not forever. reported_missing_v4: Option, - /// What was last applied to the host interface. - /// - /// The overlay IPv4 address is allocated at run time and can change while - /// the agent runs, so the interface has to be brought back in line - /// without being recreated — recreating it would drop every tunnel. - applied: Option, } #[derive(Debug, Default)] @@ -232,16 +220,16 @@ struct Worker { config: WireguardConfig, /// This agent's endpoint id, learned when the plugin is attached. local_id: OnceLock, - tun_factory: Arc, store: WgKeyStore, shared: Mutex, context: OnceLock, + /// Where decrypted packets go, once the agent has attached one. + sink: OnceLock>, } impl std::fmt::Debug for Worker { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Worker") - .field("tun", &self.tun_factory.name()) .field("store", &self.store.path()) .finish() } @@ -260,10 +248,7 @@ impl WireguardPlugin { /// /// Must be called from inside a tokio runtime; the plugin starts no /// runtime of its own. - pub async fn open( - config: WireguardConfig, - tun_factory: Arc, - ) -> Result, PluginError> { + pub async fn open(config: WireguardConfig) -> Result, PluginError> { // Validate the prefix once, here, rather than failing per network. interface_name(&config.interface_prefix, NetworkId::from_bytes([0u8; 32]))?; @@ -283,10 +268,10 @@ impl WireguardPlugin { let worker = Arc::new(Worker { config, local_id: OnceLock::new(), - tun_factory, store, shared: Mutex::new(Shared::default()), context: OnceLock::new(), + sink: OnceLock::new(), }); let (commands, receiver) = mpsc::channel(64); @@ -337,20 +322,6 @@ impl WireguardPlugin { overlay_address_v4: state.allocations.get(&self.worker.local_id()).copied(), ipv4_range: state.ipv4_range, peers, - unroutable_packets: state - .device - .as_ref() - .map(|device| device.unroutable_packets()) - .unwrap_or(0), - multicast_packets: state - .device - .as_ref() - .map(|device| device.multicast_packets()) - .unwrap_or(0), - unroutable_sample: state - .device - .as_ref() - .and_then(|device| device.unroutable_sample()), }) } @@ -436,7 +407,6 @@ impl Worker { allocations: HashMap::new(), ipv4_range: None, reported_missing_v4: None, - applied: None, }); } @@ -449,75 +419,35 @@ impl Worker { Ok(true) } - /// What the host interface for a network should look like. - fn desired_request(&self, state: &NetworkState) -> TunRequest { - let own_range = state.ipv4_range; - TunRequest { - name: state.interface.clone(), - address: state.allocations.get(&self.local_id()).copied(), - prefix_len: own_range.map_or(0, |range| range.prefix_len), - mtu: self.config.mtu, - } - } - - /// Creates the packet interface and starts the WireGuard device. + /// Starts this network's tunnels. + /// + /// No interface is created: one agent has one, it belongs to the system + /// level, and decrypted packets are handed there rather than written + /// out. async fn ensure_device(&self, network: NetworkId) -> Result<(), PluginError> { - let (request, key) = { + let key = { let shared = self.lock_shared(); match shared.networks.get(&network) { - Some(state) if state.device.is_none() => { - (self.desired_request(state), state.key.clone()) - } + Some(state) if state.device.is_none() => state.key.clone(), _ => return Ok(()), } }; - let applied = request.clone(); - let tun = self.tun_factory.create(request).await?; - let device = Arc::new(WireguardDevice::start(network, key, tun)); + let sink = match self.sink.get() { + Some(sink) => Arc::clone(sink), + // Not attached to an agent: the protocol still runs, and its + // packets have nowhere to go. + None => Arc::new(crate::dataplane::DiscardPackets) as Arc, + }; + let device = Arc::new(WireguardDevice::start(network, key, sink)); let mut shared = self.lock_shared(); if let Some(state) = shared.networks.get_mut(&network) { - state.interface = device.interface().to_string(); state.device = Some(device); - state.applied = Some(applied); } Ok(()) } - /// Brings a live interface back in line after the overlay changed its - /// mind about this agent's address. - async fn ensure_addresses(&self, network: NetworkId) { - let wanted = { - let shared = self.lock_shared(); - match shared.networks.get(&network) { - Some(state) if state.device.is_some() => { - let wanted = self.desired_request(state); - if state.applied.as_ref() == Some(&wanted) { - return; - } - wanted - } - _ => return, - } - }; - - match self.tun_factory.reconfigure(wanted.clone()).await { - Ok(()) => { - let mut shared = self.lock_shared(); - if let Some(state) = shared.networks.get_mut(&network) { - state.applied = Some(wanted); - } - } - Err(err) => { - // Not fatal: the tunnels keep running on the addresses that - // are there, and the next reconciliation tries again. - tracing::warn!(%err, "cannot update the overlay interface addresses"); - self.report(network, err); - } - } - } - /// Brings the running tunnels in line with what is known. /// /// A peer gets a tunnel once both halves have arrived: its announcement, @@ -637,11 +567,10 @@ impl Worker { // kernel to remove an interface this agent created; the explicit // destroy makes that immediate and definite rather than dependent on // the last reader letting go. - let removed = self.lock_shared().networks.remove(&network); - if let Some(state) = removed { - drop(state.device); - self.tun_factory.destroy(&state.interface).await; - } + // Dropping the state drops the tunnels, which stops their tasks and + // closes their links. There is no interface to remove: the agent owns + // it, and it outlives any one network. + self.lock_shared().networks.remove(&network); } fn known_networks(&self) -> Vec { @@ -707,7 +636,6 @@ async fn run(worker: Arc, mut commands: mpsc::Receiver) { }, if wait_until.is_some() => { deadline = None; for network in std::mem::take(&mut pending) { - worker.ensure_addresses(network).await; worker.sync(network); } } @@ -718,7 +646,6 @@ async fn run(worker: Arc, mut commands: mpsc::Receiver) { if let Err(err) = worker.ensure_device(network).await { tracing::debug!(%err, "packet interface still unavailable"); } - worker.ensure_addresses(network).await; worker.sync(network); } } @@ -735,6 +662,7 @@ impl IpPlugin for WireguardPlugin { if let Some(local) = context.local_endpoint_id() { let _ = self.worker.local_id.set(local); } + let _ = self.worker.sink.set(context.packet_sink()); let _ = self.worker.context.set(context); } @@ -827,6 +755,15 @@ impl IpPlugin for WireguardPlugin { } } + fn carry(&self, network: NetworkId, peer: EndpointId, packet: bytes::Bytes) -> bool { + let shared = self.worker.lock_shared(); + shared + .networks + .get(&network) + .and_then(|state| state.device.as_ref()) + .is_some_and(|device| device.carry(peer, &packet)) + } + fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) { self.nudge(Command::Link { network, diff --git a/crates/tsunagi/src/overlay/interface.rs b/crates/tsunagi/src/overlay/interface.rs index 13349e7..a05768f 100644 --- a/crates/tsunagi/src/overlay/interface.rs +++ b/crates/tsunagi/src/overlay/interface.rs @@ -123,20 +123,28 @@ impl Tally { #[derive(Debug)] pub struct Interface { device: Arc, + factory: Arc, + name: String, + mtu: u32, routes: Arc, tally: Arc, + /// What was last applied to the host, so an unchanged table is not + /// re-applied on every pass. + applied: std::sync::Mutex>, task: std::sync::Mutex>>, } impl Interface { /// Creates the interface and starts moving packets. pub async fn start( - factory: &dyn TunFactory, - request: TunRequest, + factory: Arc, + name: impl Into, + mtu: u32, routes: Arc, carrier: Arc, ) -> Result { - let device = factory.create(request).await?; + let name = name.into(); + let device = factory.create(TunRequest::bare(name.clone(), mtu)).await?; let tally = Arc::new(Tally::default()); let task = { @@ -172,12 +180,52 @@ impl Interface { Ok(Self { device, + factory, + name, + mtu, routes, tally, + applied: std::sync::Mutex::new(Vec::new()), task: std::sync::Mutex::new(Some(task)), }) } + /// Brings the addresses on the host in line with the routing table. + /// + /// Called whenever a network agrees a different address for this agent. + /// The interface is never recreated for it: that would drop every tunnel + /// riding on it for the sake of one address. + pub async fn sync_addresses(&self) -> Result<(), OverlayError> { + let wanted = self.wanted_addresses(); + { + let applied = match self.applied.lock() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + }; + if applied == wanted { + return Ok(()); + } + } + + // One address per network, so at most a handful; the request carries + // the first and the provisioner reconciles the rest. + let request = TunRequest { + name: self.name.clone(), + address: wanted.first().and_then(|cidr| match cidr.addr { + IpAddr::V4(address) => Some(address), + IpAddr::V6(_) => None, + }), + prefix_len: wanted.first().map_or(0, |cidr| cidr.prefix_len), + mtu: self.mtu, + }; + self.factory.reconfigure(request).await?; + match self.applied.lock() { + Ok(mut guard) => *guard = wanted, + Err(poisoned) => *poisoned.into_inner() = wanted, + } + Ok(()) + } + /// The interface name the operating system gave. pub fn name(&self) -> &str { self.device.name() @@ -188,6 +236,11 @@ impl Interface { self.device.mtu() } + /// Removes the interface from the host. + pub async fn remove(&self) { + self.factory.destroy(self.device.name()).await; + } + /// The counters as they stand. pub fn counters(&self) -> Counters { self.tally.snapshot() @@ -355,10 +408,11 @@ mod tests { carried: std::sync::Mutex::new(Vec::new()), refuse, }); - let factory = MemoryTunFactory::new(); + let factory = Arc::new(MemoryTunFactory::new()); let interface = Interface::start( - &factory, - TunRequest::bare("tsuntest", 1280), + Arc::clone(&factory) as Arc, + "tsuntest", + 1280, Arc::clone(&routes), Arc::clone(&carrier) as Arc, ) diff --git a/crates/tsunagi/src/overlay/router.rs b/crates/tsunagi/src/overlay/router.rs index 3e09a87..f030316 100644 --- a/crates/tsunagi/src/overlay/router.rs +++ b/crates/tsunagi/src/overlay/router.rs @@ -175,6 +175,23 @@ impl RoutingTable { addresses } + /// Whether a range would collide with one another network already uses. + /// + /// Asked before proposing a range rather than after: with one interface + /// an agent that claimed an address it could not route would also be + /// telling everybody else to use that range, and "the range of the + /// lowest author wins" would spread the collision instead of containing + /// it. + pub fn would_overlap(&self, network: NetworkId, range: Ipv4Range) -> Option { + self.read().iter().find_map(|(other, existing)| { + if *other == network { + return None; + } + let other_range = existing.range?; + ranges_overlap(range, other_range).then_some(other_range) + }) + } + /// How many networks the table covers. pub fn len(&self) -> usize { self.read().len() @@ -319,6 +336,25 @@ mod tests { assert_eq!(table.route(addr(2)).map(|route| route.network), Some(first)); } + #[test] + fn a_range_can_be_asked_about_before_it_is_proposed() { + // The point of asking first: an agent that claims an address it + // cannot route also tells everybody else to use that range. + let table = RoutingTable::new(); + let first = network("first"); + let second = network("second"); + table.set_network(first, routes()).unwrap(); + + let mine: Ipv4Range = "10.13.37.0/24".parse().unwrap(); + assert_eq!(table.would_overlap(second, mine), Some(mine)); + // Its own range is not a collision with itself. + assert_eq!(table.would_overlap(first, mine), None); + assert_eq!( + table.would_overlap(second, "10.99.0.0/16".parse().unwrap()), + None + ); + } + #[test] fn a_nested_range_counts_as_overlapping() { let table = RoutingTable::new(); diff --git a/crates/tsunagi/tests/interface_provisioning.rs b/crates/tsunagi/tests/interface_provisioning.rs index 6f62e22..5c3d0fd 100644 --- a/crates/tsunagi/tests/interface_provisioning.rs +++ b/crates/tsunagi/tests/interface_provisioning.rs @@ -33,7 +33,7 @@ use tsunagi::{Agent, NetworkStatus}; struct HostedAgent { _dir: TempDir, agent: Agent, - plugin: Arc, + _plugin: Arc, host: MockHost, } @@ -43,12 +43,12 @@ impl HostedAgent { let provisioner = Arc::new(MockProvisioner::new(host.clone())); let factory = Arc::new(ManagedTunFactory::new(provisioner)); let config = WireguardConfig::new(dir.path().join("wireguard")) - .with_interface_prefix(tag) .with_reconcile(Duration::from_millis(20), Duration::from_millis(100)); - let plugin = WireguardPlugin::open(config, factory).await.unwrap(); + let plugin = WireguardPlugin::open(config).await.unwrap(); let agent = Agent::spawn( config_with(dir.path(), discovery) .with_overlay_ipv4_range(Some(DEFAULT_IPV4_RANGE)) + .with_interface(factory, tag, 1280) .with_plugin(plugin.clone() as Arc), ) .await @@ -56,17 +56,18 @@ impl HostedAgent { Self { _dir: dir, agent, - plugin, + _plugin: plugin, host, } } /// The interface name the plugin settled on for a network. - async fn interface(&self, network: NetworkId) -> String { - wait_until("the plugin named its interface", || async { - self.plugin - .overview(network) - .map(|view| view.interface) + /// The one interface this agent owns. Not per network. + async fn interface(&self, _network: NetworkId) -> String { + wait_until("the agent named its interface", || async { + self.agent + .overlay() + .map(|overlay| overlay.interface) .filter(|name| !name.is_empty()) }) .await @@ -217,7 +218,10 @@ async fn an_interface_belonging_to_something_else_is_left_alone() { } #[tokio::test] -async fn leaving_a_network_removes_the_interface_from_the_host() { +async fn leaving_a_network_takes_its_address_off_the_interface_but_not_the_interface() { + // The interface belongs to the agent, so it outlives any one network: + // another may still be using it. What a network takes with it is its own + // address. let discovery = SharedMemoryDiscovery::new(); let (name, secret) = network("provision-cleanup"); let agent = HostedAgent::spawn(&discovery, "tsunx", MockHost::new()).await; @@ -225,21 +229,25 @@ async fn leaving_a_network_removes_the_interface_from_the_host() { let network_id = agent.agent.join_network(&name, &secret).await.unwrap(); let interface = agent.interface(network_id).await; agent - .wait_for_host("the interface to exist", &interface, |state| state) + .wait_for_host("the address to be assigned", &interface, |state| { + state.filter(addressed) + }) .await; agent.agent.deactivate_network(network_id).await.unwrap(); - agent - .wait_for_host("the interface to be removed", &interface, |state| { - state.is_none().then_some(()) + let state = agent + .wait_for_host("the address to be withdrawn", &interface, |state| { + state.filter(|state| !addressed(state)) }) .await; + assert_eq!(state.kind, LinkKind::Tun, "the interface is still there"); + + // And it goes when the agent does. + agent.agent.shutdown().await; assert!( agent.host.names().is_empty(), "nothing is left behind: {:?}", agent.host.names() ); - - agent.agent.shutdown().await; } diff --git a/crates/tsunagi/tests/local_control.rs b/crates/tsunagi/tests/local_control.rs index 508efd5..980aea3 100644 --- a/crates/tsunagi/tests/local_control.rs +++ b/crates/tsunagi/tests/local_control.rs @@ -46,8 +46,11 @@ fn source(agent: Agent, plugin: Arc) -> Arc), + config_with(dir_a.path(), &discovery) + .with_interface(Arc::new(tuns), "tca0", 1280) + .with_plugin(plugin.clone() as Arc), ) .await .unwrap(); @@ -110,14 +113,14 @@ async fn a_client_sees_the_agent_and_its_overlay() { let dir_b = TempDir::new().unwrap(); let plugin_b = WireguardPlugin::open( WireguardConfig::new(dir_b.path().join("wg")) - .with_interface_prefix("tcb") .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), - Arc::new(MemoryTunFactory::new()), ) .await .unwrap(); let agent_b = Agent::spawn( - config_with(dir_b.path(), &discovery).with_plugin(plugin_b.clone() as Arc), + config_with(dir_b.path(), &discovery) + .with_interface(Arc::new(MemoryTunFactory::new()), "tcb0", 1280) + .with_plugin(plugin_b.clone() as Arc), ) .await .unwrap(); diff --git a/crates/tsunagi/tests/wireguard.rs b/crates/tsunagi/tests/wireguard.rs index 5e30e2d..f609662 100644 --- a/crates/tsunagi/tests/wireguard.rs +++ b/crates/tsunagi/tests/wireguard.rs @@ -106,15 +106,15 @@ impl WgAgent { let tuns = MemoryTunFactory::new(); let wg = tune( WireguardConfig::new(root.join("wireguard")) - .with_interface_prefix(tag) .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), ); - let plugin = WireguardPlugin::open(wg, Arc::new(tuns.clone())) - .await - .unwrap(); + let plugin = WireguardPlugin::open(wg).await.unwrap(); + // The interface belongs to the agent now, so the tag names it + // directly rather than prefixing one per network. let agent = Agent::spawn( config_with(root, discovery) .with_overlay_ipv4_range(range) + .with_interface(Arc::new(tuns.clone()), tag, 1280) .with_plugin(plugin.clone() as Arc), ) .await @@ -137,14 +137,16 @@ impl WgAgent { .await } - /// The in-memory packet interface for a network. - async fn tun(&self, network: NetworkId) -> Arc { + /// The one in-memory packet interface this agent owns. + /// + /// Not per network: one agent has one interface, and which network a + /// packet on it belongs to is decided by its address. + async fn tun(&self, _network: NetworkId) -> Arc { let name = wait_until("the packet interface exists", || async { - let view = self.plugin.overview(network)?; - self.tuns.device(&view.interface).map(|_| view.interface) + self.agent.overlay().map(|overlay| overlay.interface) }) .await; - self.tuns.device(&name).unwrap() + self.tuns.device(&name).expect("the device was created") } /// Waits until `count` tunnels have completed a WireGuard handshake. @@ -269,10 +271,11 @@ async fn a_peer_cannot_send_from_an_address_it_does_not_own() { tun_a.push_from_os(ipv4_packet(someone_else, addr_b, b"spoofed")); // B must drop it: the source is not the address A holds. + // Counted on the interface, not on the tunnel: the protocol proved who + // sent the packet, and whether that member may use the address it chose + // is a question about a signed claim, which the system level holds. wait_until("the spoofed packet is dropped", || async { - let view = b.plugin.overview(network_id)?; - let tunnel = view.peers.first()?.tunnel.as_ref()?; - (tunnel.stats.dropped_wrong_source >= 1).then_some(()) + (b.agent.overlay()?.counters.wrong_source >= 1).then_some(()) }) .await; @@ -370,9 +373,7 @@ async fn an_ipv4_source_a_peer_does_not_own_is_dropped() { tun_a.push_from_os(ipv4_packet(forged, v4_b, b"spoofed v4")); wait_until("the spoofed IPv4 packet is dropped", || async { - let view = b.plugin.overview(network_id)?; - let tunnel = view.peers.first()?.tunnel.as_ref()?; - (tunnel.stats.dropped_wrong_source >= 1).then_some(()) + (b.agent.overlay()?.counters.wrong_source >= 1).then_some(()) }) .await; @@ -611,7 +612,8 @@ async fn packets_for_an_unknown_address_are_counted_not_broadcast() { wait_until("the packet is counted as unroutable", || async { let view = a.plugin.overview(network_id)?; - (view.unroutable_packets >= 1).then_some(()) + let _ = view; + (a.agent.overlay()?.counters.unroutable >= 1).then_some(()) }) .await; @@ -712,34 +714,61 @@ async fn a_departing_peer_loses_its_tunnel() { } #[tokio::test] -async fn two_networks_get_separate_interfaces_keys_and_overlays() { +async fn two_networks_share_one_interface_with_keys_and_ranges_of_their_own() { + // One agent, one interface — so two networks on it must use different + // ranges, or an address would belong to both. Each network's range is + // settled by whoever got there first, and the agent adopts what it + // finds; see the routing table's own tests for the refusal when they + // overlap. let discovery = SharedMemoryDiscovery::new(); let (name_a, secret_a) = network("wg-left"); let (name_b, secret_b) = network("wg-right"); + let beta_range = Some("10.99.0.0/16".parse::().unwrap()); let hub = WgAgent::spawn(&discovery, "th").await; let left = WgAgent::spawn(&discovery, "tl").await; - let right = WgAgent::spawn(&discovery, "tr").await; + let right = WgAgent::spawn_range(&discovery, "tr", beta_range).await; - let alpha = hub.agent.join_network(&name_a, &secret_a).await.unwrap(); - let beta = hub.agent.join_network(&name_b, &secret_b).await.unwrap(); - left.agent.join_network(&name_a, &secret_a).await.unwrap(); - right.agent.join_network(&name_b, &secret_b).await.unwrap(); + // The other members settle each range before the hub joins, so it has + // something to adopt rather than a default to collide with. + let alpha = left.agent.join_network(&name_a, &secret_a).await.unwrap(); + let beta = right.agent.join_network(&name_b, &secret_b).await.unwrap(); + left.overlay(alpha).await; + right.overlay(beta).await; + hub.agent.join_network(&name_a, &secret_a).await.unwrap(); + hub.agent.join_network(&name_b, &secret_b).await.unwrap(); hub.wait_for_tunnels(alpha, 1).await; hub.wait_for_tunnels(beta, 1).await; - let view_alpha = hub.plugin.overview(alpha).unwrap(); - let view_beta = hub.plugin.overview(beta).unwrap(); - assert_ne!(view_alpha.interface, view_beta.interface); assert_ne!( - view_alpha.public_key, view_beta.public_key, + hub.plugin.overview(alpha).unwrap().public_key, + hub.plugin.overview(beta).unwrap().public_key, "one WireGuard identity per network, not one per host" ); - assert_eq!(hub.tuns.devices().len(), 2); + assert_eq!( + hub.tuns.devices().len(), + 1, + "one agent has one interface, whatever it is a member of" + ); - // Traffic in one overlay never surfaces in the other. - let hub_alpha = hub.overlay(alpha).await; + // Each network's address comes from its own range. + let hub_alpha = wait_until("the hub settles into alpha's range", || async { + let address = hub.plugin.overview(alpha)?.overlay_address_v4?; + tsunagi::state::DEFAULT_IPV4_RANGE + .contains(address) + .then_some(address) + }) + .await; + let hub_beta = wait_until("the hub settles into beta's range", || async { + let address = hub.plugin.overview(beta)?.overlay_address_v4?; + beta_range?.contains(address).then_some(address) + }) + .await; + assert_ne!(hub_alpha, hub_beta); + + // Traffic in one overlay never surfaces in the other, though both cross + // the same interface. let left_addr = left.overlay(alpha).await; hub.tun(alpha) .await @@ -759,14 +788,6 @@ async fn two_networks_get_separate_interfaces_keys_and_overlays() { "the other overlay must see nothing" ); - // Deactivating one network removes only its interface. - hub.agent.deactivate_network(alpha).await.unwrap(); - wait_until("the alpha interface is gone", || async { - hub.plugin.overview(alpha).is_none().then_some(()) - }) - .await; - assert!(hub.plugin.overview(beta).is_some()); - hub.shutdown().await; left.shutdown().await; right.shutdown().await; @@ -971,11 +992,8 @@ async fn an_mtu_below_what_ipv4_guarantees_is_refused() { // IPv6's sake; the overlay is IPv4 now and a relayed path with small // datagrams can be matched instead of warned about. let dir = TempDir::new().unwrap(); - let result = WireguardPlugin::open( - WireguardConfig::new(dir.path()).with_mtu(MIN_MTU - 1), - Arc::new(MemoryTunFactory::new()), - ) - .await; + let result = + WireguardPlugin::open(WireguardConfig::new(dir.path()).with_mtu(MIN_MTU - 1)).await; match result { Err(err) => { let text = err.to_string(); @@ -989,11 +1007,8 @@ async fn an_mtu_below_what_ipv4_guarantees_is_refused() { const { assert!(DEFAULT_MTU > MIN_MTU) }; assert_eq!(WIREGUARD_OVERHEAD, 32); assert!( - WireguardPlugin::open( - WireguardConfig::new(dir.path().join("ok")), - Arc::new(MemoryTunFactory::new()), - ) - .await - .is_ok() + WireguardPlugin::open(WireguardConfig::new(dir.path().join("ok"))) + .await + .is_ok() ); }