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, }))