Take the interface off the protocol and give it to the agent

One agent, one interface. The plugin no longer creates one, no longer
holds a TUN factory, no longer keeps a routing table and no longer
decides who owns an address. What is left of it is the protocol: a
WireGuard key per network, a tunnel per peer, encryption on the way out
and decryption on the way in.

The packet path is now explicit about where each decision lives. Out: the
agent's interface reads a packet, the routing table says whose
destination it is, and each protocol is asked in turn whether it can
carry it there. In: the protocol decrypts and hands the packet up with
the peer it came from attached, and the agent checks that peer is
entitled to the source address before writing it out. A protocol proves
who; only the system level knows what they may say.

Two things found by making it work.

`carry` sent the packet unencrypted at first. The encryption had lived in
the interface loop that moved to the core, so taking that out quietly
removed it — the receiving end rejected plaintext as a bad WireGuard
datagram and the counters said nothing at all. Encryption belongs with
the protocol and is now there, with packets dropped for having no session
yet counted apart, because a handful while a tunnel comes up is normal
and a number that keeps climbing is not.

An agent could impose a range it could not itself route. With one
interface two networks need different ranges, and "the lowest author's
range wins" would have carried one agent's colliding default to
everybody. The configured range is now reserved when a network is
activated — on the serialised path, so the answer does not depend on
which task ran first — and an agent that cannot have it proposes nothing
and adopts whatever the network settles on.

Leaving a network takes its address off the interface and leaves the
interface; the interface goes when the agent does.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
2026-09-21 19:20:30 +01:00
co-authored by Claude Opus 5
parent c852de2c78
commit 9415866193
13 changed files with 691 additions and 393 deletions
+42 -14
View File
@@ -1884,17 +1884,24 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
// The data plane is optional and never required for the control plane. // The data plane is optional and never required for the control plane.
let wireguard = if args.wireguard { 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<dyn TunFactory> = if args.no_tun { let tun_factory: Arc<dyn TunFactory> = if args.no_tun {
Arc::new(MemoryTunFactory::new()) Arc::new(MemoryTunFactory::new())
} else { } else {
system_tun_factory()? system_tun_factory()?
}; };
let mut wg = WireguardConfig::new(paths.state_dir.join("wireguard")) let mtu = args
.with_interface_prefix(args.wg_prefix.clone()); .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 { if let Some(mtu) = args.wg_mtu {
wg = wg.with_mtu(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<dyn IpPlugin>); config = config.with_plugin(plugin.clone() as Arc<dyn IpPlugin>);
Some(plugin) Some(plugin)
} else { } else {
@@ -2030,6 +2037,7 @@ async fn build_report(
StatusReport, StatusReport,
}; };
let overlay = agent.overlay();
let dns = dns.map(|dns| DnsReport { let dns = dns.map(|dns| DnsReport {
zone: dns.zone, zone: dns.zone,
listening: dns.listening.map(|address| address.to_string()), listening: dns.listening.map(|address| address.to_string()),
@@ -2051,8 +2059,10 @@ async fn build_report(
let overlay = wireguard let overlay = wireguard
.and_then(|plugin| plugin.overview(network.network_id)) .and_then(|plugin| plugin.overview(network.network_id))
.map(|view| OverlayReport { .map(|view| OverlayReport {
interface: view.interface.clone(), interface: overlay
mtu: view.mtu, .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()), address: view.overlay_address_v4.map(|addr| addr.to_string()),
prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len), prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len),
peers: view peers: view
@@ -2089,9 +2099,21 @@ async fn build_report(
.unwrap_or_else(|| "no data link".into()), .unwrap_or_else(|| "no data link".into()),
}) })
.collect(), .collect(),
unroutable_packets: view.unroutable_packets, // The interface belongs to the agent, so the counters
multicast_packets: view.multicast_packets, // about it come from there and are the same for every
unroutable_sample: view.unroutable_sample.map(|address| address.to_string()), // 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 { NetworkReport {
@@ -2309,12 +2331,18 @@ async fn print_status(agent: &Agent, network: NetworkId, wireguard: Option<&Wire
), ),
} }
} }
if view.unroutable_packets > 0 { }
println!( if let Some(overlay) = agent.overlay()
" {} packet(s) for unknown addresses", && overlay.counters.unroutable > 0
view.unroutable_packets {
); println!(
} " {} packet(s) for unknown addresses{}",
overlay.counters.unroutable,
match overlay.counters.unroutable_sample {
Some(sample) => format!(" (for example {sample})"),
None => String::new(),
}
);
} }
println!(); println!();
} }
+162 -7
View File
@@ -27,7 +27,7 @@ mod status;
pub use events::Event; pub use events::Event;
pub use status::{ pub use status::{
AgentStatus, CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus, AgentStatus, CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus,
PeerStatus, OverlayStatus, PeerStatus,
}; };
use std::collections::HashMap; use std::collections::HashMap;
@@ -37,13 +37,15 @@ use iroh::{EndpointAddr, EndpointId};
use tokio::sync::{RwLock, broadcast, mpsc, oneshot}; use tokio::sync::{RwLock, broadcast, mpsc, oneshot};
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use crate::BoxFuture;
use crate::config::{AgentConfig, Limits}; use crate::config::{AgentConfig, Limits};
use crate::dataplane::transport::PacketTransport; use crate::dataplane::transport::PacketTransport;
use crate::dataplane::transport::iroh_link::{IrohTransport, TransportContext}; 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::error::{Error, Result};
use crate::identity::{DeviceIdentity, NetworkId, NetworkKeys, NetworkName, NetworkSecret}; use crate::identity::{DeviceIdentity, NetworkId, NetworkKeys, NetworkName, NetworkSecret};
use crate::net::EndpointAdapter; use crate::net::EndpointAdapter;
use crate::overlay::PacketCarrier;
use crate::proto::handshake; use crate::proto::handshake;
use crate::proto::message::ControlMessage; use crate::proto::message::ControlMessage;
use crate::storage::{CacheOutcome, Storage}; use crate::storage::{CacheOutcome, Storage};
@@ -87,6 +89,10 @@ struct Inner {
accept_task: std::sync::Mutex<Option<JoinHandle<()>>>, accept_task: std::sync::Mutex<Option<JoinHandle<()>>>,
plugin_task: std::sync::Mutex<Option<JoinHandle<()>>>, plugin_task: std::sync::Mutex<Option<JoinHandle<()>>>,
transport: std::sync::OnceLock<Arc<dyn PacketTransport>>, transport: std::sync::OnceLock<Arc<dyn PacketTransport>>,
/// Who holds which overlay address, across every network.
routes: Arc<crate::overlay::RoutingTable>,
/// The one interface, when the agent was given a way to make one.
interface: std::sync::OnceLock<Arc<crate::overlay::Interface>>,
} }
impl Inner { 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<Inner>);
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<Inner>);
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. /// Answers the data plane transport's questions about the agent.
/// ///
/// Holds a weak reference on purpose: the transport lives inside the agent, so /// Holds a weak reference for the same reason as [`Carrier`]: the transport
/// a strong one would be a cycle and the agent — with its open databases and /// lives inside the agent, so a strong one would be a cycle and the agent —
/// its directory lock — would never be released. /// with its open databases and its directory lock — would never be released.
#[derive(Debug)] #[derive(Debug)]
struct TransportCtx(Weak<Inner>); struct TransportCtx(Weak<Inner>);
@@ -178,9 +242,38 @@ impl Agent {
accept_task: std::sync::Mutex::new(None), accept_task: std::sync::Mutex::new(None),
plugin_task: std::sync::Mutex::new(None), plugin_task: std::sync::Mutex::new(None),
transport: std::sync::OnceLock::new(), transport: std::sync::OnceLock::new(),
routes: Arc::new(crate::overlay::RoutingTable::new()),
interface: std::sync::OnceLock::new(),
config, 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<dyn PacketCarrier>;
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 // 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 // punching and relay fallback from. It is a separate ALPN and a
// separate connection, so the two planes stay independent. // separate connection, so the two planes stay independent.
@@ -205,7 +298,11 @@ impl Agent {
// bound; overflow drops the request rather than stalling the plugin. // bound; overflow drops the request rather than stalling the plugin.
if !inner.config.plugins.is_empty() { if !inner.config.plugins.is_empty() {
let (plugin_tx, plugin_rx) = mpsc::channel(64); 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<dyn PacketSink>);
let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink);
for plugin in &inner.config.plugins { for plugin in &inner.config.plugins {
plugin.attach(context.clone()); plugin.attach(context.clone());
} }
@@ -245,6 +342,47 @@ impl Agent {
self.inner.adapter.loopback_addr() 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<crate::state::Ipv4Range> {
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<OverlayStatus> {
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. /// The hostname announced to peers.
pub fn hostname(&self) -> String { pub fn hostname(&self) -> String {
self.inner.read_hostname() self.inner.read_hostname()
@@ -367,10 +505,12 @@ impl Agent {
discovery: self.inner.config.discovery.clone(), discovery: self.inner.config.discovery.clone(),
discovery_interval: self.inner.config.discovery_interval, discovery_interval: self.inner.config.discovery_interval,
plugins: self.inner.config.plugins.clone(), plugins: self.inner.config.plugins.clone(),
routes: Arc::clone(&self.inner.routes),
interface: self.inner.interface.get().cloned(),
hostname: self.inner.read_hostname(), hostname: self.inner.read_hostname(),
transport: self.inner.transport.get().cloned(), transport: self.inner.transport.get().cloned(),
device_secret: self.inner.identity.signing_key(), 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); networks.insert(network_id, handle);
drop(networks); drop(networks);
@@ -401,6 +541,15 @@ impl Agent {
for plugin in &self.inner.config.plugins { for plugin in &self.inner.config.plugins {
plugin.on_network_deactivated(network_id); 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?; self.inner.storage.set_auto_start(network_id, false).await?;
Ok(()) 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 // Plugins remove whatever system objects they created. A plugin that
// misbehaves here must not hold up the agent, so this is bounded. // misbehaves here must not hold up the agent, so this is bounded.
for plugin in &self.inner.config.plugins { for plugin in &self.inner.config.plugins {
+55 -4
View File
@@ -120,6 +120,10 @@ pub(crate) struct RuntimeParams {
pub(crate) discovery: Option<Arc<dyn NetworkDiscovery>>, pub(crate) discovery: Option<Arc<dyn NetworkDiscovery>>,
pub(crate) discovery_interval: Duration, pub(crate) discovery_interval: Duration,
pub(crate) plugins: Vec<SharedPlugin>, pub(crate) plugins: Vec<SharedPlugin>,
/// Who holds which overlay address, shared with every other network.
pub(crate) routes: Arc<crate::overlay::RoutingTable>,
/// The one interface, when the agent has one.
pub(crate) interface: Option<Arc<crate::overlay::Interface>>,
pub(crate) hostname: String, pub(crate) hostname: String,
/// Signing key for this agent's own records. /// Signing key for this agent's own records.
pub(crate) device_secret: iroh::SecretKey, pub(crate) device_secret: iroh::SecretKey,
@@ -615,7 +619,7 @@ impl Runtime {
.unwrap_or(0); .unwrap_or(0);
self.ensure_own_claim().await; self.ensure_own_claim().await;
self.publish_allocations(); self.publish_allocations().await;
} }
/// The range this network uses: whatever it has already settled on, else /// 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 /// Adopting the agreed one is what lets a participant join without being
/// told the range out of band. /// told the range out of band.
fn effective_range(&self) -> Option<Ipv4Range> { fn effective_range(&self) -> Option<Ipv4Range> {
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. /// 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. // Somebody may have taken the address we were using.
self.ensure_own_claim().await; self.ensure_own_claim().await;
self.publish_allocations(); self.publish_allocations().await;
if self.state.records() != before { if self.state.records() != before {
self.broadcast_state(); self.broadcast_state();
} }
} }
/// Tells the plugins who holds which overlay address. /// 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 { let Some(range) = self.effective_range() else {
return; return;
}; };
@@ -832,6 +849,40 @@ impl Runtime {
for plugin in &self.params.plugins { for plugin in &self.params.plugins {
plugin.on_address_allocation(self.network_id, range, &allocations); 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 // ------------------------------------------------------------ data plane
+13
View File
@@ -36,6 +36,19 @@ pub struct CandidateStatus {
pub consecutive_failures: u32, 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<crate::overlay::Cidr>,
/// What has happened on it.
pub counters: crate::overlay::Counters,
}
/// A member the signed state knows about, connected or not. /// A member the signed state knows about, connected or not.
/// ///
/// This is the durable roster: it comes from signed records, so a member that /// This is the durable roster: it comes from signed records, so a member that
+27
View File
@@ -237,6 +237,17 @@ pub struct AgentConfig {
pub reconnect: ReconnectPolicy, pub reconnect: ReconnectPolicy,
/// IP plugins whose capabilities are announced and dispatched. /// IP plugins whose capabilities are announced and dispatched.
pub plugins: Vec<SharedPlugin>, pub plugins: Vec<SharedPlugin>,
/// 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<Arc<dyn crate::overlay::TunFactory>>,
/// 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. /// The IPv4 overlay range this agent proposes.
/// ///
/// Addresses are allocated from it and recorded in signed state, so a /// Addresses are allocated from it and recorded in signed state, so a
@@ -259,6 +270,9 @@ impl AgentConfig {
limits: Limits::default(), limits: Limits::default(),
reconnect: ReconnectPolicy::default(), reconnect: ReconnectPolicy::default(),
plugins: Vec::new(), plugins: Vec::new(),
tun_factory: None,
interface_name: "tsun0".to_string(),
interface_mtu: 1280,
overlay_ipv4_range: Some(crate::state::DEFAULT_IPV4_RANGE), overlay_ipv4_range: Some(crate::state::DEFAULT_IPV4_RANGE),
} }
} }
@@ -314,6 +328,19 @@ impl AgentConfig {
self self
} }
/// Gives the agent an overlay interface, from this factory.
pub fn with_interface(
mut self,
factory: Arc<dyn crate::overlay::TunFactory>,
name: impl Into<String>,
mtu: u32,
) -> Self {
self.tun_factory = Some(factory);
self.interface_name = name.into();
self.interface_mtu = mtu;
self
}
/// Replaces the limits. /// Replaces the limits.
pub fn with_limits(mut self, limits: Limits) -> Self { pub fn with_limits(mut self, limits: Limits) -> Self {
self.limits = limits; self.limits = limits;
+67 -1
View File
@@ -52,6 +52,38 @@ pub struct PluginCapability {
pub data: Vec<u8>, pub data: Vec<u8>,
} }
/// 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. /// Errors a plugin may return. They are recorded, never fatal for the agent.
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
#[non_exhaustive] #[non_exhaustive]
@@ -109,13 +141,32 @@ pub(crate) enum PluginRequest {
pub struct PluginContext { pub struct PluginContext {
sender: Option<mpsc::Sender<PluginRequest>>, sender: Option<mpsc::Sender<PluginRequest>>,
local: Option<EndpointId>, local: Option<EndpointId>,
/// Where decrypted packets go. Absent when nothing is carrying traffic,
/// in which case a protocol still runs and its packets are discarded.
sink: Option<Arc<dyn PacketSink>>,
} }
impl PluginContext { impl PluginContext {
pub(crate) fn new(sender: mpsc::Sender<PluginRequest>, local: EndpointId) -> Self { pub(crate) fn new(
sender: mpsc::Sender<PluginRequest>,
local: EndpointId,
sink: Option<Arc<dyn PacketSink>>,
) -> Self {
Self { Self {
sender: Some(sender), sender: Some(sender),
local: Some(local), 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<dyn PacketSink> {
match &self.sink {
Some(sink) => Arc::clone(sink),
None => Arc::new(DiscardPackets),
} }
} }
@@ -124,6 +175,7 @@ impl PluginContext {
Self { Self {
sender: None, sender: None,
local: None, local: None,
sink: None,
} }
} }
@@ -238,6 +290,20 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static {
let _ = (network, range, allocations); 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. /// 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 /// The plugin moves its packets over this link and never learns how the
+91 -186
View File
@@ -29,7 +29,7 @@
//! against. //! against.
use std::collections::HashMap; use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr}; use std::net::Ipv4Addr;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock}; use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration; use std::time::Duration;
@@ -39,13 +39,11 @@ use bytes::Bytes;
use iroh::EndpointId; use iroh::EndpointId;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use crate::dataplane::PluginError;
use crate::dataplane::transport::{SharedLink, TransportError}; use crate::dataplane::transport::{SharedLink, TransportError};
use crate::dataplane::{PacketSink, PluginError};
use crate::identity::NetworkId; use crate::identity::NetworkId;
use super::keys::{WgPublicKey, WgSecretKey}; use super::keys::{WgPublicKey, WgSecretKey};
use crate::overlay::packet::IpHeader;
use crate::overlay::tun::TunDevice;
/// How often WireGuard's own timers are driven. /// How often WireGuard's own timers are driven.
/// ///
@@ -65,6 +63,8 @@ struct PeerCounters {
rx_bytes: AtomicU64, rx_bytes: AtomicU64,
dropped_wrong_source: AtomicU64, dropped_wrong_source: AtomicU64,
dropped_oversize: AtomicU64, dropped_oversize: AtomicU64,
/// Packets there was no session to encrypt with yet.
dropped_no_session: AtomicU64,
protocol_errors: AtomicU64, protocol_errors: AtomicU64,
} }
@@ -85,6 +85,11 @@ pub struct PeerStats {
pub dropped_wrong_source: u64, pub dropped_wrong_source: u64,
/// Packets dropped because they did not fit in one link datagram. /// Packets dropped because they did not fit in one link datagram.
pub dropped_oversize: u64, 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. /// WireGuard protocol errors, including packets that failed to decrypt.
pub protocol_errors: u64, pub protocol_errors: u64,
} }
@@ -149,6 +154,7 @@ impl Peer {
rx_bytes: self.counters.rx_bytes.load(Ordering::Relaxed), rx_bytes: self.counters.rx_bytes.load(Ordering::Relaxed),
dropped_wrong_source: self.counters.dropped_wrong_source.load(Ordering::Relaxed), dropped_wrong_source: self.counters.dropped_wrong_source.load(Ordering::Relaxed),
dropped_oversize: self.counters.dropped_oversize.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), protocol_errors: self.counters.protocol_errors.load(Ordering::Relaxed),
} }
} }
@@ -186,29 +192,25 @@ pub struct PeerSummary {
struct Inner { struct Inner {
network: NetworkId, network: NetworkId,
private_key: WgSecretKey, private_key: WgSecretKey,
tun: Arc<dyn TunDevice>,
/// The IPv4 overlay range, when the overlay is dual stack.
peers: RwLock<HashMap<WgPublicKey, Arc<Peer>>>, peers: RwLock<HashMap<WgPublicKey, Arc<Peer>>>,
/// Both families, so one lookup routes any packet. /// Where decrypted packets go.
routes: RwLock<HashMap<IpAddr, WgPublicKey>>, ///
/// 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<dyn PacketSink>,
next_index: AtomicU32, next_index: AtomicU32,
unroutable: AtomicU64,
/// One destination nobody owned, kept so the counter can be acted on.
unroutable_sample: Mutex<Option<IpAddr>>,
multicast: AtomicU64,
ipv4_conflicts: AtomicU64,
} }
impl std::fmt::Debug for Inner { impl std::fmt::Debug for Inner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Inner") f.debug_struct("Inner")
.field("network", &self.network.fmt_short()) .field("network", &self.network.fmt_short())
.field("tun", &self.tun.name())
.finish() .finish()
} }
} }
/// A userspace WireGuard interface for one network. /// The userspace WireGuard tunnels of one network.
#[derive(Debug)] #[derive(Debug)]
pub struct WireguardDevice { pub struct WireguardDevice {
inner: Arc<Inner>, inner: Arc<Inner>,
@@ -216,38 +218,73 @@ pub struct WireguardDevice {
} }
impl WireguardDevice { impl WireguardDevice {
/// Starts a device on top of `tun`. /// Starts the tunnels for one network.
pub fn start(network: NetworkId, private_key: WgSecretKey, tun: Arc<dyn TunDevice>) -> Self { ///
/// 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<dyn PacketSink>) -> Self {
let inner = Arc::new(Inner { let inner = Arc::new(Inner {
network, network,
private_key, private_key,
tun,
peers: RwLock::new(HashMap::new()), peers: RwLock::new(HashMap::new()),
routes: RwLock::new(HashMap::new()), sink,
next_index: AtomicU32::new(1), 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))); let timers = tokio::spawn(drive_timers(Arc::clone(&inner)));
Self { Self {
inner, inner,
tasks: vec![reader, timers], tasks: vec![timers],
} }
} }
/// The interface name in use. /// Encrypts a packet and sends it to a peer.
pub fn interface(&self) -> &str { ///
self.inner.tun.name() /// `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. let mut scratch = vec![0u8; SCRATCH];
pub fn mtu(&self) -> u32 { // The encryption is the whole of what this protocol contributes, so
self.inner.tun.mtu() // 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. /// 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)); 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, // Start the handshake now instead of waiting for the next timer tick,
// so the tunnel is usable as soon as the link exists. // so the tunnel is usable as soon as the link exists.
@@ -307,21 +341,9 @@ impl WireguardDevice {
/// Removes a peer and stops its tunnel. /// Removes a peer and stops its tunnel.
pub fn remove_peer(&self, public_key: &WgPublicKey) { pub fn remove_peer(&self, public_key: &WgPublicKey) {
if let Some(peer) = write_lock(&self.inner.peers).remove(public_key) { // Dropping it stops the task and closes the link; there is no route
let mut routes = write_lock(&self.inner.routes); // to withdraw, because routes are not kept here.
let v4 = match peer.overlay_v4.lock() { write_lock(&self.inner.peers).remove(public_key);
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)
} }
/// Removes every peer whose key is not in `keep`. /// Removes every peer whose key is not in `keep`.
@@ -364,34 +386,6 @@ impl WireguardDevice {
peers.sort_by_key(|peer| peer.public_key); peers.sort_by_key(|peer| peer.public_key);
peers 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<IpAddr> {
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 { 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<Inner>) {
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<Inner>, peer: Arc<Peer>) { async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
loop { loop {
let Some(datagram) = peer.link.recv().await else { let Some(datagram) = peer.link.recv().await else {
@@ -543,12 +466,11 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
}; };
match tunn.decapsulate(None, input.unwrap_or(&[]), &mut scratch) { match tunn.decapsulate(None, input.unwrap_or(&[]), &mut scratch) {
TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len()), TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len()),
TunnResult::WriteToTunnelV6(out, source) => { // The source boringtun reports is not consulted here:
Outcome::ToTunnel(out.len(), IpAddr::V6(source)) // whether the peer may use it is checked where the
} // claims live.
TunnResult::WriteToTunnelV4(out, source) => { TunnResult::WriteToTunnelV6(out, _) => Outcome::ToTunnel(out.len()),
Outcome::ToTunnel(out.len(), IpAddr::V4(source)) TunnResult::WriteToTunnelV4(out, _) => Outcome::ToTunnel(out.len()),
}
TunnResult::Done => Outcome::Done, TunnResult::Done => Outcome::Done,
TunnResult::Err(err) => { TunnResult::Err(err) => {
tracing::trace!(?err, "wireguard decapsulation failed"); tracing::trace!(?err, "wireguard decapsulation failed");
@@ -564,37 +486,20 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
input = None; input = None;
continue; 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]); let payload = Bytes::copy_from_slice(&scratch[..len]);
// Enforce address ownership: a peer may only send from inner
// the address the control plane agreed it holds. An .sink
// overlay address is signed by its holder, so this is .deliver(inner.network, peer.endpoint_id, payload)
// checked against the agreement and never against .await;
// anything the peer said here. peer.counters.rx_packets.fetch_add(1, Ordering::Relaxed);
let owned = match source { peer.counters
IpAddr::V4(addr) => { .rx_bytes
let held = match peer.overlay_v4.lock() { .fetch_add(len as u64, Ordering::Relaxed);
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);
}
break; break;
} }
Outcome::Failed => { Outcome::Failed => {
@@ -611,7 +516,7 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
enum Outcome { enum Outcome {
ToNetwork(usize), ToNetwork(usize),
ToTunnel(usize, IpAddr), ToTunnel(usize),
Done, Done,
Failed, Failed,
} }
+40 -103
View File
@@ -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 //! 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` //! 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 //! [`PacketLink`](crate::dataplane::transport::PacketLink) per peer and runs
//! a WireGuard tunnel over it. Reachability, hole punching and relaying are //! a WireGuard tunnel over it. Reachability, hole punching and relaying are
//! the transport's problem. //! 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 //! * It owns one WireGuard key per network, in its own store, unrelated to the
//! iroh device key and to the network secret. //! 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 //! 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 //! and no `wg` tool to depend on, and nothing this plugin does needs
//! packet interface, and even that is behind [`TunFactory`] so the whole data //! privileges: creating the interface is somebody else's job now.
//! plane can run unprivileged in tests.
//! //!
//! A failure here is reported and retried. It never stops the control plane. //! 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 tokio::task::JoinHandle;
use crate::BoxFuture; use crate::BoxFuture;
use crate::dataplane::PacketSink;
use crate::dataplane::transport::SharedLink; use crate::dataplane::transport::SharedLink;
use crate::dataplane::{IpPlugin, PluginCapability, PluginContext, PluginError}; use crate::dataplane::{IpPlugin, PluginCapability, PluginContext, PluginError};
use crate::identity::NetworkId; use crate::identity::NetworkId;
@@ -44,7 +45,6 @@ use super::device::{PeerSummary, WireguardDevice};
use super::keys::{WgPublicKey, WgSecretKey}; use super::keys::{WgPublicKey, WgSecretKey};
use super::store::WgKeyStore; use super::store::WgKeyStore;
use crate::overlay::config::{DEFAULT_INTERFACE_PREFIX, interface_name}; use crate::overlay::config::{DEFAULT_INTERFACE_PREFIX, interface_name};
use crate::overlay::tun::{TunFactory, TunRequest};
use crate::state::Ipv4Range; use crate::state::Ipv4Range;
/// The protocol identifier this plugin announces. /// The protocol identifier this plugin announces.
@@ -150,12 +150,6 @@ pub struct NetworkOverview {
pub ipv4_range: Option<Ipv4Range>, pub ipv4_range: Option<Ipv4Range>,
/// Peers this agent knows about. /// Peers this agent knows about.
pub peers: Vec<PeerOverview>, pub peers: Vec<PeerOverview>,
/// 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<IpAddr>,
} }
impl NetworkOverview { impl NetworkOverview {
@@ -202,12 +196,6 @@ struct NetworkState {
ipv4_range: Option<Ipv4Range>, ipv4_range: Option<Ipv4Range>,
/// The address last reported as missing, so it is said once, not forever. /// The address last reported as missing, so it is said once, not forever.
reported_missing_v4: Option<Ipv4Addr>, reported_missing_v4: Option<Ipv4Addr>,
/// 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<TunRequest>,
} }
#[derive(Debug, Default)] #[derive(Debug, Default)]
@@ -232,16 +220,16 @@ struct Worker {
config: WireguardConfig, config: WireguardConfig,
/// This agent's endpoint id, learned when the plugin is attached. /// This agent's endpoint id, learned when the plugin is attached.
local_id: OnceLock<EndpointId>, local_id: OnceLock<EndpointId>,
tun_factory: Arc<dyn TunFactory>,
store: WgKeyStore, store: WgKeyStore,
shared: Mutex<Shared>, shared: Mutex<Shared>,
context: OnceLock<PluginContext>, context: OnceLock<PluginContext>,
/// Where decrypted packets go, once the agent has attached one.
sink: OnceLock<Arc<dyn PacketSink>>,
} }
impl std::fmt::Debug for Worker { impl std::fmt::Debug for Worker {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Worker") f.debug_struct("Worker")
.field("tun", &self.tun_factory.name())
.field("store", &self.store.path()) .field("store", &self.store.path())
.finish() .finish()
} }
@@ -260,10 +248,7 @@ impl WireguardPlugin {
/// ///
/// Must be called from inside a tokio runtime; the plugin starts no /// Must be called from inside a tokio runtime; the plugin starts no
/// runtime of its own. /// runtime of its own.
pub async fn open( pub async fn open(config: WireguardConfig) -> Result<Arc<Self>, PluginError> {
config: WireguardConfig,
tun_factory: Arc<dyn TunFactory>,
) -> Result<Arc<Self>, PluginError> {
// Validate the prefix once, here, rather than failing per network. // Validate the prefix once, here, rather than failing per network.
interface_name(&config.interface_prefix, NetworkId::from_bytes([0u8; 32]))?; interface_name(&config.interface_prefix, NetworkId::from_bytes([0u8; 32]))?;
@@ -283,10 +268,10 @@ impl WireguardPlugin {
let worker = Arc::new(Worker { let worker = Arc::new(Worker {
config, config,
local_id: OnceLock::new(), local_id: OnceLock::new(),
tun_factory,
store, store,
shared: Mutex::new(Shared::default()), shared: Mutex::new(Shared::default()),
context: OnceLock::new(), context: OnceLock::new(),
sink: OnceLock::new(),
}); });
let (commands, receiver) = mpsc::channel(64); let (commands, receiver) = mpsc::channel(64);
@@ -337,20 +322,6 @@ impl WireguardPlugin {
overlay_address_v4: state.allocations.get(&self.worker.local_id()).copied(), overlay_address_v4: state.allocations.get(&self.worker.local_id()).copied(),
ipv4_range: state.ipv4_range, ipv4_range: state.ipv4_range,
peers, 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(), allocations: HashMap::new(),
ipv4_range: None, ipv4_range: None,
reported_missing_v4: None, reported_missing_v4: None,
applied: None,
}); });
} }
@@ -449,75 +419,35 @@ impl Worker {
Ok(true) Ok(true)
} }
/// What the host interface for a network should look like. /// Starts this network's tunnels.
fn desired_request(&self, state: &NetworkState) -> TunRequest { ///
let own_range = state.ipv4_range; /// No interface is created: one agent has one, it belongs to the system
TunRequest { /// level, and decrypted packets are handed there rather than written
name: state.interface.clone(), /// out.
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.
async fn ensure_device(&self, network: NetworkId) -> Result<(), PluginError> { async fn ensure_device(&self, network: NetworkId) -> Result<(), PluginError> {
let (request, key) = { let key = {
let shared = self.lock_shared(); let shared = self.lock_shared();
match shared.networks.get(&network) { match shared.networks.get(&network) {
Some(state) if state.device.is_none() => { Some(state) if state.device.is_none() => state.key.clone(),
(self.desired_request(state), state.key.clone())
}
_ => return Ok(()), _ => return Ok(()),
} }
}; };
let applied = request.clone(); let sink = match self.sink.get() {
let tun = self.tun_factory.create(request).await?; Some(sink) => Arc::clone(sink),
let device = Arc::new(WireguardDevice::start(network, key, tun)); // Not attached to an agent: the protocol still runs, and its
// packets have nowhere to go.
None => Arc::new(crate::dataplane::DiscardPackets) as Arc<dyn PacketSink>,
};
let device = Arc::new(WireguardDevice::start(network, key, sink));
let mut shared = self.lock_shared(); let mut shared = self.lock_shared();
if let Some(state) = shared.networks.get_mut(&network) { if let Some(state) = shared.networks.get_mut(&network) {
state.interface = device.interface().to_string();
state.device = Some(device); state.device = Some(device);
state.applied = Some(applied);
} }
Ok(()) 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. /// Brings the running tunnels in line with what is known.
/// ///
/// A peer gets a tunnel once both halves have arrived: its announcement, /// 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 // kernel to remove an interface this agent created; the explicit
// destroy makes that immediate and definite rather than dependent on // destroy makes that immediate and definite rather than dependent on
// the last reader letting go. // the last reader letting go.
let removed = self.lock_shared().networks.remove(&network); // Dropping the state drops the tunnels, which stops their tasks and
if let Some(state) = removed { // closes their links. There is no interface to remove: the agent owns
drop(state.device); // it, and it outlives any one network.
self.tun_factory.destroy(&state.interface).await; self.lock_shared().networks.remove(&network);
}
} }
fn known_networks(&self) -> Vec<NetworkId> { fn known_networks(&self) -> Vec<NetworkId> {
@@ -707,7 +636,6 @@ async fn run(worker: Arc<Worker>, mut commands: mpsc::Receiver<Command>) {
}, if wait_until.is_some() => { }, if wait_until.is_some() => {
deadline = None; deadline = None;
for network in std::mem::take(&mut pending) { for network in std::mem::take(&mut pending) {
worker.ensure_addresses(network).await;
worker.sync(network); worker.sync(network);
} }
} }
@@ -718,7 +646,6 @@ async fn run(worker: Arc<Worker>, mut commands: mpsc::Receiver<Command>) {
if let Err(err) = worker.ensure_device(network).await { if let Err(err) = worker.ensure_device(network).await {
tracing::debug!(%err, "packet interface still unavailable"); tracing::debug!(%err, "packet interface still unavailable");
} }
worker.ensure_addresses(network).await;
worker.sync(network); worker.sync(network);
} }
} }
@@ -735,6 +662,7 @@ impl IpPlugin for WireguardPlugin {
if let Some(local) = context.local_endpoint_id() { if let Some(local) = context.local_endpoint_id() {
let _ = self.worker.local_id.set(local); let _ = self.worker.local_id.set(local);
} }
let _ = self.worker.sink.set(context.packet_sink());
let _ = self.worker.context.set(context); 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) { fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) {
self.nudge(Command::Link { self.nudge(Command::Link {
network, network,
+60 -6
View File
@@ -123,20 +123,28 @@ impl Tally {
#[derive(Debug)] #[derive(Debug)]
pub struct Interface { pub struct Interface {
device: Arc<dyn TunDevice>, device: Arc<dyn TunDevice>,
factory: Arc<dyn TunFactory>,
name: String,
mtu: u32,
routes: Arc<RoutingTable>, routes: Arc<RoutingTable>,
tally: Arc<Tally>, tally: Arc<Tally>,
/// What was last applied to the host, so an unchanged table is not
/// re-applied on every pass.
applied: std::sync::Mutex<Vec<Cidr>>,
task: std::sync::Mutex<Option<JoinHandle<()>>>, task: std::sync::Mutex<Option<JoinHandle<()>>>,
} }
impl Interface { impl Interface {
/// Creates the interface and starts moving packets. /// Creates the interface and starts moving packets.
pub async fn start( pub async fn start(
factory: &dyn TunFactory, factory: Arc<dyn TunFactory>,
request: TunRequest, name: impl Into<String>,
mtu: u32,
routes: Arc<RoutingTable>, routes: Arc<RoutingTable>,
carrier: Arc<dyn PacketCarrier>, carrier: Arc<dyn PacketCarrier>,
) -> Result<Self, OverlayError> { ) -> Result<Self, OverlayError> {
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 tally = Arc::new(Tally::default());
let task = { let task = {
@@ -172,12 +180,52 @@ impl Interface {
Ok(Self { Ok(Self {
device, device,
factory,
name,
mtu,
routes, routes,
tally, tally,
applied: std::sync::Mutex::new(Vec::new()),
task: std::sync::Mutex::new(Some(task)), 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. /// The interface name the operating system gave.
pub fn name(&self) -> &str { pub fn name(&self) -> &str {
self.device.name() self.device.name()
@@ -188,6 +236,11 @@ impl Interface {
self.device.mtu() 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. /// The counters as they stand.
pub fn counters(&self) -> Counters { pub fn counters(&self) -> Counters {
self.tally.snapshot() self.tally.snapshot()
@@ -355,10 +408,11 @@ mod tests {
carried: std::sync::Mutex::new(Vec::new()), carried: std::sync::Mutex::new(Vec::new()),
refuse, refuse,
}); });
let factory = MemoryTunFactory::new(); let factory = Arc::new(MemoryTunFactory::new());
let interface = Interface::start( let interface = Interface::start(
&factory, Arc::clone(&factory) as Arc<dyn TunFactory>,
TunRequest::bare("tsuntest", 1280), "tsuntest",
1280,
Arc::clone(&routes), Arc::clone(&routes),
Arc::clone(&carrier) as Arc<dyn PacketCarrier>, Arc::clone(&carrier) as Arc<dyn PacketCarrier>,
) )
+36
View File
@@ -175,6 +175,23 @@ impl RoutingTable {
addresses 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<Ipv4Range> {
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. /// How many networks the table covers.
pub fn len(&self) -> usize { pub fn len(&self) -> usize {
self.read().len() self.read().len()
@@ -319,6 +336,25 @@ mod tests {
assert_eq!(table.route(addr(2)).map(|route| route.network), Some(first)); 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] #[test]
fn a_nested_range_counts_as_overlapping() { fn a_nested_range_counts_as_overlapping() {
let table = RoutingTable::new(); let table = RoutingTable::new();
+24 -16
View File
@@ -33,7 +33,7 @@ use tsunagi::{Agent, NetworkStatus};
struct HostedAgent { struct HostedAgent {
_dir: TempDir, _dir: TempDir,
agent: Agent, agent: Agent,
plugin: Arc<WireguardPlugin>, _plugin: Arc<WireguardPlugin>,
host: MockHost, host: MockHost,
} }
@@ -43,12 +43,12 @@ impl HostedAgent {
let provisioner = Arc::new(MockProvisioner::new(host.clone())); let provisioner = Arc::new(MockProvisioner::new(host.clone()));
let factory = Arc::new(ManagedTunFactory::new(provisioner)); let factory = Arc::new(ManagedTunFactory::new(provisioner));
let config = WireguardConfig::new(dir.path().join("wireguard")) let config = WireguardConfig::new(dir.path().join("wireguard"))
.with_interface_prefix(tag)
.with_reconcile(Duration::from_millis(20), Duration::from_millis(100)); .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( let agent = Agent::spawn(
config_with(dir.path(), discovery) config_with(dir.path(), discovery)
.with_overlay_ipv4_range(Some(DEFAULT_IPV4_RANGE)) .with_overlay_ipv4_range(Some(DEFAULT_IPV4_RANGE))
.with_interface(factory, tag, 1280)
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>), .with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
) )
.await .await
@@ -56,17 +56,18 @@ impl HostedAgent {
Self { Self {
_dir: dir, _dir: dir,
agent, agent,
plugin, _plugin: plugin,
host, host,
} }
} }
/// The interface name the plugin settled on for a network. /// The interface name the plugin settled on for a network.
async fn interface(&self, network: NetworkId) -> String { /// The one interface this agent owns. Not per network.
wait_until("the plugin named its interface", || async { async fn interface(&self, _network: NetworkId) -> String {
self.plugin wait_until("the agent named its interface", || async {
.overview(network) self.agent
.map(|view| view.interface) .overlay()
.map(|overlay| overlay.interface)
.filter(|name| !name.is_empty()) .filter(|name| !name.is_empty())
}) })
.await .await
@@ -217,7 +218,10 @@ async fn an_interface_belonging_to_something_else_is_left_alone() {
} }
#[tokio::test] #[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 discovery = SharedMemoryDiscovery::new();
let (name, secret) = network("provision-cleanup"); let (name, secret) = network("provision-cleanup");
let agent = HostedAgent::spawn(&discovery, "tsunx", MockHost::new()).await; 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 network_id = agent.agent.join_network(&name, &secret).await.unwrap();
let interface = agent.interface(network_id).await; let interface = agent.interface(network_id).await;
agent 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; .await;
agent.agent.deactivate_network(network_id).await.unwrap(); agent.agent.deactivate_network(network_id).await.unwrap();
agent let state = agent
.wait_for_host("the interface to be removed", &interface, |state| { .wait_for_host("the address to be withdrawn", &interface, |state| {
state.is_none().then_some(()) state.filter(|state| !addressed(state))
}) })
.await; .await;
assert_eq!(state.kind, LinkKind::Tun, "the interface is still there");
// And it goes when the agent does.
agent.agent.shutdown().await;
assert!( assert!(
agent.host.names().is_empty(), agent.host.names().is_empty(),
"nothing is left behind: {:?}", "nothing is left behind: {:?}",
agent.host.names() agent.host.names()
); );
agent.agent.shutdown().await;
} }
+11 -8
View File
@@ -46,8 +46,11 @@ fn source(agent: Agent, plugin: Arc<WireguardPlugin>) -> Arc<dyn tsunagi::ipc::u
.collect(), .collect(),
overlay: plugin.overview(net.network_id).map(|view| { overlay: plugin.overview(net.network_id).map(|view| {
tsunagi::ipc::OverlayReport { tsunagi::ipc::OverlayReport {
interface: view.interface.clone(), // The interface belongs to the agent now.
mtu: view.mtu, interface: agent
.overlay()
.map_or_else(String::new, |overlay| overlay.interface),
mtu: agent.overlay().map_or(0, |overlay| overlay.mtu),
address: view.overlay_address_v4.map(|a| a.to_string()), address: view.overlay_address_v4.map(|a| a.to_string()),
prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len), prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len),
peers: view peers: view
@@ -95,14 +98,14 @@ async fn a_client_sees_the_agent_and_its_overlay() {
let tuns = MemoryTunFactory::new(); let tuns = MemoryTunFactory::new();
let plugin = WireguardPlugin::open( let plugin = WireguardPlugin::open(
WireguardConfig::new(dir_a.path().join("wg")) WireguardConfig::new(dir_a.path().join("wg"))
.with_interface_prefix("tca")
.with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)),
Arc::new(tuns),
) )
.await .await
.unwrap(); .unwrap();
let agent = Agent::spawn( let agent = Agent::spawn(
config_with(dir_a.path(), &discovery).with_plugin(plugin.clone() as Arc<dyn IpPlugin>), config_with(dir_a.path(), &discovery)
.with_interface(Arc::new(tuns), "tca0", 1280)
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
) )
.await .await
.unwrap(); .unwrap();
@@ -110,14 +113,14 @@ async fn a_client_sees_the_agent_and_its_overlay() {
let dir_b = TempDir::new().unwrap(); let dir_b = TempDir::new().unwrap();
let plugin_b = WireguardPlugin::open( let plugin_b = WireguardPlugin::open(
WireguardConfig::new(dir_b.path().join("wg")) WireguardConfig::new(dir_b.path().join("wg"))
.with_interface_prefix("tcb")
.with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)),
Arc::new(MemoryTunFactory::new()),
) )
.await .await
.unwrap(); .unwrap();
let agent_b = Agent::spawn( let agent_b = Agent::spawn(
config_with(dir_b.path(), &discovery).with_plugin(plugin_b.clone() as Arc<dyn IpPlugin>), config_with(dir_b.path(), &discovery)
.with_interface(Arc::new(MemoryTunFactory::new()), "tcb0", 1280)
.with_plugin(plugin_b.clone() as Arc<dyn IpPlugin>),
) )
.await .await
.unwrap(); .unwrap();
+63 -48
View File
@@ -106,15 +106,15 @@ impl WgAgent {
let tuns = MemoryTunFactory::new(); let tuns = MemoryTunFactory::new();
let wg = tune( let wg = tune(
WireguardConfig::new(root.join("wireguard")) WireguardConfig::new(root.join("wireguard"))
.with_interface_prefix(tag)
.with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)),
); );
let plugin = WireguardPlugin::open(wg, Arc::new(tuns.clone())) let plugin = WireguardPlugin::open(wg).await.unwrap();
.await // The interface belongs to the agent now, so the tag names it
.unwrap(); // directly rather than prefixing one per network.
let agent = Agent::spawn( let agent = Agent::spawn(
config_with(root, discovery) config_with(root, discovery)
.with_overlay_ipv4_range(range) .with_overlay_ipv4_range(range)
.with_interface(Arc::new(tuns.clone()), tag, 1280)
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>), .with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
) )
.await .await
@@ -137,14 +137,16 @@ impl WgAgent {
.await .await
} }
/// The in-memory packet interface for a network. /// The one in-memory packet interface this agent owns.
async fn tun(&self, network: NetworkId) -> Arc<MemoryTun> { ///
/// 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<MemoryTun> {
let name = wait_until("the packet interface exists", || async { let name = wait_until("the packet interface exists", || async {
let view = self.plugin.overview(network)?; self.agent.overlay().map(|overlay| overlay.interface)
self.tuns.device(&view.interface).map(|_| view.interface)
}) })
.await; .await;
self.tuns.device(&name).unwrap() self.tuns.device(&name).expect("the device was created")
} }
/// Waits until `count` tunnels have completed a WireGuard handshake. /// 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")); 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. // 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 { wait_until("the spoofed packet is dropped", || async {
let view = b.plugin.overview(network_id)?; (b.agent.overlay()?.counters.wrong_source >= 1).then_some(())
let tunnel = view.peers.first()?.tunnel.as_ref()?;
(tunnel.stats.dropped_wrong_source >= 1).then_some(())
}) })
.await; .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")); tun_a.push_from_os(ipv4_packet(forged, v4_b, b"spoofed v4"));
wait_until("the spoofed IPv4 packet is dropped", || async { wait_until("the spoofed IPv4 packet is dropped", || async {
let view = b.plugin.overview(network_id)?; (b.agent.overlay()?.counters.wrong_source >= 1).then_some(())
let tunnel = view.peers.first()?.tunnel.as_ref()?;
(tunnel.stats.dropped_wrong_source >= 1).then_some(())
}) })
.await; .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 { wait_until("the packet is counted as unroutable", || async {
let view = a.plugin.overview(network_id)?; let view = a.plugin.overview(network_id)?;
(view.unroutable_packets >= 1).then_some(()) let _ = view;
(a.agent.overlay()?.counters.unroutable >= 1).then_some(())
}) })
.await; .await;
@@ -712,34 +714,61 @@ async fn a_departing_peer_loses_its_tunnel() {
} }
#[tokio::test] #[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 discovery = SharedMemoryDiscovery::new();
let (name_a, secret_a) = network("wg-left"); let (name_a, secret_a) = network("wg-left");
let (name_b, secret_b) = network("wg-right"); let (name_b, secret_b) = network("wg-right");
let beta_range = Some("10.99.0.0/16".parse::<Ipv4Range>().unwrap());
let hub = WgAgent::spawn(&discovery, "th").await; let hub = WgAgent::spawn(&discovery, "th").await;
let left = WgAgent::spawn(&discovery, "tl").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(); // The other members settle each range before the hub joins, so it has
let beta = hub.agent.join_network(&name_b, &secret_b).await.unwrap(); // something to adopt rather than a default to collide with.
left.agent.join_network(&name_a, &secret_a).await.unwrap(); let alpha = left.agent.join_network(&name_a, &secret_a).await.unwrap();
right.agent.join_network(&name_b, &secret_b).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(alpha, 1).await;
hub.wait_for_tunnels(beta, 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!( 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" "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. // Each network's address comes from its own range.
let hub_alpha = hub.overlay(alpha).await; 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; let left_addr = left.overlay(alpha).await;
hub.tun(alpha) hub.tun(alpha)
.await .await
@@ -759,14 +788,6 @@ async fn two_networks_get_separate_interfaces_keys_and_overlays() {
"the other overlay must see nothing" "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; hub.shutdown().await;
left.shutdown().await; left.shutdown().await;
right.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 // IPv6's sake; the overlay is IPv4 now and a relayed path with small
// datagrams can be matched instead of warned about. // datagrams can be matched instead of warned about.
let dir = TempDir::new().unwrap(); let dir = TempDir::new().unwrap();
let result = WireguardPlugin::open( let result =
WireguardConfig::new(dir.path()).with_mtu(MIN_MTU - 1), WireguardPlugin::open(WireguardConfig::new(dir.path()).with_mtu(MIN_MTU - 1)).await;
Arc::new(MemoryTunFactory::new()),
)
.await;
match result { match result {
Err(err) => { Err(err) => {
let text = err.to_string(); 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) }; const { assert!(DEFAULT_MTU > MIN_MTU) };
assert_eq!(WIREGUARD_OVERHEAD, 32); assert_eq!(WIREGUARD_OVERHEAD, 32);
assert!( assert!(
WireguardPlugin::open( WireguardPlugin::open(WireguardConfig::new(dir.path().join("ok")))
WireguardConfig::new(dir.path().join("ok")), .await
Arc::new(MemoryTunFactory::new()), .is_ok()
)
.await
.is_ok()
); );
} }