diff --git a/README.md b/README.md index d7cd191..5168766 100644 --- a/README.md +++ b/README.md @@ -112,6 +112,23 @@ fresh masking profile and outer keys for each pair connection attempt, including reentry after fallback. Sustained data loss triggers switching independently of the iroh control connection. [Configuration and limits](docs/amnezia.md). +Change a network's protocol selection without editing its service or restarting: + +```sh +tsng protocols # available protocols and settings +tsng protocols home # this device's choice for home +tsng protocols home awg,wg-quic # prefer AWG, retain QUIC as fallback +tsng protocols home none # control plane only +``` + +Exact network names, full ids and unique id prefixes work here. The choice is +local to this device, saved per network and applied immediately; an offline +change applies on the next start. In the tray, expand **protocols** on a network +tile to toggle protocols or change their preference order. Both endpoints need +a common protocol at the same wire version. `tsng up --protocol` supplies the +initial defaults for networks without a saved choice; it does not restrict +which built-in protocols can be enabled later. + ## Platforms | Platform | Build | CLI `tsng` | Tray | Exit node: offer / use | Packages | diff --git a/crates/tsunagi-awg/tests/pairs.rs b/crates/tsunagi-awg/tests/pairs.rs index 5adc745..73573b1 100644 --- a/crates/tsunagi-awg/tests/pairs.rs +++ b/crates/tsunagi-awg/tests/pairs.rs @@ -25,6 +25,7 @@ use tsunagi_wg_quic::WireguardConfig; struct Node { agent: Agent, plugin: Arc, + quic: Option>, transport: Arc, tuns: MemoryTunFactory, loss: Arc, @@ -47,6 +48,16 @@ impl Node { interface: &str, fallback: bool, mtu: u32, + ) -> Self { + Self::open_selected(root, discovery, interface, fallback, mtu, None).await + } + async fn open_selected( + root: &Path, + discovery: &SharedMemoryDiscovery, + interface: &str, + fallback: bool, + mtu: u32, + default_protocols: Option>, ) -> Self { let identity = tsunagi::storage::StateStore::open(tsunagi::StoragePaths::under(root).state_db()) @@ -70,9 +81,9 @@ impl Node { let tuns = MemoryTunFactory::new(); let mut config = config_with(root, discovery) .with_interface(Arc::new(tuns.clone()), interface, mtu) - .with_overlay_ipv4_range(Some(tsunagi::state::DEFAULT_IPV4_RANGE)) .with_plugin(plugin.clone() as Arc) .with_custom_transport(AWG_PROTOCOL, faults); + let mut quic_plugin = None; if fallback { let quic = tsunagi_wg_quic::WireguardPlugin::open( WireguardConfig::new(root.join("wg-quic")) @@ -81,8 +92,10 @@ impl Node { ) .await .unwrap(); - config = config.with_plugin(quic); + config = config.with_plugin(quic.clone()); + quic_plugin = Some(quic); } + config.default_protocols = default_protocols; let agent = Agent::spawn(config).await.unwrap(); let inbound_agent = agent.clone(); let inbound_transport = transport.clone(); @@ -96,6 +109,7 @@ impl Node { Self { agent, plugin, + quic: quic_plugin, transport, tuns, loss, @@ -163,6 +177,235 @@ async fn exchange(a: &Node, b: &Node, id: NetworkId) { ); } } + +#[tokio::test] +async fn a_control_only_agent_enables_real_data_without_restart_and_cancels_stale_attempts() { + let discovery = SharedMemoryDiscovery::new(); + let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()]; + let a = Node::open_selected( + roots[0].path(), + &discovery, + "initial-none-a", + true, + 1280, + Some(vec![]), + ) + .await; + let b = Node::open_selected( + roots[1].path(), + &discovery, + "initial-none-b", + true, + 1280, + Some(vec![]), + ) + .await; + let (name, secret) = network("enable-after-control-only"); + let id = a.agent.join_network(&name, &secret).await.unwrap(); + b.agent.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&a.agent, id, 1).await; + wait_for_peers(&b.agent, id, 1).await; + assert!(a.agent.overlay().is_none()); + assert!(b.agent.overlay().is_none()); + for node in [&a, &b] { + assert!( + node.agent + .network_status(id) + .await + .unwrap() + .protocols + .is_empty() + ); + node.agent + .set_protocols(id, vec!["awg".into()]) + .await + .unwrap(); + assert!(node.agent.overlay().is_some()); + } + a.ready(id, 1).await; + b.ready(id, 1).await; + exchange(&a, &b, id).await; + let before = a.agent.overlay().unwrap().interface; + // Changes race unfinished opens and prepared plugin callbacks; the last + // selection must win even if an earlier open has completed in its queue. + for selection in [ + vec!["wg-quic".into()], + vec![], + vec!["awg".into()], + vec![], + vec!["awg".into()], + ] { + a.agent.set_protocols(id, selection).await.unwrap(); + } + a.ready(id, 1).await; + b.ready(id, 1).await; + exchange(&a, &b, id).await; + assert_eq!(a.agent.overlay().unwrap().interface, before); + for node in [&a, &b] { + let status = node.agent.network_status(id).await.unwrap(); + assert_eq!(status.protocols, ["awg"]); + assert_eq!(status.metrics.sessions_established, 1); + assert_eq!(status.metrics.disconnects, 0); + } + a.agent.shutdown().await; + b.agent.shutdown().await; +} + +#[tokio::test] +async fn runtime_protocol_selection_replaces_real_tunnels_and_survives_restart() { + let discovery = SharedMemoryDiscovery::new(); + let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()]; + let a = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await; + let b = Node::open_with_fallback(roots[1].path(), &discovery, "runtime-b", true).await; + let (name, secret) = network("runtime-protocols"); + let id = a.agent.join_network(&name, &secret).await.unwrap(); + b.agent.join_network(&name, &secret).await.unwrap(); + a.ready(id, 1).await; + b.ready(id, 1).await; + exchange(&a, &b, id).await; + + let (other_name, other_secret, other_id) = loop { + let (name, secret) = network("unchanged-runtime-network"); + let other_id = tsunagi::identity::NetworkKeys::derive(&name, &secret).network_id(); + if tsunagi::state::derived_ipv4_range(id) != tsunagi::state::derived_ipv4_range(other_id) { + break (name, secret, other_id); + } + }; + a.agent + .join_network(&other_name, &other_secret) + .await + .unwrap(); + b.agent + .join_network(&other_name, &other_secret) + .await + .unwrap(); + a.ready(other_id, 1).await; + b.ready(other_id, 1).await; + let other_before = a + .agent + .network_status(other_id) + .await + .unwrap() + .metrics + .data_links_established; + + for node in [&a, &b] { + let before = node.agent.network_status(id).await.unwrap().protocols; + for invalid in [vec!["unknown".into()], vec!["awg".into(), "awg".into()]] { + assert!(node.agent.set_protocols(id, invalid).await.is_err()); + assert_eq!( + node.agent.network_status(id).await.unwrap().protocols, + before + ); + } + } + // Exercise updates initiated by either endpoint-id role, including an + // empty set. Membership and the control session stay alive throughout. + for node in [&a, &b] { + node.agent.set_protocols(id, Vec::new()).await.unwrap(); + wait_until("peer accepts control-only selection", || async { + let other = if node.agent.endpoint_id() == a.agent.endpoint_id() { + &b + } else { + &a + }; + let status = other.agent.network_status(id).await.ok()?; + (status.peers.len() == 1 + && status.peers[0].protocols.is_empty() + && matches!(status.peers[0].data_path, tsunagi::agent::DataPath::None)) + .then_some(()) + }) + .await; + assert_eq!(node.agent.network_status(id).await.unwrap().peers.len(), 1); + node.agent + .set_protocols(id, vec!["wg-quic".into(), "awg".into()]) + .await + .unwrap(); + } + a.agent + .set_protocols(id, vec!["wg-quic".into()]) + .await + .unwrap(); + wait_until("real QUIC replaces AWG", || async { + let a_status = a.agent.network_status(id).await.ok()?; + let b_status = b.agent.network_status(id).await.ok()?; + (a_status.peers[0].protocols == ["wg-quic"] + && b_status.peers[0].protocols == ["wg-quic"] + && !matches!(a_status.peers[0].data_path, tsunagi::agent::DataPath::None) + && !matches!(b_status.peers[0].data_path, tsunagi::agent::DataPath::None) + && a.transport.overview(id).iter().all(|p| p.closed) + && b.transport.overview(id).iter().all(|p| p.closed)) + .then_some(()) + }) + .await; + // Get addresses from the system allocation: the disabled AWG plugin no + // longer has a live view of this network. + for (source, dest) in [(&a, &b), (&b, &a)] { + wait_until("QUIC end-to-end handshake", || async { + source + .quic + .as_ref()? + .overview(id)? + .peers + .iter() + .any(|p| p.tunnel.as_ref().is_some_and(|t| t.health.is_up())) + .then_some(()) + }) + .await; + let addr = |n: &Node| { + n.quic + .as_ref() + .unwrap() + .overview(id) + .unwrap() + .overlay_address_v4 + .unwrap() + }; + let bytes = packet(addr(source), addr(dest), 1280); + source.tun().push_from_os(bytes.clone()); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), dest.tun().pop_to_os()) + .await + .unwrap() + .unwrap(), + bytes + ); + } + assert_eq!( + a.agent + .network_status(other_id) + .await + .unwrap() + .metrics + .data_links_established, + other_before + ); + exchange(&a, &b, other_id).await; + a.agent.shutdown().await; + let restarted = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await; + wait_for_peers(&restarted.agent, id, 1).await; + assert_eq!( + restarted.agent.network_status(id).await.unwrap().protocols, + ["wg-quic"] + ); + assert_eq!( + b.agent.network_status(other_id).await.unwrap().protocols, + ["awg", "wg-quic"] + ); + restarted.ready(other_id, 1).await; + b.ready(other_id, 1).await; + exchange(&restarted, &b, other_id).await; + restarted + .agent + .set_protocols(id, vec!["awg".into()]) + .await + .unwrap(); + restarted.ready(id, 1).await; + b.ready(id, 1).await; + exchange(&restarted, &b, id).await; + restarted.agent.shutdown().await; + b.agent.shutdown().await; +} #[tokio::test] async fn real_pairs_carry_full_size_packets_and_restart_with_fresh_profiles() { tsunagi::testing::init_tracing(); diff --git a/crates/tsunagi-core/src/agent/mod.rs b/crates/tsunagi-core/src/agent/mod.rs index d70526f..1a7e060 100644 --- a/crates/tsunagi-core/src/agent/mod.rs +++ b/crates/tsunagi-core/src/agent/mod.rs @@ -58,6 +58,8 @@ use shutdown::Shutdown; /// Summary of a configured network, whether or not it is running. #[derive(Debug, Clone)] pub struct ConfiguredNetwork { + /// Enabled local data-plane protocols, in preference order. + pub protocols: Vec, /// Saved local broadcast participation. pub broadcast: bool, /// Public network identifier. @@ -104,6 +106,37 @@ struct Inner { } impl Inner { + /// Called under the lifecycle lock after startup: an initially + /// control-only agent creates its one interface on first enablement. + async fn ensure_interface(self: &Arc) { + if self.interface.get().is_some() { + return; + } + if let Some(factory) = self.config.tun_factory.clone() { + let carrier = Arc::new(Carrier(Arc::downgrade(self))) as Arc; + match crate::overlay::Interface::start( + factory, + self.config.interface_name.clone(), + self.config.interface_mtu, + Arc::clone(&self.routes), + carrier, + ) + .await + { + Ok(interface) => { + let _ = self.interface.set(Arc::new(interface)); + } + Err(err) => { + let _ = self.events.send(Event::PluginError { + network: NetworkId::from_bytes([0u8; 32]), + protocol: "overlay".into(), + reason: err.to_string(), + }); + } + } + } + } + /// The name this agent currently answers to. fn read_hostname(&self) -> String { match self.hostname.read() { @@ -265,12 +298,90 @@ impl TransportContext for TransportCtx { } impl Agent { + fn effective_protocols(&self, saved: Option<&Vec>) -> Vec { + saved + .or(self.inner.config.default_protocols.as_ref()) + .cloned() + .unwrap_or_else(|| { + self.inner + .config + .plugins + .iter() + .map(|p| p.protocol_id().to_owned()) + .collect() + }) + } + + fn selected_plugins(&self, protocols: &[String]) -> Vec { + protocols + .iter() + .filter_map(|name| { + self.inner + .config + .plugins + .iter() + .find(|p| p.protocol_id() == name) + .cloned() + }) + .collect() + } + + /// Saves and immediately applies a network's ordered local protocol set. + /// An empty set disables data traffic while retaining control sessions. + pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec) -> Result<()> { + crate::dataplane::validate_protocols(&protocols, self.inner.limits.max_capabilities)?; + for name in &protocols { + if !self + .inner + .config + .plugins + .iter() + .any(|p| p.protocol_id() == name) + { + return Err(Error::InvalidProtocols(format!( + "this agent has no `{name}` plugin" + ))); + } + } + let _lifecycle = self.inner.lifecycle.lock().await; + self.inner + .storage + .set_protocols(network_id, protocols.clone()) + .await?; + if self.is_active(network_id).await { + let (reply, receive) = oneshot::channel(); + if !protocols.is_empty() { + self.inner.ensure_interface().await; + } + self.command( + network_id, + NetCommand::SetProtocols { + protocols, + interface: self.inner.interface.get().cloned(), + reply, + }, + ) + .await?; + receive.await.map_err(|_| Error::Stopped)?; + } + Ok(()) + } /// Starts an agent. /// /// Opens the state store (taking its ownership lock), restores the /// persistent device identity, binds the iroh endpoint and activates every /// configured network whose auto-start flag is set. pub async fn spawn(config: AgentConfig) -> Result { + if let Some(protocols) = &config.default_protocols { + crate::dataplane::validate_protocols(protocols, config.limits.max_capabilities)?; + for name in protocols { + if !config.plugins.iter().any(|p| p.protocol_id() == name) { + return Err(Error::InvalidProtocols(format!( + "this agent has no `{name}` plugin" + ))); + } + } + } let storage = Storage::open(&config.paths)?; let identity = storage.device_identity().await?; let adapter = EndpointAdapter::bind(&config, &identity).await?; @@ -303,31 +414,13 @@ impl Agent { config, }); - // One interface for the agent, if it was given a way to make one. - // Failing to is reported and not fatal: the control plane works, and - // so do the protocols, they just have nowhere to put packets. - if let Some(factory) = inner.config.tun_factory.clone() { - let carrier = Arc::new(Carrier(Arc::downgrade(&inner))) as Arc; - match crate::overlay::Interface::start( - factory, - inner.config.interface_name.clone(), - inner.config.interface_mtu, - Arc::clone(&inner.routes), - carrier, - ) - .await - { - Ok(interface) => { - let _ = inner.interface.set(Arc::new(interface)); - } - Err(err) => { - let _ = inner.events.send(Event::PluginError { - network: NetworkId::from_bytes([0u8; 32]), - protocol: "overlay".into(), - reason: err.to_string(), - }); - } - } + if inner + .config + .default_protocols + .as_ref() + .is_none_or(|names| !names.is_empty()) + { + inner.ensure_interface().await; } // The data plane rides on iroh too, which is where it gets hole @@ -364,8 +457,9 @@ impl Agent { if !inner.config.plugins.is_empty() { let (plugin_tx, plugin_rx) = mpsc::channel(64); let sink = inner - .interface - .get() + .config + .tun_factory + .as_ref() .map(|_| Arc::new(Sink(Arc::downgrade(&inner))) as Arc); let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink); for plugin in &inner.config.plugins { @@ -824,6 +918,12 @@ impl Agent { }; let broadcast = stored.as_ref().is_none_or(|stored| stored.broadcast); + let protocols = + self.effective_protocols(stored.as_ref().and_then(|s| s.protocols.as_ref())); + let plugins = self.selected_plugins(&protocols); + if !plugins.is_empty() { + self.inner.ensure_interface().await; + } let exit_offer = stored.as_ref().is_some_and(|stored| stored.exit_node); let exit_via = stored .as_ref() @@ -842,7 +942,8 @@ impl Agent { discovery, discovery_interval: self.inner.config.discovery_interval, discovery_policy: self.inner.config.discovery_policy.clone(), - plugins: self.inner.config.plugins.clone(), + available_plugins: self.inner.config.plugins.clone(), + plugins: plugins.clone(), routes: Arc::clone(&self.inner.routes), interface: self.inner.interface.get().cloned(), hostname: self.inner.read_hostname(), @@ -856,7 +957,7 @@ impl Agent { networks.insert(network_id, handle); drop(networks); - for plugin in &self.inner.config.plugins { + for plugin in &plugins { plugin.on_network_activated(network_id); } @@ -997,6 +1098,7 @@ impl Agent { .await? .into_iter() .map(|stored| ConfiguredNetwork { + protocols: self.effective_protocols(stored.protocols.as_ref()), broadcast: stored.broadcast, network_id: stored.network_id, name: stored.name, @@ -1075,6 +1177,7 @@ impl Agent { let range = self.desired_range(stored.network_id).await?; let range_conflicts = self.range_conflicts(stored.network_id, range).await?; networks.push(NetworkStatus { + protocols: self.effective_protocols(stored.protocols.as_ref()), auto_start: stored.auto_start, broadcast: stored.broadcast, exit_offer: stored.exit_node, diff --git a/crates/tsunagi-core/src/agent/network.rs b/crates/tsunagi-core/src/agent/network.rs index 9e0f791..de7b04c 100644 --- a/crates/tsunagi-core/src/agent/network.rs +++ b/crates/tsunagi-core/src/agent/network.rs @@ -72,6 +72,11 @@ pub(crate) enum NetCommand { enabled: bool, reply: oneshot::Sender<()>, }, + SetProtocols { + interface: Option>, + protocols: Vec, + reply: oneshot::Sender<()>, + }, /// Offer this agent as an exit node in this network, or stop. SetExitOffer { enabled: bool, @@ -122,6 +127,7 @@ impl std::fmt::Debug for NetCommand { NetCommand::Status { .. } => f.write_str("Status"), NetCommand::Recheck => f.write_str("Recheck"), NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"), + NetCommand::SetProtocols { protocols, .. } => write!(f, "SetProtocols({protocols:?})"), NetCommand::SetExitOffer { enabled, .. } => write!(f, "SetExitOffer({enabled})"), NetCommand::SetExitVia { via, .. } => write!(f, "SetExitVia({})", via.is_some()), NetCommand::Reannounce => f.write_str("Reannounce"), @@ -171,6 +177,7 @@ pub(crate) struct RuntimeParams { pub(crate) discovery_interval: Duration, pub(crate) discovery_policy: DiscoveryPolicy, pub(crate) plugins: Vec, + pub(crate) available_plugins: Vec, /// Who holds which overlay address, shared with every other network. pub(crate) routes: Arc, /// The one interface, when the agent has one. @@ -192,6 +199,7 @@ pub(crate) struct RuntimeParams { /// Outcome of one attempt to open a data plane link. struct LinkOutcome { + generation: u64, peer: EndpointId, protocol: String, result: Result, @@ -332,6 +340,9 @@ struct Runtime { announced_reach: HashSet, /// Links currently being opened, so we do not start two. opening: HashSet<(EndpointId, String)>, + link_tasks: HashMap<(EndpointId, String), tokio::task::AbortHandle>, + link_attempts: HashMap<(EndpointId, String), u64>, + next_link_attempt: u64, link_results_tx: mpsc::Sender, link_results_rx: mpsc::Receiver, /// Protocol dial failure count and backoff cooldown per peer and protocol. @@ -425,6 +436,9 @@ impl Runtime { local_capabilities: Vec::new(), announced_reach: HashSet::new(), opening: HashSet::new(), + link_tasks: HashMap::new(), + link_attempts: HashMap::new(), + next_link_attempt: 0, link_results_tx, link_results_rx, protocol_dial_states: HashMap::new(), @@ -511,6 +525,10 @@ impl Runtime { link.close(); } self.links.clear(); + for (_, task) in self.link_tasks.drain() { + task.abort(); + } + self.opening.clear(); self.pending_health.clear(); self.last_failed_protocol.clear(); self.protocol_dial_states.clear(); @@ -553,6 +571,60 @@ impl Runtime { NetCommand::Status { reply } => { let _ = reply.send(Box::new(self.status())); } + NetCommand::SetProtocols { + protocols, + interface, + reply, + } => { + self.params.interface = interface; + let current: Vec<_> = self + .params + .plugins + .iter() + .map(|p| p.protocol_id().to_owned()) + .collect(); + if current != protocols { + let peers: Vec<_> = self.sessions.keys().copied().collect(); + for peer in &peers { + self.reset_peer_data(*peer, "local protocol selection changed"); + } + for plugin in &self.params.plugins { + if !protocols.iter().any(|name| name == plugin.protocol_id()) { + plugin.on_network_deactivated(self.network_id); + } + } + self.params.plugins = protocols + .iter() + .filter_map(|name| { + self.params + .available_plugins + .iter() + .find(|p| p.protocol_id() == name) + .cloned() + }) + .collect(); + for plugin in &self.params.plugins { + if !current.iter().any(|name| name == plugin.protocol_id()) { + plugin.on_network_activated(self.network_id); + } + } + self.hub.set_priority(protocols); + self.publish_allocations().await; + for peer in peers { + let caps = self + .sessions + .get(&peer) + .map(|s| s.capabilities.clone()) + .unwrap_or_default(); + self.dispatch_capabilities(peer, &caps); + } + self.reannounce(); + self.ensure_links(); + self.update_paths(); + self.announce_reach(true); + } + let _ = reply.send(()); + } NetCommand::SetBroadcast { enabled, reply } => { self.params.broadcast = enabled; self.update_broadcast(); @@ -584,7 +656,22 @@ impl Runtime { let _ = reply.send(outcome); } NetCommand::Recheck => self.discovery_round().await, - NetCommand::Reannounce => self.reannounce(), + NetCommand::Reannounce => { + // A newly enabled plugin may prepare its key asynchronously. + // Replay allocations and peer identities once it is ready. + self.publish_allocations().await; + let peers: Vec<_> = self + .sessions + .iter() + .map(|(peer, s)| (*peer, s.capabilities.clone())) + .collect(); + for (peer, caps) in peers { + self.dispatch_capabilities(peer, &caps); + } + self.reannounce(); + self.ensure_links(); + self.update_paths(); + } NetCommand::Release { reply } => { // A tombstone, so the address and the name this agent held // are freed for somebody else instead of staying reserved @@ -1578,13 +1665,16 @@ impl Runtime { if self.links.contains_key(&key) || self.opening.contains(&key) { continue; } - self.opening.insert(key); + self.opening.insert(key.clone()); 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 { + self.next_link_attempt = self.next_link_attempt.wrapping_add(1); + let generation = self.next_link_attempt; + self.link_attempts.insert(key.clone(), generation); + let task = tokio::spawn(async move { let result = tokio::select! { biased; _ = shutdown.wait() => Err("network deactivated".to_string()), @@ -1594,12 +1684,14 @@ impl Runtime { }; let _ = results .send(LinkOutcome { + generation, peer, protocol, result, }) .await; }); + self.link_tasks.insert(key, task.abort_handle()); } } @@ -1834,7 +1926,15 @@ impl Runtime { fn handle_link_result(&mut self, outcome: LinkOutcome) { let key = (outcome.peer, outcome.protocol.clone()); + if self.link_attempts.get(&key) != Some(&outcome.generation) { + if let Ok(link) = outcome.result { + link.close(); + } + return; + } + self.link_attempts.remove(&key); self.opening.remove(&key); + self.link_tasks.remove(&key); match outcome.result { Ok(link) => { self.record_protocol_success(outcome.peer, &outcome.protocol); @@ -1903,12 +2003,17 @@ impl Runtime { link.close(); return; } - if !self - .params - .plugins - .iter() - .any(|plugin| plugin.protocol_id() == protocol) - { + if !self.params.plugins.iter().any(|plugin| { + plugin.protocol_id() == protocol + && self.sessions.get(&peer).is_some_and(|s| { + s.capabilities.iter().any(|c| { + c.enabled + && c.protocol == protocol + && c.version == plugin.protocol_version() + }) + }) + }) { + link.close(); return; } @@ -1966,14 +2071,25 @@ impl Runtime { /// Drops every link to a peer. fn drop_links_for(&mut self, peer: EndpointId) { - // The peer itself is going, not just a path to it, so the - // protocol's link goes with it. + self.reset_peer_data(peer, "peer session ended"); self.hub.remove_peer(peer); self.reachable.remove(&peer); - self.protocol_dial_states.retain(|(p, _), _| *p != peer); - self.last_failed_protocol.remove(&peer); - self.pending_health.retain(|(p, _), _| *p != peer); - let keys: Vec<(EndpointId, String)> = self + } + + /// Retire only the data plane; authenticated membership and topology remain. + fn reset_peer_data(&mut self, peer: EndpointId, reason: &str) { + self.link_attempts.retain(|(id, _), _| *id != peer); + self.opening.retain(|(id, _)| *id != peer); + self.link_tasks.retain(|(id, _), task| { + if *id == peer { + task.abort(); + false + } else { + true + } + }); + self.hub.clear_peer_links(peer); + let keys: Vec<_> = self .links .keys() .filter(|(id, _)| *id == peer) @@ -1987,9 +2103,15 @@ impl Runtime { network: self.network_id, peer, protocol: key.1, - reason: "peer session ended".into(), + reason: reason.into(), }); } + self.protocol_dial_states.retain(|(id, _), _| *id != peer); + self.last_failed_protocol.remove(&peer); + self.pending_health.retain(|(id, _), _| *id != peer); + for plugin in &self.params.plugins { + plugin.on_peer_gone(self.network_id, peer); + } } // --------------------------------------------------------------- sessions @@ -2182,9 +2304,6 @@ impl Runtime { self.watch_exit_via(); self.update_paths(); self.announce_reach(false); - for plugin in &self.params.plugins { - plugin.on_peer_gone(self.network_id, peer); - } // Retry promptly, then back off if it keeps failing. let policy = &self.params.reconnect; let state = self @@ -2277,6 +2396,18 @@ impl Runtime { } ControlMessage::Announce(announcement) => { let capabilities = announcement.capabilities.clone(); + let changed = self.sessions.get(&peer).is_some_and(|session| { + let selection = |caps: &[PluginCapability]| { + caps.iter() + .filter(|c| c.enabled) + .map(|c| (c.protocol.clone(), c.version)) + .collect::>() + }; + selection(&session.capabilities) != selection(&capabilities) + }); + if changed { + self.reset_peer_data(peer, "peer protocol selection changed"); + } if let Some(session) = self.sessions.get_mut(&peer) { session.hostname = Some(announcement.hostname.clone()); session.broadcast = announcement.broadcast; @@ -2289,6 +2420,7 @@ impl Runtime { self.dispatch_capabilities(peer, &capabilities); self.ensure_links(); self.update_paths(); + self.announce_reach(false); } ControlMessage::Ping { seq, payload } => { let pong = ControlMessage::Pong { @@ -2522,6 +2654,12 @@ impl Runtime { members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id); NetworkStatus { + protocols: self + .params + .plugins + .iter() + .map(|p| p.protocol_id().to_owned()) + .collect(), auto_start: true, broadcast: self.params.broadcast, exit_offer: self.params.exit_offer, diff --git a/crates/tsunagi-core/src/agent/status.rs b/crates/tsunagi-core/src/agent/status.rs index 9e39688..f99263b 100644 --- a/crates/tsunagi-core/src/agent/status.rs +++ b/crates/tsunagi-core/src/agent/status.rs @@ -223,6 +223,8 @@ pub struct RangeConflict { /// Status of one network. #[derive(Debug, Clone)] pub struct NetworkStatus { + /// Enabled local data-plane protocols, in preference order. + pub protocols: Vec, /// Whether this network is requested to start or resume automatically. pub auto_start: bool, /// Local broadcast participation in this network. diff --git a/crates/tsunagi-core/src/config.rs b/crates/tsunagi-core/src/config.rs index 9fe9df4..3b56f78 100644 --- a/crates/tsunagi-core/src/config.rs +++ b/crates/tsunagi-core/src/config.rs @@ -328,6 +328,9 @@ pub struct AgentConfig { pub reconnect: ReconnectPolicy, /// IP plugins whose capabilities are announced and dispatched. pub plugins: Vec, + /// Initial ordered selection for networks without a saved override. + /// `None` enables all registered plugins; an empty list is control-only. + pub default_protocols: Option>, /// Custom packet transports registered per protocol. pub custom_transports: std::collections::HashMap>, @@ -376,6 +379,7 @@ impl AgentConfig { limits: Limits::default(), reconnect: ReconnectPolicy::default(), plugins: Vec::new(), + default_protocols: None, custom_transports: std::collections::HashMap::new(), tun_factory: None, interface_name: "tsun0".to_string(), diff --git a/crates/tsunagi-core/src/dataplane/mod.rs b/crates/tsunagi-core/src/dataplane/mod.rs index 7ed05ce..79daabd 100644 --- a/crates/tsunagi-core/src/dataplane/mod.rs +++ b/crates/tsunagi-core/src/dataplane/mod.rs @@ -40,6 +40,33 @@ pub use tunnel::{GenericTunnelPlugin, TunnelCodec}; /// Maximum length of a plugin protocol identifier. pub const MAX_PROTOCOL_ID_LEN: usize = 32; +/// Validates bounded local protocol preferences before storing or applying them. +pub(crate) fn validate_protocols(protocols: &[String], max: usize) -> crate::error::Result<()> { + if protocols.len() > max { + return Err(crate::Error::InvalidProtocols(format!( + "at most {max} protocols are allowed" + ))); + } + for (index, name) in protocols.iter().enumerate() { + if name.is_empty() + || name.len() > MAX_PROTOCOL_ID_LEN + || !name + .bytes() + .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_') + { + return Err(crate::Error::InvalidProtocols( + "names must be 1–32 ASCII letters, digits, hyphens or underscores".into(), + )); + } + if protocols[..index].contains(name) { + return Err(crate::Error::InvalidProtocols(format!( + "`{name}` is listed more than once" + ))); + } + } + Ok(()) +} + /// An announcement of one IP plugin's capability. /// /// `data` is opaque to the core. Nothing in it may be interpreted as a shell diff --git a/crates/tsunagi-core/src/dataplane/relay.rs b/crates/tsunagi-core/src/dataplane/relay.rs index ff119e3..4e337af 100644 --- a/crates/tsunagi-core/src/dataplane/relay.rs +++ b/crates/tsunagi-core/src/dataplane/relay.rs @@ -492,11 +492,12 @@ impl RelayHub { } } - /// Revokes all links and topology involving a departed authenticated peer. - pub fn remove_peer(&self, peer: EndpointId) { + /// Retires data channels without removing an authenticated peer's topology. + pub fn clear_peer_links(&self, peer: EndpointId) { let mut state = self.state(); if let Some(links) = state.raw.remove(peer.as_bytes()) { for raw in links { + raw.link.close(); raw.reader.abort(); } } @@ -511,6 +512,13 @@ impl RelayHub { link.close(); } } + state.publish(); + } + + /// Revokes all links and topology involving a departed authenticated peer. + pub fn remove_peer(&self, peer: EndpointId) { + self.clear_peer_links(peer); + let mut state = self.state(); state.members.remove(peer.as_bytes()); state.graph.remove(peer.as_bytes()); for neighbors in state.graph.values_mut() { diff --git a/crates/tsunagi-core/src/error.rs b/crates/tsunagi-core/src/error.rs index 66f79df..b52b34a 100644 --- a/crates/tsunagi-core/src/error.rs +++ b/crates/tsunagi-core/src/error.rs @@ -15,6 +15,9 @@ pub type Result = std::result::Result; #[derive(Debug, thiserror::Error)] #[non_exhaustive] pub enum Error { + /// An ordered local data-plane protocol selection is invalid. + #[error("invalid protocol selection: {0}")] + InvalidProtocols(String), /// The supplied network name does not satisfy the documented rules. #[error("invalid network name: {0}")] InvalidNetworkName(&'static str), diff --git a/crates/tsunagi-core/src/ipc/mod.rs b/crates/tsunagi-core/src/ipc/mod.rs index 8b2b574..879b653 100644 --- a/crates/tsunagi-core/src/ipc/mod.rs +++ b/crates/tsunagi-core/src/ipc/mod.rs @@ -171,6 +171,13 @@ pub fn default_control_socket_path() -> PathBuf { #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub enum Request { + /// Replace one network's ordered local data-plane selection. + SetProtocols { + /// Public network identifier. + network_id: String, + /// Protocol names, best first; empty selects control only. + protocols: Vec, + }, /// Report what the agent is doing. Status, /// Set local broadcast participation in one configured network. @@ -255,6 +262,13 @@ pub enum Request { impl std::fmt::Debug for Request { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { + Request::SetProtocols { + network_id, + protocols, + } => write!( + f, + "SetProtocols {{ {network_id}, protocols: {protocols:?} }}" + ), Request::Status => f.write_str("Status"), Request::SetBroadcast { network_id, @@ -286,6 +300,8 @@ impl std::fmt::Debug for Request { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub enum Response { + /// Accepted ordered protocol selection. + Protocols(Vec), /// A status report. Status(Box), /// Accepted local broadcast participation. @@ -377,6 +393,8 @@ pub struct LeftReport { /// Everything the agent is doing, in one snapshot. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct StatusReport { + /// Data-plane protocols this agent can enable at runtime. + pub available_protocols: Vec, /// This device's persistent endpoint id. pub endpoint_id: String, /// Hostname announced to peers. @@ -458,9 +476,20 @@ pub struct RangeConflictReport { pub range: String, } +/// A data-plane protocol available in the running agent. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProtocolReport { + /// Stable protocol name used by the CLI and network negotiation. + pub name: String, + /// A short description of encryption and transport. + pub summary: String, +} + /// One network. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct NetworkReport { + /// Enabled local data-plane protocols, in preference order. + pub protocols: Vec, /// Whether the network starts or resumes automatically when its range is free. #[serde(default)] pub auto_start: bool, @@ -720,6 +749,14 @@ impl OverlayPeerReport { /// running and what they can report or change. That is what keeps this module /// free of any knowledge of them. pub trait ReportSource: Send + Sync + 'static { + /// Saves and applies a local ordered protocol set for one network. + fn set_protocols( + &self, + _network_id: String, + _protocols: Vec, + ) -> BoxFuture<'_, std::result::Result, String>> { + Box::pin(async { Err("this agent cannot change protocols".into()) }) + } /// Produces a fresh report. fn report(&self) -> BoxFuture<'_, StatusReport>; @@ -849,7 +886,7 @@ pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5); /// /// Bump it whenever [`Request`], [`Response`] or anything they contain /// changes shape. -pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 19]); +pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 20]); /// Reads one request off an accepted stream, answers it, writes the response. /// @@ -873,6 +910,13 @@ where } }; let response = match request { + Request::SetProtocols { + network_id, + protocols, + } => match source.set_protocols(network_id, protocols).await { + Ok(protocols) => Response::Protocols(protocols), + Err(reason) => Response::Error(reason), + }, Request::Status => Response::Status(Box::new(source.report().await)), Request::SetBroadcast { network_id, @@ -1028,7 +1072,24 @@ pub async fn join_network_with_broadcast( } } -/// Changes one network's broadcast participation while the agent runs. +/// Replaces the enabled data-plane protocols of one configured network. +pub async fn set_protocols( + socket: &Path, + network_id: &str, + protocols: Vec, +) -> Result> { + let request = Request::SetProtocols { + network_id: network_id.to_owned(), + protocols, + }; + match exchange(socket, &request, EXCHANGE_TIMEOUT).await? { + Response::Protocols(protocols) => Ok(protocols), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected response: {other:?}"))), + } +} + +/// Changes local broadcast participation in a network. pub async fn set_broadcast( path: impl AsRef, network_id: &str, diff --git a/crates/tsunagi-core/src/storage/mod.rs b/crates/tsunagi-core/src/storage/mod.rs index 8d402c9..a7b4bf4 100644 --- a/crates/tsunagi-core/src/storage/mod.rs +++ b/crates/tsunagi-core/src/storage/mod.rs @@ -254,6 +254,12 @@ impl Storage { .await } + /// Persists the ordered protocol selection off the async executor. + pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec) -> Result<()> { + self.with_state(move |state| state.set_protocols(network_id, &protocols)) + .await + } + /// Persists whether this agent offers itself as an exit node in a network. pub async fn set_exit_node(&self, network_id: NetworkId, enabled: bool) -> Result<()> { self.with_state(move |state| state.set_exit_node(network_id, enabled)) diff --git a/crates/tsunagi-core/src/storage/state.rs b/crates/tsunagi-core/src/storage/state.rs index 8f0104a..fac7b5d 100644 --- a/crates/tsunagi-core/src/storage/state.rs +++ b/crates/tsunagi-core/src/storage/state.rs @@ -21,7 +21,7 @@ use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret}; use crate::state::{RecordBody, SignedRecord}; /// Schema version written by this build. -pub const SCHEMA_VERSION: i64 = 7; +pub const SCHEMA_VERSION: i64 = 8; /// Key of the stored hostname setting. const SETTING_HOSTNAME: &str = "hostname"; @@ -32,6 +32,8 @@ const SETTING_HOSTNAME: &str = "hostname"; /// and zeroizes on drop. #[derive(Debug, Clone)] pub struct StoredNetwork { + /// Ordered local protocol override; absent inherits the agent's defaults. + pub protocols: Option>, /// Derived public network identifier. pub network_id: NetworkId, /// Network name. @@ -238,6 +240,16 @@ impl StateStore { ) .map_err(|err| self.corrupt(format!("cannot migrate schema to 7: {err}")))?; } + if found < 8 { + self.conn + .execute_batch( + "BEGIN; + ALTER TABLE networks ADD COLUMN protocols TEXT; + PRAGMA user_version = 8; + COMMIT;", + ) + .map_err(|err| self.corrupt(format!("cannot migrate schema to 8: {err}")))?; + } Ok(()) } @@ -476,6 +488,25 @@ impl StateStore { Ok(()) } + /// Saves a local ordered protocol selection without changing membership. + pub fn set_protocols(&self, network_id: NetworkId, protocols: &[String]) -> Result<()> { + crate::dataplane::validate_protocols( + protocols, + crate::config::Limits::default().max_capabilities, + )?; + let changed = self + .conn + .execute( + "UPDATE networks SET protocols = ?2 WHERE network_id = ?1", + params![network_id.as_bytes().as_slice(), protocols.join(",")], + ) + .map_err(|err| Error::Storage(format!("cannot update protocols: {err}")))?; + if changed == 0 { + return Err(Error::NetworkUnknown(network_id)); + } + Ok(()) + } + /// Removes a network configuration entirely. pub fn remove_network(&self, network_id: NetworkId) -> Result<()> { self.conn @@ -536,7 +567,7 @@ impl StateStore { pub fn list_networks(&self) -> Result> { let mut stmt = self .conn - .prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via FROM networks ORDER BY name") + .prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols FROM networks ORDER BY name") .map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?; let rows = stmt .query_map([], |row| { @@ -552,19 +583,37 @@ impl StateStore { row.get::<_, bool>(4)?, row.get::<_, bool>(5)?, row.get::<_, Option>>(6)?, + row.get::<_, Option>(7)?, )) }) .map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?; let mut out = Vec::new(); for row in rows { - let (id, name, secret, auto_start, broadcast, exit_node, exit_via) = + let (id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols) = row.map_err(|err| Error::Storage(format!("cannot read network row: {err}")))?; + let protocols = protocols + .map(|text| { + let max = crate::config::Limits::default().max_capabilities; + if text.len() > max * (crate::dataplane::MAX_PROTOCOL_ID_LEN + 1) { + return Err(self.corrupt("stored protocol selection is too long")); + } + let names = if text.is_empty() { + Vec::new() + } else { + text.split(',').map(str::to_owned).collect() + }; + crate::dataplane::validate_protocols(&names, max) + .map_err(|err| self.corrupt(err))?; + Ok(names) + }) + .transpose()?; let id: [u8; 32] = id .as_slice() .try_into() .map_err(|_| self.corrupt("stored network id is not 32 bytes"))?; out.push(StoredNetwork { + protocols, network_id: NetworkId::from_bytes(id), name: NetworkName::new(name)?, secret: NetworkSecret::from_bytes(secret)?, @@ -957,6 +1006,68 @@ mod broadcast_storage_tests { assert!(store.peer_hostnames(id).unwrap().is_empty()); } + #[test] + fn v7_migration_keeps_signed_state_and_protocol_overrides_are_local_and_durable() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.sqlite"); + let store = StateStore::open(&path).unwrap(); + let identity = store.load_or_create_device_identity().unwrap(); + let name = NetworkName::new("protocol-migration").unwrap(); + let secret = NetworkSecret::from_bytes([4; 32]).unwrap(); + let id = NetworkKeys::derive(&name, &secret).network_id(); + store.upsert_network(id, &name, &secret, true).unwrap(); + store.set_broadcast(id, false).unwrap(); + let record = SignedRecord::sign(&identity.signing_key(), id, 17, RecordBody::Release); + store.publish_own_record(&record).unwrap(); + store + .conn + .execute_batch("ALTER TABLE networks DROP COLUMN protocols; PRAGMA user_version=7;") + .unwrap(); + drop(store); + let store = StateStore::open(&path).unwrap(); + assert_eq!( + store + .load_or_create_device_identity() + .unwrap() + .endpoint_id(), + identity.endpoint_id() + ); + assert_eq!(store.signed_records(id).unwrap(), vec![record]); + let saved = store.list_networks().unwrap(); + assert!(saved[0].protocols.is_none()); + assert!(!saved[0].broadcast); + assert!(saved[0].auto_start); + assert_eq!(saved[0].secret, secret); + let chosen = vec!["awg".into(), "wg-quic".into()]; + store.set_protocols(id, &chosen).unwrap(); + assert!( + store + .set_protocols(id, &["awg".into(), "awg".into()]) + .is_err() + ); + assert!( + store + .set_protocols(NetworkId::from_bytes([1; 32]), &[]) + .is_err() + ); + drop(store); + let store = StateStore::open(&path).unwrap(); + assert_eq!(store.list_networks().unwrap()[0].protocols, Some(chosen)); + store.upsert_network(id, &name, &secret, true).unwrap(); + store.set_protocols(id, &[]).unwrap(); + drop(store); + let store = StateStore::open(&path).unwrap(); + assert_eq!(store.list_networks().unwrap()[0].protocols, Some(vec![])); + store + .conn + .execute("UPDATE networks SET protocols = 'awg,awg'", []) + .unwrap(); + assert!(matches!( + store.list_networks(), + Err(Error::StateCorrupted { .. }) + )); + } + #[test] fn v3_migration_preserves_identity_and_networks_and_defaults_broadcast_on() { let dir = tempfile::tempdir().unwrap(); @@ -977,6 +1088,7 @@ mod broadcast_storage_tests { "ALTER TABLE networks DROP COLUMN broadcast; \ ALTER TABLE networks DROP COLUMN exit_node; \ ALTER TABLE networks DROP COLUMN exit_via; \ + ALTER TABLE networks DROP COLUMN protocols; \ PRAGMA user_version=3;", ) .unwrap(); diff --git a/crates/tsunagi-tray/src/agent.rs b/crates/tsunagi-tray/src/agent.rs index 2e5dd6a..f99a13f 100644 --- a/crates/tsunagi-tray/src/agent.rs +++ b/crates/tsunagi-tray/src/agent.rs @@ -21,6 +21,11 @@ const POLL_INTERVAL: Duration = Duration::from_secs(2); /// A request from the UI to the agent. pub(crate) enum Command { + /// Replace the network's enabled protocols in preference order. + SetProtocols { + network_id: String, + protocols: Vec, + }, /// Stop serving a network, or start serving it again. SetActive { network_id: String, active: bool }, /// Turn local broadcast participation on or off for a network. @@ -106,7 +111,13 @@ impl AgentClient { /// Sends a command; a dropped worker simply means nothing happens. pub(crate) fn send(&self, command: Command) { - let _ = self.tx.send(command); + // Disable protocol editing until the refreshed snapshot arrives; + // otherwise a second checkbox click could replace the first edit + // using the same stale network selection. + lock(&self.shared).busy = true; + if self.tx.send(command).is_err() { + lock(&self.shared).busy = false; + } } /// A copy of the latest snapshot. @@ -142,6 +153,8 @@ async fn worker( let Some(command) = command else { return }; run(&resolve_socket(), command, &shared, &pending_copy, &ctx).await; refresh(&shared, &ctx).await; + lock(&shared).busy = false; + ctx.request_repaint(); } } } @@ -179,6 +192,22 @@ async fn run( ctx.request_repaint(); let outcome: Result = match command { + Command::SetProtocols { + network_id, + protocols, + } => ipc::set_protocols(socket, &network_id, protocols) + .await + .map(|names| { + format!( + "protocols: {}", + if names.is_empty() { + "none".into() + } else { + names.join(", ") + } + ) + }) + .map_err(|err| err.to_string()), Command::SetActive { network_id, active } => ipc::set_active(socket, &network_id, active) .await .map(|report| { @@ -260,7 +289,6 @@ async fn run( { let mut guard = lock(shared); - guard.busy = false; guard.last_action = if matches!(&outcome, Ok(message) if message.is_empty()) { None } else { diff --git a/crates/tsunagi-tray/src/ui.rs b/crates/tsunagi-tray/src/ui.rs index 525f516..e0f858a 100644 --- a/crates/tsunagi-tray/src/ui.rs +++ b/crates/tsunagi-tray/src/ui.rs @@ -123,7 +123,7 @@ pub(crate) fn draw( ui.weak("No networks yet — join one below."); } for network in &report.networks { - draw_network(ui, agent, state, traffic, network, &report.endpoint_id); + draw_network(ui, agent, state, traffic, network, report); } }); } @@ -202,7 +202,7 @@ fn draw_network( state: &mut UiState, traffic: &Traffic, network: &NetworkReport, - own_id: &str, + report: &StatusReport, ) { ui.add_space(6.0); egui::Frame::group(ui.style()).show(ui, |ui| { @@ -244,12 +244,14 @@ fn draw_network( ui.weak("peers"); ui.label(format!( "{} online · {} known", - devices::online_count(network, own_id), - devices::known_count(network, own_id) + devices::online_count(network, &report.endpoint_id), + devices::known_count(network, &report.endpoint_id) )); ui.end_row(); }); + draw_protocols(ui, agent, network, &report.available_protocols); + format::sparkline(ui, series, state.unit, 34.0); if !network.range_conflicts.is_empty() { @@ -345,6 +347,83 @@ fn draw_network( }); } +fn draw_protocols( + ui: &mut egui::Ui, + agent: &AgentClient, + network: &NetworkReport, + available: &[tsunagi::ipc::ProtocolReport], +) { + let summary = if network.protocols.is_empty() { + "none".into() + } else { + network.protocols.join(", ") + }; + egui::CollapsingHeader::new(format!("protocols {summary}")) + .id_salt(("protocols", &network.network_id)) + .show(ui, |ui| { + ui.weak("Preference order; changes apply immediately and are saved."); + let mut selected = network.protocols.clone(); + let mut changed = false; + ui.add_enabled_ui(!agent.snapshot().busy, |ui| { + // Enabled items follow the user's preference; unchecked items + // follow the agent's catalog. The worker refreshes after edits. + let ordered: Vec<_> = network + .protocols + .iter() + .filter_map(|name| available.iter().find(|p| &p.name == name)) + .chain( + available + .iter() + .filter(|p| !network.protocols.contains(&p.name)), + ) + .collect(); + for protocol in ordered { + ui.horizontal(|ui| { + let mut on = selected.contains(&protocol.name); + if ui.checkbox(&mut on, &protocol.name).changed() { + if on { + selected.push(protocol.name.clone()); + } else { + selected.retain(|p| p != &protocol.name); + } + changed = true; + } + if let Some(index) = selected.iter().position(|p| p == &protocol.name) { + if ui + .add_enabled(index > 0, egui::Button::new("↑")) + .on_hover_text("prefer earlier") + .clicked() + { + selected.swap(index, index - 1); + changed = true; + } + if ui + .add_enabled(index + 1 < selected.len(), egui::Button::new("↓")) + .on_hover_text("prefer later") + .clicked() + { + selected.swap(index, index + 1); + changed = true; + } + } + }); + ui.indent(("description", &network.network_id, &protocol.name), |ui| { + ui.weak(&protocol.summary); + }); + } + }); + if selected.is_empty() { + ui.weak("Control only; no IP traffic in this network."); + } + if changed { + agent.send(Command::SetProtocols { + network_id: network.network_id.clone(), + protocols: selected, + }); + } + }); +} + /// A prominent line about an exit node. One that carries a command copies it /// when clicked. fn draw_exit_warning(ui: &mut egui::Ui, warning: &exit::Warning) { diff --git a/crates/tsunagi/src/main.rs b/crates/tsunagi/src/main.rs index 953d917..b41609b 100644 --- a/crates/tsunagi/src/main.rs +++ b/crates/tsunagi/src/main.rs @@ -69,8 +69,8 @@ enum Command { Up(Box), /// Reports this device, what the agent is doing, and what this host can do. Status(StatusArgs), - /// Shows the protocols this build can carry packets with. - Protocols, + /// Lists protocols, or shows and changes one network's selection at runtime. + Protocols(ProtocolsArgs), /// Makes a network, or joins one — the command that does the work. /// /// The same as `tsng network join`, at the top level because it is @@ -107,6 +107,23 @@ struct DnsArgs { action: Option, } +#[derive(Debug, Args)] +#[command( + after_help = "Examples:\n tsng protocols\n tsng protocols home\n tsng protocols home awg,wg-quic\n tsng protocols home none\n\nThe selection is local to this device, saved per network and applied immediately.\nBoth endpoints need a common protocol at the same wire version." +)] +struct ProtocolsArgs { + #[command(flatten)] + paths: PathArgs, + /// Control socket to talk to; uses the running service by default. + #[arg(long, env = "TSUNAGI_CONTROL_SOCKET")] + control_socket: Option, + /// Network's exact name, id or unique id prefix. + network: Option, + /// Replace the selection, best first: awg,wg-quic; none disables data traffic. + #[arg(value_name = "LIST", requires = "network")] + protocols: Option, +} + #[derive(Debug, Subcommand)] enum DnsAction { /// Starts serving, now and after every restart. @@ -465,11 +482,13 @@ struct UpArgs { #[arg(long, help_heading = "System")] no_tun: bool, - /// Protocols to carry packets with, best first. + /// Initial protocol selection for networks without a saved choice, best first. /// /// A pair of peers uses one they both have at the same wire version. A /// peer with none in common keeps its control plane and gets no data - /// plane. `none` runs the control plane alone. + /// plane. `none` runs the control plane alone. `tsng protocols NETWORK LIST` + /// changes and saves each network's choice while this agent runs; any + /// built-in protocol can be enabled, even if absent from this initial list. #[arg( long = "protocol", value_name = "LIST", @@ -897,7 +916,7 @@ async fn run(command: Command) -> Result<(), Box> { Command::Id(args) => id(args).await, Command::Up(args) => up(*args).await, Command::Status(args) => status(args).await, - Command::Protocols => show_protocols(), + Command::Protocols(args) => protocols_command(args).await, Command::Join(args) => join_command(args).await, Command::Network(args) => network_command(args).await, Command::Dns(args) => dns_command(args).await, @@ -1454,6 +1473,7 @@ const PROTOCOLS: &[ProtocolSpec] = &[ ]; /// One `-o` setting, and the protocol it was aimed at. +#[derive(Clone)] struct Setting { /// `Some` when written as `protocol:key=value`. protocol: Option, @@ -1704,6 +1724,23 @@ impl tsunagi::ipc::ReportSource for AgentControl { }) } + fn set_protocols( + &self, + network_id: String, + protocols: Vec, + ) -> tsunagi::BoxFuture<'_, Result, String>> { + Box::pin(async move { + let id = network_id + .parse() + .map_err(|err| format!("invalid network id: {err}"))?; + self.agent + .set_protocols(id, protocols.clone()) + .await + .map_err(|err| err.to_string())?; + Ok(protocols) + }) + } + fn set_exit_offer( &self, network_id: String, @@ -1866,45 +1903,138 @@ fn resolve_network<'a>( networks: &'a [tsunagi::storage::StoredNetwork], wanted: &str, ) -> Result<&'a tsunagi::storage::StoredNetwork, String> { + let entries: Vec<_> = networks + .iter() + .map(|n| (n.network_id.to_string(), n.name.as_str().to_owned())) + .collect(); + resolve_network_index(&entries, wanted).map(|i| &networks[i]) +} + +/// Shared selector rules for stored networks and secret-free live reports. +fn resolve_network_index(networks: &[(String, String)], wanted: &str) -> Result { let wanted = wanted.trim_matches(|c: char| c.is_ascii_whitespace()); if wanted.is_empty() { - return Err("select a network by name or id; `tsng network` lists them".to_string()); + return Err("select a network by name or id; `tsng network` lists them".into()); } - if let Some(network) = networks - .iter() - .find(|network| network.network_id.to_string() == wanted) - { - return Ok(network); + if let Some(index) = networks.iter().position(|(id, _)| id == wanted) { + return Ok(index); } - // Only id prefixes lose a copied report ellipsis; names are used verbatim. let prefix = wanted.trim_end_matches('…'); - let matched: Vec<&tsunagi::storage::StoredNetwork> = networks + let matches: Vec<_> = networks .iter() - .filter(|network| { - network.name.as_str() == wanted - || (!prefix.is_empty() && network.network_id.to_string().starts_with(prefix)) - }) + .enumerate() + .filter(|(_, (id, name))| name == wanted || (!prefix.is_empty() && id.starts_with(prefix))) .collect(); - match matched.as_slice() { - [one] => Ok(one), + match matches.as_slice() { + [(index, _)] => Ok(*index), [] => Err(format!( - "no configured network matches name or id `{wanted}`; \ - `tsng network` lists them" + "no configured network matches name or id `{wanted}`; `tsng network` lists them" )), - several => { - let candidates = several + several => Err(format!( + "`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{}", + several.len(), + several .iter() - .map(|network| format!(" `{}` {}", network.name, network.network_id)) + .map(|(_, (id, name))| format!(" `{name}` {id}")) .collect::>() - .join("\n"); - Err(format!( - "`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{candidates}", - several.len() - )) - } + .join("\n") + )), } } +fn parse_protocol_list(text: &str) -> Result, String> { + if text.eq_ignore_ascii_case("none") { + return Ok(Vec::new()); + } + let mut names = Vec::new(); + for name in text.split(',') { + let name = name.trim(); + let name = if name.eq_ignore_ascii_case("wireguard") { + WG_PROTOCOL + } else { + name + }; + if !PROTOCOLS.iter().any(|spec| spec.name == name) { + return Err(format!( + "unknown protocol `{name}`; use wg,awg,wg-quic,tcp-tls or `none` alone. Run `tsng protocols` for details" + )); + } + if names.iter().any(|n| n == name) { + return Err(format!("protocol `{name}` is listed more than once")); + } + names.push(name.to_owned()); + } + Ok(names) +} + +async fn protocols_command(args: ProtocolsArgs) -> Result<(), Box> { + let Some(selector) = args.network else { + return show_protocols(); + }; + let requested = args + .protocols + .as_deref() + .map(parse_protocol_list) + .transpose()?; + let custom = args.paths.state_dir.is_some(); + let paths = args.paths.resolve()?; + let socket = control_socket(&paths, args.control_socket.as_ref(), custom); + if tsunagi::ipc::is_serving(&socket).await { + let report = tsunagi::ipc::request_status(&socket).await?; + let entries: Vec<_> = report + .networks + .iter() + .map(|n| (n.network_id.clone(), n.name.clone())) + .collect(); + let network = &report.networks[resolve_network_index(&entries, &selector)?]; + let protocols = match requested { + Some(protocols) => { + tsunagi::ipc::set_protocols(&socket, &network.network_id, protocols).await? + } + None => network.protocols.clone(), + }; + println!( + "protocols {} for `{}` ({}) · preference order", + if protocols.is_empty() { + "none".into() + } else { + protocols.join(",") + }, + network.name, + network.network_id + ); + } else { + // Take the directory ownership lock even for an offline edit. Never + // write behind a live agent whose socket cannot be reached. + let storage = tsunagi::storage::Storage::open(&paths)?; + let configured = storage.list_networks().await?; + let network = resolve_network(&configured, &selector)?; + let protocols = match requested { + Some(protocols) => { + storage + .set_protocols(network.network_id, protocols.clone()) + .await?; + Some(protocols) + } + None => network.protocols.clone(), + }; + let shown = protocols + .map(|names| { + if names.is_empty() { + "none".into() + } else { + names.join(",") + } + }) + .unwrap_or_else(|| "defaults from tsng up --protocol".into()); + println!( + "protocols {shown} for `{}` ({}) · agent stopped; applies on next start", + network.name, network.network_id + ); + } + Ok(()) +} + /// `tsng network`: what this device belongs to, and leaving it. async fn network_command(args: NetworkArgs) -> Result<(), Box> { let paths = args.paths.resolve()?; @@ -4445,43 +4575,24 @@ async fn up(args: UpArgs) -> Result<(), Box> { // What the user asked for is checked first, before anything that could // fail for a reason of its own: a misspelled protocol or setting is // their mistake to see, not something to bury under a privilege error. - let wanted: Vec<&ProtocolSpec> = { - let mut wanted = Vec::new(); - for name in args - .protocols - .iter() - .filter(|name| !name.eq_ignore_ascii_case("none")) - { - // Accept "wireguard" as a convenience alias for the canonical "wg". - let canonical = if name.eq_ignore_ascii_case("wireguard") { - WG_PROTOCOL - } else { - name.as_str() - }; - let Some(spec) = PROTOCOLS.iter().find(|spec| spec.name == canonical) else { - let known: Vec<&str> = PROTOCOLS.iter().map(|spec| spec.name).collect(); - return Err(format!( - "this build has no protocol called `{name}`; it has {}. \ - Run `tsng protocols` to see what each one takes.", - known.join(", ") - ) - .into()); - }; - wanted.push(spec); - } - wanted - }; + let initial = parse_protocol_list(&args.protocols.join(","))?; + let wanted: Vec<&ProtocolSpec> = initial + .iter() + .filter_map(|name| PROTOCOLS.iter().find(|spec| spec.name == name)) + .collect(); + config.default_protocols = Some(initial); let settings = parse_settings(&args.protocol_options)?; // A setting nobody takes is a mistake, not a preference: one that was // silently dropped looks exactly like one that did not work. for setting in &settings { - if !wanted + if !PROTOCOLS .iter() + .filter(|spec| setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name)) .any(|spec| !settings_for(spec, std::slice::from_ref(setting)).is_empty()) { return Err(format!( - "no selected protocol takes `{}`; run `tsng protocols` to see what they do", + "no configured protocol takes `{}`; qualify disabled protocol settings as protocol:key=value. Run `tsng protocols` for details", setting.key ) .into()); @@ -4497,7 +4608,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { let mut tcp_tls_plugin = None; let mut pure_wg_plugin: Option> = None; let mut pure_wg_transport: Option> = None; - if !wanted.is_empty() { + if !PROTOCOLS.is_empty() { let tun_factory: Arc = if args.no_tun { Arc::new(MemoryTunFactory::new()) } else { @@ -4507,8 +4618,17 @@ async fn up(args: UpArgs) -> Result<(), Box> { config = config.with_interface(tun_factory, args.interface.clone(), mtu); } - for spec in &wanted { - let options = settings_for(spec, &settings); + for spec in PROTOCOLS { + // Unqualified options keep targeting the initial selection; explicitly + // scoped options also configure protocols that will be enabled later. + let scoped: Vec<_> = settings + .iter() + .filter(|setting| { + setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name) + }) + .cloned() + .collect(); + let options = settings_for(spec, &scoped); match spec.name { AWG_PROTOCOL => { let mut wg = WireguardConfig::new(paths.state_dir.join("awg")); @@ -5112,6 +5232,7 @@ async fn build_report( }; NetworkReport { + protocols: network.protocols.clone(), auto_start: network.auto_start, exit: exit_report(network), broadcast: network.broadcast, @@ -5182,6 +5303,13 @@ async fn build_report( .collect(); StatusReport { + available_protocols: PROTOCOLS + .iter() + .map(|p| tsunagi::ipc::ProtocolReport { + name: p.name.into(), + summary: p.summary.into(), + }) + .collect(), endpoint_id: status.endpoint_id.to_string(), hostname: status.hostname.clone(), bound_sockets: status @@ -5536,6 +5664,7 @@ mod status_tests { /// The situation that prompted this: one peer left and came back. fn network_after_a_peer_returned() -> NetworkReport { NetworkReport { + protocols: vec!["wg-quic".into()], auto_start: true, exit: Default::default(), hostname_conflict: None, @@ -6219,6 +6348,7 @@ mod network_tests { NetworkSecret::from_bytes(&[secret.as_bytes(), &[0u8; 32]].concat()[..32]).unwrap(); let keys = NetworkKeys::derive(&name, &secret); StoredNetwork { + protocols: None, exit_node: false, exit_via: None, broadcast: true, @@ -6442,6 +6572,7 @@ mod network_context_tests { let secret = NetworkSecret::from_bytes([seed; 32]).unwrap(); let keys = NetworkKeys::derive(&name, &secret); StoredNetwork { + protocols: None, exit_node: false, exit_via: None, broadcast: true, diff --git a/crates/tsunagi/tests/network_cli.rs b/crates/tsunagi/tests/network_cli.rs index 192d00a..794de55 100644 --- a/crates/tsunagi/tests/network_cli.rs +++ b/crates/tsunagi/tests/network_cli.rs @@ -54,6 +54,10 @@ fn start_bare(port: u16) -> Running { } fn start_bare_range(port: u16, range: Option<&str>) -> Running { + start_bare_config(port, range, None) +} + +fn start_bare_config(port: u16, range: Option<&str>, protocols: Option<&str>) -> Running { let dir = TempDir::new().unwrap(); let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_tsng")); command @@ -71,6 +75,9 @@ fn start_bare_range(port: u16, range: Option<&str>) -> Running { if let Some(range) = range { command.args(["--ipv4-range", range]); } + if let Some(protocols) = protocols { + command.args(["--protocol", protocols]); + } let child = command.spawn().expect("the agent binary starts"); let mut agent = Running { child, dir }; let runtime = tokio::runtime::Builder::new_current_thread() @@ -112,6 +119,147 @@ fn start(network: &str, port: u16) -> Running { agent } +#[test] +fn protocols_absent_from_startup_can_be_enabled_and_changed_offline() { + let mut agent = start_bare_config(0, None, Some("none")); + assert!( + agent + .run(&[ + "join", + "--network", + "from-control-only", + "--secret", + "a-control-only-test-secret" + ]) + .status + .success() + ); + let initial = agent.run(&["protocols", "from-control-only"]); + assert!(initial.status.success()); + assert!(String::from_utf8_lossy(&initial.stdout).contains("protocols none")); + let enabled = agent.run(&["protocols", "from-control-only", "awg"]); + assert!( + enabled.status.success(), + "{}", + String::from_utf8_lossy(&enabled.stderr) + ); + assert_eq!( + read_networks(&agent.dir)[0].protocols, + Some(vec!["awg".into()]) + ); + agent.child.kill().unwrap(); + agent.child.wait().unwrap(); + let saved = agent.run(&["protocols", "from-control-only", "wg-quic,awg"]); + assert!( + saved.status.success(), + "{}", + String::from_utf8_lossy(&saved.stderr) + ); + assert!(String::from_utf8_lossy(&saved.stdout).contains("agent stopped")); + assert_eq!( + read_networks(&agent.dir)[0].protocols, + Some(vec!["wg-quic".into(), "awg".into()]) + ); +} + +#[test] +fn protocols_are_changed_live_per_network_and_saved_without_changing_the_service() { + let agent = start("protocol-choice", 0); + assert!( + agent + .run(&[ + "join", + "--network", + "other-choice", + "--secret", + "other-secret-for-cli-test" + ]) + .status + .success() + ); + let before = read_networks(&agent.dir); + let other_id = before + .iter() + .find(|n| n.name.as_str() == "other-choice") + .unwrap() + .network_id; + let changed = agent.run(&["protocols", "protocol-choice", "awg,wg-quic"]); + assert!( + changed.status.success(), + "{}", + String::from_utf8_lossy(&changed.stderr) + ); + let shown = agent.run(&["protocols", "protocol-choice"]); + assert!(shown.status.success()); + assert!(String::from_utf8_lossy(&shown.stdout).contains("awg,wg-quic")); + let saved = read_networks(&agent.dir); + assert_eq!( + saved + .iter() + .find(|n| n.name.as_str() == "protocol-choice") + .unwrap() + .protocols, + Some(vec!["awg".into(), "wg-quic".into()]) + ); + assert_eq!( + saved + .iter() + .find(|n| n.network_id == other_id) + .unwrap() + .protocols, + before + .iter() + .find(|n| n.network_id == other_id) + .unwrap() + .protocols + ); + for invalid in ["unknown", "awg,awg", "none,awg"] { + assert!( + !agent + .run(&["protocols", "protocol-choice", invalid]) + .status + .success() + ); + } + assert!(!agent.run(&["protocols", "missing", "awg"]).status.success()); + let disabled = agent.run(&["protocols", "protocol-choice", "none"]); + assert!( + disabled.status.success(), + "{}", + String::from_utf8_lossy(&disabled.stderr) + ); + assert_eq!( + read_networks(&agent.dir) + .iter() + .find(|n| n.name.as_str() == "protocol-choice") + .unwrap() + .protocols, + Some(Vec::new()) + ); + assert!( + agent + .run(&["network", "stop", "protocol-choice"]) + .status + .success() + ); + assert!( + agent + .run(&["protocols", "protocol-choice", "wg-quic"]) + .status + .success() + ); + assert!( + agent + .run(&["network", "start", "protocol-choice"]) + .status + .success() + ); + assert!( + String::from_utf8_lossy(&agent.run(&["protocols", "protocol-choice"]).stdout) + .contains("wg-quic") + ); +} + #[test] fn joining_with_no_secret_makes_one_and_prints_what_to_send() { // The hurry case: a network with somebody, for as long as it is needed, diff --git a/docs/amnezia.md b/docs/amnezia.md index 1741c2b..417b02e 100644 --- a/docs/amnezia.md +++ b/docs/amnezia.md @@ -8,6 +8,21 @@ tsng up --protocol awg tsng up --protocol wg,awg,wg-quic,tcp-tls ``` +The startup list is the initial per-network preference. A saved network choice +overrides it and can enable any built-in protocol at runtime: + +```sh +tsng protocols home awg,wg-quic +tsng protocols home awg +``` + +Changes are local to this device, persist across restarts, and retire existing +data links and end-to-end channels for that network before renegotiation. +Authenticated control sessions, membership, addresses and other networks stay +live. Removing every protocol (`none` in the CLI, all checkboxes off in the tray) +leaves the network control-only. This is different from automatic carrier +fallback, which preserves the selected end-to-end channel. + `awg` is Tsunagi AWG wire version 1. Its packet masking engine is adapted from [RunaRoo/AmneziaWG-RS](https://github.com/RunaRoo/AmneziaWG-RS), with the pinned source and license recorded in diff --git a/docs/testing.md b/docs/testing.md index 7e11cba..82a8157 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -184,3 +184,17 @@ without substituting iroh, membership, encryption or SQLite. It checks runtime loss detection, fallback and fresh profiles, in addition to restart and input validation. Tests are offline; public STUN/DPI/NAT effectiveness needs separate real-network validation. + +Runtime protocol selection tests switch real AWG and iroh/QUIC carriers and +end-to-end WireGuard tunnels from either endpoint-id role, carry full-size +packets, keep another active network's tunnels unchanged, disable and re-enable +the data plane without losing control sessions, and preserve selection after +restart. CLI tests use a real agent process and control socket to exercise +selection by network name, stopped networks, invalid lists, and persisted +control-only configuration, offline updates, and enabling a protocol absent +from the startup list. A real AWG test starts without a TUN, creates the single +memory interface on enablement, and races rapid changes against pending opens. +SQLite tests migrate schema 7 to 8 without changing +identity, secrets or signed records, preserve order and explicit empty choices, +and reject malformed stored selections. These checks use memory TUNs and do +not establish GUI rendering or behavior on an actual OS interface.