From 7c1be332e3efb82466ed65dab67e870297c09c35 Mon Sep 17 00:00:00 2001 From: tsunagi Date: Mon, 21 Sep 2026 19:34:38 +0100 Subject: [PATCH] Agree a protocol by name and wire version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A peer's announcement already carried a protocol and a version; only the name was being checked. Now both are, and a peer offering a protocol at a version this build does not speak simply has no data plane — the control plane keeps working, messages and signed state still flow, and the difference is reported once rather than on every announcement. The version compared is the *wire* version, not the software version, and the trait says so: a plugin crate has its own version and it is nobody else's business. Two peers on different releases work together for as long as the bytes between them have not changed, and nothing in the negotiation may be derived from anything that moves with a release. A test pins that agreement turns on the name and version alone, with a peer whose announcement carries a payload this build has never seen. `PeerStatus` gained the protocols agreed with each peer, so an empty list is visible as what it is: a session that is up, carrying control traffic, with no protocol in common. The forging test plugin was announcing version 1 while claiming to speak the current one, so its payload was being set aside for the wrong reason. Co-Authored-By: Claude Opus 5 (1M context) --- crates/tsunagi/src/agent/network.rs | 55 +++++- crates/tsunagi/src/agent/status.rs | 6 + crates/tsunagi/src/dataplane/mod.rs | 40 +++++ crates/tsunagi/src/dataplane/wireguard/mod.rs | 2 +- .../tsunagi/src/dataplane/wireguard/plugin.rs | 17 ++ crates/tsunagi/tests/wireguard.rs | 159 +++++++++++++++++- 6 files changed, 272 insertions(+), 7 deletions(-) diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 8e19222..8fdb229 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -210,6 +210,9 @@ struct Runtime { opening: HashSet<(EndpointId, String)>, link_results_tx: mpsc::Sender, link_results_rx: mpsc::Receiver, + /// Peers already told about a protocol version that cannot match, so it + /// is said once rather than on every announcement. + reported_mismatch: HashSet<(EndpointId, String)>, /// Signed records, merged from every replica we have talked to. state: StateSet, /// Snapshots received while dispatching, handled on the next loop pass. @@ -242,6 +245,7 @@ impl Runtime { opening: HashSet::new(), link_results_tx, link_results_rx, + reported_mismatch: HashSet::new(), state: StateSet::new(), pending_state: Vec::new(), own_version: 0, @@ -887,12 +891,12 @@ impl Runtime { // ------------------------------------------------------------ data plane - /// Protocol ids this agent has a plugin for. - fn served_protocols(&self) -> Vec { + /// The protocols this agent speaks, with the version it speaks them at. + fn served_protocols(&self) -> Vec<(String, u16)> { self.params .plugins .iter() - .map(|plugin| plugin.protocol_id().to_string()) + .map(|plugin| (plugin.protocol_id().to_string(), plugin.protocol_version())) .collect() } @@ -925,6 +929,10 @@ impl Runtime { }); } + // Both the name and the version have to match. A peer offering + // `wg-quic` at a version this build does not speak is not a peer to + // carry traffic with, and a link opened anyway would fail later and + // say less about why. let wanted: Vec<(EndpointId, String)> = self .sessions .values() @@ -934,9 +942,14 @@ impl Runtime { .capabilities .iter() .filter(|capability| capability.enabled) - .map(move |capability| (peer, capability.protocol.clone())) + .map(move |capability| (peer, capability.protocol.clone(), capability.version)) }) - .filter(|(_, protocol)| served.contains(protocol)) + .filter(|(_, protocol, version)| { + served + .iter() + .any(|(name, ours)| name == protocol && ours == version) + }) + .map(|(peer, protocol, _)| (peer, protocol)) .collect(); for (peer, protocol) in wanted { @@ -1271,6 +1284,26 @@ impl Runtime { if plugin.protocol_id() != capability.protocol { continue; } + if plugin.protocol_version() != capability.version { + // Said once per peer and protocol. A version that will + // not match this build will not match it on the next + // announcement either, and repeating it every round + // would bury everything else. + let key = (peer, capability.protocol.clone()); + if self.reported_mismatch.insert(key) { + errors.push(( + capability.protocol.clone(), + format!( + "peer speaks {} version {} and this build speaks {}, so there \ + is no data plane with it; the control plane is unaffected", + capability.protocol, + capability.version, + plugin.protocol_version() + ), + )); + } + continue; + } // The core hands the opaque payload over without interpreting it. if let Err(err) = plugin.on_peer_capability(self.network_id, peer, capability) { errors.push((plugin.protocol_id().to_string(), err.to_string())); @@ -1290,6 +1323,7 @@ impl Runtime { // ----------------------------------------------------------------- status fn status(&self) -> NetworkStatus { + let served = self.served_protocols(); let mut peers: Vec = self .sessions .values() @@ -1299,6 +1333,17 @@ impl Runtime { endpoint_id: session.peer, role: session.role, hostname: session.hostname.clone(), + protocols: session + .capabilities + .iter() + .filter(|capability| capability.enabled) + .filter(|capability| { + served.iter().any(|(name, ours)| { + *name == capability.protocol && *ours == capability.version + }) + }) + .map(|capability| capability.protocol.clone()) + .collect(), capabilities: session.capabilities.clone(), connected_for: session.established.elapsed(), paths: snapshot.paths, diff --git a/crates/tsunagi/src/agent/status.rs b/crates/tsunagi/src/agent/status.rs index cf0446b..3030a47 100644 --- a/crates/tsunagi/src/agent/status.rs +++ b/crates/tsunagi/src/agent/status.rs @@ -83,6 +83,12 @@ pub struct PeerStatus { /// /// A mutable binding, not an identity. pub hostname: Option, + /// Protocols agreed with this peer: both sides have them at the same + /// version. + /// + /// Empty means no data plane with that peer, which leaves the control + /// plane working — messages and state still flow. + pub protocols: Vec, /// Capabilities the peer announced. Payloads stay opaque. pub capabilities: Vec, /// How long the session has been up. diff --git a/crates/tsunagi/src/dataplane/mod.rs b/crates/tsunagi/src/dataplane/mod.rs index 7960ab3..eef06df 100644 --- a/crates/tsunagi/src/dataplane/mod.rs +++ b/crates/tsunagi/src/dataplane/mod.rs @@ -84,6 +84,19 @@ impl PacketSink for DiscardPackets { } } +/// One setting a protocol accepts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ProtocolOption { + /// The key, as written in `key=value`. + pub key: &'static str, + /// A word for the value, for the help line: `BYTES`, `NAME`. + pub value: &'static str, + /// What it does. + pub help: &'static str, + /// What happens when it is not given. + pub default: Option<&'static str>, +} + /// Errors a plugin may return. They are recorded, never fatal for the agent. #[derive(Debug, thiserror::Error)] #[non_exhaustive] @@ -241,6 +254,29 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static { /// Must be non-empty and at most [`MAX_PROTOCOL_ID_LEN`] bytes. fn protocol_id(&self) -> &str; + /// The wire version of this protocol. + /// + /// Two peers carry traffic for each other only when they have the same + /// protocol at the same version. There is no negotiating a middle + /// ground: a protocol either speaks the same words at both ends or it + /// does not, and pretending otherwise produces a link that fails later + /// and less clearly. + /// + /// **Not the software version.** A plugin crate has its own version and + /// it is nobody else's business: two peers on different builds work + /// together for as long as the wire between them has not changed. This + /// number moves only when the bytes do, so it must never be derived from + /// `CARGO_PKG_VERSION` or anything else that moves with a release. + fn protocol_version(&self) -> u16; + + /// The settings this protocol accepts, and what they mean. + /// + /// Declared rather than documented elsewhere, so the agent can list them + /// without knowing anything about the protocol. + fn options(&self) -> &'static [ProtocolOption] { + &[] + } + /// Called once, when the agent starts, before any network is activated. /// /// The plugin keeps the context to ask for re-announcements and to report @@ -370,6 +406,10 @@ impl IpPlugin for TestCapabilityPlugin { &self.protocol } + fn protocol_version(&self) -> u16 { + 1 + } + fn local_capability( &self, _network: NetworkId, diff --git a/crates/tsunagi/src/dataplane/wireguard/mod.rs b/crates/tsunagi/src/dataplane/wireguard/mod.rs index d088acf..8253497 100644 --- a/crates/tsunagi/src/dataplane/wireguard/mod.rs +++ b/crates/tsunagi/src/dataplane/wireguard/mod.rs @@ -42,7 +42,7 @@ pub mod plugin; pub mod store; pub use crate::state::Ipv4Range; -pub use announcement::{ValidatedAnnouncement, WgAnnouncement}; +pub use announcement::{ANNOUNCEMENT_VERSION, ValidatedAnnouncement, WgAnnouncement}; // The interface, its addresses and how it is created belong to the system // level now: one agent has one interface, and no protocol owns it. Re-exported // here while callers are moved over. diff --git a/crates/tsunagi/src/dataplane/wireguard/plugin.rs b/crates/tsunagi/src/dataplane/wireguard/plugin.rs index 14a4c82..652b85d 100644 --- a/crates/tsunagi/src/dataplane/wireguard/plugin.rs +++ b/crates/tsunagi/src/dataplane/wireguard/plugin.rs @@ -244,6 +244,15 @@ pub struct WireguardPlugin { } impl WireguardPlugin { + /// The settings this protocol accepts. + pub const OPTIONS: &'static [crate::dataplane::ProtocolOption] = + &[crate::dataplane::ProtocolOption { + key: "keepalive", + value: "SECONDS", + help: "persistent keepalive interval; 0 turns it off", + default: Some("25"), + }]; + /// Opens the plugin's key store and starts its reconciliation task. /// /// Must be called from inside a tokio runtime; the plugin starts no @@ -658,6 +667,14 @@ impl IpPlugin for WireguardPlugin { WIREGUARD_PROTOCOL } + fn protocol_version(&self) -> u16 { + super::announcement::ANNOUNCEMENT_VERSION + } + + fn options(&self) -> &'static [crate::dataplane::ProtocolOption] { + Self::OPTIONS + } + fn attach(&self, context: PluginContext) { if let Some(local) = context.local_endpoint_id() { let _ = self.worker.local_id.set(local); diff --git a/crates/tsunagi/tests/wireguard.rs b/crates/tsunagi/tests/wireguard.rs index f609662..3f0a900 100644 --- a/crates/tsunagi/tests/wireguard.rs +++ b/crates/tsunagi/tests/wireguard.rs @@ -942,6 +942,155 @@ async fn an_announcement_for_another_network_never_reaches_a_tunnel() { drop(attacker_dir); } +#[tokio::test] +async fn a_peer_at_another_protocol_version_gets_no_data_plane_and_keeps_the_control_plane() { + // Both sides must have the protocol at the same version. There is no + // middle ground to negotiate: either the words mean the same thing at + // both ends or they do not. + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-version"); + + let here = WgAgent::spawn(&discovery, "tvh").await; + let network_id = here.agent.join_network(&name, &secret).await.unwrap(); + + let ahead = Arc::new(FromTheFuture); + let dir = TempDir::new().unwrap(); + let other = Agent::spawn( + config_with(dir.path(), &discovery).with_plugin(ahead.clone() as Arc), + ) + .await + .unwrap(); + + let mut events = here.agent.subscribe(); + other.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&here.agent, network_id, 1).await; + + let reason = wait_event(&mut events, |event| match event { + Event::PluginError { reason, .. } if reason.contains("version") => Some(reason.clone()), + _ => None, + }) + .await; + assert!( + reason.contains("control plane is unaffected"), + "the message should say what still works: {reason}" + ); + + // No tunnel, and no attempt at one. + settle().await; + assert!( + here.plugin.overview(network_id).unwrap().peers.is_empty(), + "a version that cannot match must not become a tunnel" + ); + let status = here.agent.network_status(network_id).await.unwrap(); + assert_eq!(status.peers.len(), 1, "the session is up all the same"); + assert!( + status.peers[0].protocols.is_empty(), + "nothing was agreed with it: {:?}", + status.peers[0].protocols + ); + + // And the control plane carries a message to it regardless. + assert_eq!( + here.agent + .broadcast( + network_id, + tsunagi::proto::ControlMessage::Ping { + seq: 1, + payload: b"still talking".to_vec(), + }, + ) + .await + .unwrap(), + 1 + ); + + other.shutdown().await; + here.shutdown().await; + drop(dir); +} + +#[tokio::test] +async fn agreement_turns_on_the_protocol_version_and_nothing_else() { + // A different build is not a different protocol. What is compared is the + // wire version and the name; everything else about a peer — its release, + // and whatever opaque payload its announcement carries — has no say. + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-same-wire"); + + let here = WgAgent::spawn(&discovery, "tsw").await; + let network_id = here.agent.join_network(&name, &secret).await.unwrap(); + + // Announces the right protocol at the right version, with a payload + // this build has never seen. + let stranger = Arc::new(ForgingPlugin { + payload: std::sync::Mutex::new(Some(b"from some other release".to_vec())), + }); + let dir = TempDir::new().unwrap(); + let other = Agent::spawn( + config_with(dir.path(), &discovery).with_plugin(stranger.clone() as Arc), + ) + .await + .unwrap(); + other.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&here.agent, network_id, 1).await; + + // Agreed, because the wire matches. The payload is rejected by the + // protocol afterwards, which is a separate matter and says nothing about + // whether the two agreed to talk. + wait_until("the protocol is agreed with it", || async { + let status = here.agent.network_status(network_id).await.ok()?; + status + .peers + .first()? + .protocols + .contains(&WIREGUARD_PROTOCOL.to_string()) + .then_some(()) + }) + .await; + + other.shutdown().await; + here.shutdown().await; + drop(dir); +} + +/// A plugin claiming the same protocol at a version this build does not speak. +#[derive(Debug)] +struct FromTheFuture; + +impl IpPlugin for FromTheFuture { + fn protocol_id(&self) -> &str { + WIREGUARD_PROTOCOL + } + + fn protocol_version(&self) -> u16 { + tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION + 1 + } + + fn local_capability( + &self, + _network: NetworkId, + ) -> Result, tsunagi::dataplane::PluginError> { + Ok(Some(tsunagi::dataplane::PluginCapability { + protocol: WIREGUARD_PROTOCOL.to_string(), + version: self.protocol_version(), + enabled: true, + data: Vec::new(), + })) + } + + fn on_peer_capability( + &self, + _network: NetworkId, + _peer: EndpointId, + _capability: &tsunagi::dataplane::PluginCapability, + ) -> Result<(), tsunagi::dataplane::PluginError> { + Ok(()) + } + + fn on_peer_gone(&self, _network: NetworkId, _peer: EndpointId) {} + fn on_network_deactivated(&self, _network: NetworkId) {} +} + /// A plugin that announces whatever bytes it is told to, under the WireGuard /// protocol id. Used to test what a hostile member can do. #[derive(Debug)] @@ -954,6 +1103,12 @@ impl IpPlugin for ForgingPlugin { WIREGUARD_PROTOCOL } + /// The same version, so the announcement is looked at rather than + /// dismissed for the wrong reason. + fn protocol_version(&self) -> u16 { + tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION + } + fn local_capability( &self, _network: NetworkId, @@ -964,7 +1119,9 @@ impl IpPlugin for ForgingPlugin { }; Ok(payload.map(|data| tsunagi::dataplane::PluginCapability { protocol: WIREGUARD_PROTOCOL.to_string(), - version: 1, + // The version it actually speaks, so the payload is examined + // rather than set aside for the wrong reason. + version: tsunagi::dataplane::wireguard::ANNOUNCEMENT_VERSION, enabled: true, data, }))