Separate control and data logically, move WireGuard into userspace, add a CLI

Corrects the architecture on two points raised in review, while the project
is still small enough to change cheaply.

1. Control and data are separated *logically*, not physically.

The old reading — "nothing but control may ride on iroh" — threw away iroh's
whole value and would have forced the data plane to reimplement STUN, ICE and
a relay. Now both planes ride on iroh with different ALPNs and different
connections, so the data plane inherits hole punching and relay fallback,
while proto/ still knows nothing about packets and dataplane/ knows nothing
about the control protocol.

New boundary: PacketTransport / PacketLink, an authenticated unreliable
datagram channel per (network, peer, protocol). tsunagi/data/1 runs the same
membership handshake, then DataOpen/DataOpenAck, then QUIC datagrams. Only
the smaller endpoint id dials, so exactly one link exists per pair.

A plugin is handed links and never learns reachability, so the WireGuard
announcement shrank to a public key: there is no address left to lie about.

2. WireGuard now runs in userspace, on boringtun's protocol state machine.

No kernel module, no wg tool, no ip shell-out, no loopback proxy: the wgtool,
backend and bridge modules are gone. Only creating a TUN device needs
privileges, and that sits behind TunFactory, so the entire data plane —
handshake, encryption, routing, address ownership — is tested with none.

Address ownership is enforced rather than believed: outbound packets go to
the owner of the destination address, inbound packets are dropped unless
their source is the address derived for the peer that sent them.

3. A `tsunagi` binary: secret, doctor, id, up. It owns the runtime, the
logging subscriber and Ctrl-C, which the library still refuses to.

Also fixes a reference cycle where IrohTransport held Arc<Inner>, which kept
the databases open and the directory lock held after shutdown; two storage
tests caught it once the cycle existed.

81 tests pass offline with no privileges, including real IPv6 packets
crossing a real WireGuard tunnel over real iroh connections. Verified by
hand: two CLI processes forming a mesh both on loopback and via n0 discovery
using only an endpoint id.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/agent/events.rs

@@ -100,6 +100,35 @@ pub enum Event {
         /// Why it was discarded. Free of secrets.
         reason: String,
     },
+    /// A data plane link to a peer is up.
+    ///
+    /// The data plane is a separate connection from the control plane; this
+    /// says nothing about the control session, and vice versa.
+    DataLinkUp {
+        /// The network.
+        network: NetworkId,
+        /// The peer.
+        peer: EndpointId,
+        /// Plugin protocol the link carries.
+        protocol: String,
+        /// What the transport reports about the path in use.
+        path: String,
+        /// Largest datagram the link can carry.
+        max_datagram: usize,
+    },
+    /// A data plane link went away or could not be opened.
+    ///
+    /// Never fatal: the control plane keeps running and the link is retried.
+    DataLinkDown {
+        /// The network.
+        network: NetworkId,
+        /// The peer.
+        peer: EndpointId,
+        /// Plugin protocol the link would have carried.
+        protocol: String,
+        /// Why it is not up.
+        reason: String,
+    },
     /// An IP plugin reported an error. Never fatal.
     PluginError {
         /// The network the call was scoped to.

src/agent/mod.rs

@@ -37,6 +37,8 @@ use tokio::sync::{RwLock, broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 
 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::error::{Error, Result};
 use crate::identity::{DeviceIdentity, NetworkId, NetworkKeys, NetworkName, NetworkSecret};
@@ -82,20 +84,47 @@ struct Inner {
     shutdown: Shutdown,
     accept_task: std::sync::Mutex<Option<JoinHandle<()>>>,
     plugin_task: std::sync::Mutex<Option<JoinHandle<()>>>,
+    transport: std::sync::OnceLock<Arc<dyn PacketTransport>>,
 }
 
-impl Drop for Inner {
-    fn drop(&mut self) {
-        // Nothing here awaits; this is only a safety net for a handle that was
-        // dropped without an explicit shutdown.
-        self.shutdown.trigger();
-        for guard in [&self.accept_task, &self.plugin_task] {
-            if let Ok(mut guard) = guard.lock()
-                && let Some(task) = guard.take()
-            {
-                task.abort();
+/// 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.
+#[derive(Debug)]
+struct TransportCtx(Weak<Inner>);
+
+impl TransportContext for TransportCtx {
+    fn snapshot(&self) -> crate::BoxFuture<'_, HashMap<NetworkId, NetworkKeys>> {
+        Box::pin(async move {
+            let Some(inner) = self.0.upgrade() else {
+                return HashMap::new();
+            };
+            inner
+                .networks
+                .read()
+                .await
+                .iter()
+                .map(|(id, handle)| (*id, handle.keys.clone()))
+                .collect()
+        })
+    }
+
+    fn serves<'a>(&'a self, network: NetworkId, protocol: &'a str) -> crate::BoxFuture<'a, bool> {
+        Box::pin(async move {
+            let Some(inner) = self.0.upgrade() else {
+                return false;
+            };
+            if !inner.networks.read().await.contains_key(&network) {
+                return false;
             }
-        }
+            inner
+                .config
+                .plugins
+                .iter()
+                .any(|plugin| plugin.protocol_id() == protocol)
+        })
     }
 }
 
@@ -127,9 +156,20 @@ impl Agent {
             shutdown: Shutdown::new(),
             accept_task: std::sync::Mutex::new(None),
             plugin_task: std::sync::Mutex::new(None),
+            transport: std::sync::OnceLock::new(),
             config,
         });
 
+        // 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.
+        let transport: Arc<dyn PacketTransport> = Arc::new(IrohTransport::new(
+            inner.adapter.clone(),
+            Arc::clone(&inner.limits),
+            Arc::new(TransportCtx(Arc::downgrade(&inner))) as Arc<dyn TransportContext>,
+        ));
+        let _ = inner.transport.set(transport);
+
         if let CacheOutcome::Reset(reason) = inner.storage.cache_outcome().clone() {
             let _ = inner.events.send(Event::CacheReset { reason });
         }
@@ -257,6 +297,7 @@ impl Agent {
             discovery_interval: self.inner.config.discovery_interval,
             plugins: self.inner.config.plugins.clone(),
             hostname: self.inner.hostname.clone(),
+            transport: self.inner.transport.get().cloned(),
         });
         networks.insert(network_id, handle);
         drop(networks);
@@ -625,6 +666,12 @@ async fn handle_incoming(inner: Arc<Inner>, incoming: iroh::endpoint::Incoming)
     };
     let peer = conn.remote_id();
 
+    // Two protocols share the endpoint; they are told apart here and never mix.
+    if conn.alpn() == crate::proto::message::DATA_ALPN {
+        handle_inbound_data(inner, conn).await;
+        return;
+    }
+
     let (mut send, mut recv) = match conn.accept_bi().await {
         Ok(streams) => streams,
         Err(err) => {
@@ -696,6 +743,45 @@ async fn handle_incoming(inner: Arc<Inner>, incoming: iroh::endpoint::Incoming)
     }
 }
 
+/// Completes an inbound data plane connection and routes it to its network.
+async fn handle_inbound_data(inner: Arc<Inner>, conn: iroh::endpoint::Connection) {
+    let Some(transport) = inner.transport.get().cloned() else {
+        conn.close(5u32.into(), b"data plane not ready");
+        return;
+    };
+    // Downcasting is avoided by keeping the accept side on the concrete type.
+    let Some(iroh_transport) = transport.as_ref().as_any().downcast_ref::<IrohTransport>() else {
+        conn.close(5u32.into(), b"unsupported data transport");
+        return;
+    };
+
+    let inbound = match iroh_transport.accept(conn).await {
+        Ok(inbound) => inbound,
+        Err(err) => {
+            tracing::debug!(%err, "inbound data channel rejected");
+            return;
+        }
+    };
+
+    let sender = {
+        let networks = inner.networks.read().await;
+        networks
+            .get(&inbound.network)
+            .map(|handle| handle.commands.clone())
+    };
+    let Some(sender) = sender else {
+        // The network went away while the channel was being set up.
+        return;
+    };
+    if sender
+        .send(NetCommand::InboundLink(Box::new(inbound)))
+        .await
+        .is_err()
+    {
+        tracing::debug!("network runtime stopped before the data link was installed");
+    }
+}
+
 /// Picks the hostname to announce.
 ///
 /// Order: explicit configuration, then what the state store already holds, then

src/agent/network.rs

@@ -5,7 +5,7 @@
 //! explicit [`NetworkId`], so deactivating or breaking one network cannot
 //! disturb another and cannot stop the agent.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -15,6 +15,7 @@ use tokio::sync::{broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 
 use crate::config::{Limits, ReconnectPolicy};
+use crate::dataplane::transport::{InboundLink, PacketTransport, SharedLink};
 use crate::dataplane::{PluginCapability, SharedPlugin};
 use crate::discovery::{Candidate, CandidateSource, NetworkDiscovery};
 use crate::error::{Error, Result};
@@ -50,6 +51,8 @@ pub(crate) enum NetCommand {
         message: ControlMessage,
         reply: oneshot::Sender<usize>,
     },
+    /// A peer opened a data plane link towards us.
+    InboundLink(Box<InboundLink>),
     Status {
         reply: oneshot::Sender<Box<NetworkStatus>>,
     },
@@ -69,6 +72,7 @@ impl std::fmt::Debug for NetCommand {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             NetCommand::Inbound(_) => f.write_str("Inbound"),
+            NetCommand::InboundLink(link) => write!(f, "InboundLink({})", link.protocol),
             NetCommand::Send { peer, message, .. } => {
                 write!(f, "Send({}, {})", peer.fmt_short(), kind(message))
             }
@@ -110,6 +114,15 @@ pub(crate) struct RuntimeParams {
     pub(crate) discovery_interval: Duration,
     pub(crate) plugins: Vec<SharedPlugin>,
     pub(crate) hostname: String,
+    /// How data plane links are opened. `None` disables the data plane.
+    pub(crate) transport: Option<Arc<dyn PacketTransport>>,
+}
+
+/// Outcome of one attempt to open a data plane link.
+struct LinkOutcome {
+    peer: EndpointId,
+    protocol: String,
+    result: Result<SharedLink, String>,
 }
 
 /// Outcome of one outbound dial.
@@ -175,6 +188,12 @@ struct Runtime {
     session_events_rx: mpsc::Receiver<SessionEvent>,
     dial_results_tx: mpsc::Sender<DialOutcome>,
     dial_results_rx: mpsc::Receiver<DialOutcome>,
+    /// Live data plane links, keyed by peer and plugin protocol.
+    links: HashMap<(EndpointId, String), SharedLink>,
+    /// Links currently being opened, so we do not start two.
+    opening: HashSet<(EndpointId, String)>,
+    link_results_tx: mpsc::Sender<LinkOutcome>,
+    link_results_rx: mpsc::Receiver<LinkOutcome>,
 }
 
 impl Runtime {
@@ -183,6 +202,7 @@ impl Runtime {
         let local_id = params.adapter.endpoint_id();
         let (session_events_tx, session_events_rx) = mpsc::channel(256);
         let (dial_results_tx, dial_results_rx) = mpsc::channel(64);
+        let (link_results_tx, link_results_rx) = mpsc::channel(64);
         Self {
             params,
             network_id,
@@ -196,6 +216,10 @@ impl Runtime {
             session_events_rx,
             dial_results_tx,
             dial_results_rx,
+            links: HashMap::new(),
+            opening: HashSet::new(),
+            link_results_tx,
+            link_results_rx,
         }
     }
 
@@ -226,7 +250,15 @@ impl Runtime {
                         self.handle_dial_result(result).await;
                     }
                 }
-                _ = ticker.tick() => self.discovery_round().await,
+                result = self.link_results_rx.recv() => {
+                    if let Some(result) = result {
+                        self.handle_link_result(result);
+                    }
+                }
+                _ = ticker.tick() => {
+                    self.discovery_round().await;
+                    self.ensure_links();
+                }
             }
         }
 
@@ -242,6 +274,7 @@ impl Runtime {
                 .unpublish(self.params.keys.discovery_key(), self.local_id)
                 .await;
         }
+        self.links.clear();
         let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
         for peer in peers {
             if let Some(session) = self.sessions.remove(&peer) {
@@ -260,6 +293,7 @@ impl Runtime {
             NetCommand::Inbound(inbound) => {
                 self.install_session(*inbound).await;
             }
+            NetCommand::InboundLink(inbound) => self.install_link(*inbound),
             NetCommand::Send {
                 peer,
                 message,
@@ -516,6 +550,163 @@ impl Runtime {
         }
     }
 
+    // ------------------------------------------------------------ data plane
+
+    /// Protocol ids this agent has a plugin for.
+    fn served_protocols(&self) -> Vec<String> {
+        self.params
+            .plugins
+            .iter()
+            .map(|plugin| plugin.protocol_id().to_string())
+            .collect()
+    }
+
+    /// Opens whatever data plane links are missing, and forgets dead ones.
+    ///
+    /// Only one side dials, chosen by a rule both sides compute the same way,
+    /// so two agents never open two links for the same thing.
+    fn ensure_links(&mut self) {
+        let Some(transport) = self.params.transport.clone() else {
+            return;
+        };
+        let served = self.served_protocols();
+        if served.is_empty() {
+            return;
+        }
+
+        let mut dead: Vec<(EndpointId, String)> = Vec::new();
+        for (key, link) in &self.links {
+            if link.is_closed() {
+                dead.push(key.clone());
+            }
+        }
+        for (peer, protocol) in dead {
+            self.links.remove(&(peer, protocol.clone()));
+            self.emit(Event::DataLinkDown {
+                network: self.network_id,
+                peer,
+                protocol,
+                reason: "link closed".into(),
+            });
+        }
+
+        let wanted: Vec<(EndpointId, String)> = self
+            .sessions
+            .values()
+            .flat_map(|session| {
+                let peer = session.peer;
+                session
+                    .capabilities
+                    .iter()
+                    .filter(|capability| capability.enabled)
+                    .map(move |capability| (peer, capability.protocol.clone()))
+            })
+            .filter(|(_, protocol)| served.contains(protocol))
+            .collect();
+
+        for (peer, protocol) in wanted {
+            let key = (peer, protocol.clone());
+            if self.links.contains_key(&key) || self.opening.contains(&key) {
+                continue;
+            }
+            // The smaller endpoint id dials; the other side accepts. Both
+            // compute this identically, so exactly one link is created.
+            if self.local_id.as_bytes() >= peer.as_bytes() {
+                continue;
+            }
+            self.opening.insert(key);
+
+            let results = self.link_results_tx.clone();
+            let transport = Arc::clone(&transport);
+            let network = self.network_id;
+            let shutdown = self.shutdown.clone();
+            tokio::spawn(async move {
+                let result = tokio::select! {
+                    biased;
+                    _ = shutdown.wait() => Err("network deactivated".to_string()),
+                    result = transport.open(network, peer, &protocol) => {
+                        result.map_err(|err| err.to_string())
+                    }
+                };
+                let _ = results
+                    .send(LinkOutcome {
+                        peer,
+                        protocol,
+                        result,
+                    })
+                    .await;
+            });
+        }
+    }
+
+    fn handle_link_result(&mut self, outcome: LinkOutcome) {
+        let key = (outcome.peer, outcome.protocol.clone());
+        self.opening.remove(&key);
+        match outcome.result {
+            Ok(link) => self.adopt_link(outcome.peer, outcome.protocol, link),
+            Err(reason) => {
+                // A data plane that cannot be set up is reported, never fatal.
+                self.metrics.data_link_failures += 1;
+                self.emit(Event::DataLinkDown {
+                    network: self.network_id,
+                    peer: outcome.peer,
+                    protocol: outcome.protocol,
+                    reason,
+                });
+            }
+        }
+    }
+
+    fn install_link(&mut self, inbound: InboundLink) {
+        self.adopt_link(inbound.peer, inbound.protocol, inbound.link);
+    }
+
+    /// Hands a link to the plugin that owns its protocol.
+    fn adopt_link(&mut self, peer: EndpointId, protocol: String, link: SharedLink) {
+        let Some(plugin) = self
+            .params
+            .plugins
+            .iter()
+            .find(|plugin| plugin.protocol_id() == protocol)
+            .cloned()
+        else {
+            return;
+        };
+
+        let path = link.path_description();
+        let max_datagram = link.max_datagram_size();
+        self.links
+            .insert((peer, protocol.clone()), Arc::clone(&link));
+        plugin.on_peer_link(self.network_id, peer, link);
+        self.metrics.data_links_established += 1;
+        self.emit(Event::DataLinkUp {
+            network: self.network_id,
+            peer,
+            protocol,
+            path,
+            max_datagram,
+        });
+    }
+
+    /// Drops every link to a peer.
+    fn drop_links_for(&mut self, peer: EndpointId) {
+        let keys: Vec<(EndpointId, String)> = self
+            .links
+            .keys()
+            .filter(|(id, _)| *id == peer)
+            .cloned()
+            .collect();
+        for key in keys {
+            self.links.remove(&key);
+            self.emit(Event::DataLinkDown {
+                network: self.network_id,
+                peer,
+                protocol: key.1,
+                reason: "peer session ended".into(),
+            });
+        }
+    }
+
     // --------------------------------------------------------------- sessions
 
     async fn install_session(&mut self, inbound: InboundSession) {
@@ -682,6 +873,7 @@ impl Runtime {
                     session.conn.close(0u32.into(), b"session ended");
                 }
                 self.metrics.disconnects += 1;
+                self.drop_links_for(peer);
                 for plugin in &self.params.plugins {
                     plugin.on_peer_gone(self.network_id, peer);
                 }
@@ -711,6 +903,7 @@ impl Runtime {
                     session.capabilities = capabilities.clone();
                 }
                 self.dispatch_capabilities(peer, &capabilities);
+                self.ensure_links();
             }
             ControlMessage::Ping { seq, payload } => {
                 let pong = ControlMessage::Pong {

src/agent/status.rs

@@ -94,6 +94,10 @@ pub struct NetworkMetrics {
     pub protocol_violations: u64,
     /// Errors reported by IP plugins. Never fatal.
     pub plugin_errors: u64,
+    /// Data plane links that were established.
+    pub data_links_established: u64,
+    /// Attempts to open a data plane link that failed.
+    pub data_link_failures: u64,
 }
 
 /// Status of one network.