diff --git a/README.md b/README.md index f98c7be..6c317b9 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,7 @@ On the first machine: cargo build --release ./target/release/tsunagi id secret generate # prints tsn1...; share it privately ./target/release/tsunagi status # this device, the agent, and this host +./target/release/tsunagi network # the networks this device belongs to ./target/release/tsunagi up --network lab --secret "$SECRET" ``` @@ -495,6 +496,36 @@ directories by default; `--state-dir` and `--cache-dir` override them. One state directory belongs to one live agent, enforced with a real OS file lock rather than an existence check. +### Leaving, and starting over + +Membership outlives a session, so it also has to be possible to end it. + +```bash +tsunagi network # what this device belongs to +tsunagi network leave # give up the address and the name, then forget it +tsunagi wipe --yes # remove everything and be a stranger again +``` + +`leave` publishes a signed `Release` **first**, while the agent is running and +its sessions are up, so the address and the name are freed for the others +instead of staying reserved to a member that has gone. They pass the tombstone +on, so a member that was away hears it from them. With no agent running, +nothing can sign or send it: the command says so and refuses, and `--offline` +drops the network locally while leaving the others holding the old claim. The +protocol key for that network goes too — rejoining is joining, not resuming. + +A network is named by its id, never by its name: two networks can share a +name, and choosing between them for the user is how the wrong one gets left. A +unique prefix is enough. + +`wipe` removes both directories' contents: the device identity, every network, +every signed record and everything a protocol kept beside them. It refuses +while an agent is running, and refuses a directory with no `state.sqlite` in +it, so a mistyped `--state-dir` cannot take somebody's documents with it. +Without `--yes` it only says what it would remove. It is not a goodbye: nobody +is told, because after it there is no key left to sign anything with. Leave the +networks first if the addresses should be freed. + ## Documentation - [docs/architecture.md](docs/architecture.md) — module boundaries and runtime. diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index a921e9b..293ef9c 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -51,6 +51,58 @@ enum Command { Status(StatusArgs), /// Shows the protocols this build can carry packets with. Protocols, + /// Shows the networks this device belongs to, and leaves them. + Network(NetworkArgs), + /// Removes everything this device has stored and starts over. + Wipe(WipeArgs), +} + +#[derive(Debug, Args)] +struct NetworkArgs { + #[command(flatten)] + paths: PathArgs, + + /// Control socket to talk to. Derived from the state directory by default. + #[arg(long, global = true)] + control_socket: Option, + + #[command(subcommand)] + action: Option, +} + +#[derive(Debug, Subcommand)] +enum NetworkAction { + /// Gives up this device's address and name in a network, and forgets it. + /// + /// A signed release goes out first, so the address and name are freed + /// for the others rather than staying reserved to a member that has + /// gone. That needs the agent running; without it nothing can be sent. + Leave { + /// Which network, by id. A unique prefix is enough; the name is not, + /// because two networks may share one. + network: String, + + /// Remove it without telling anybody. + /// + /// For a network nobody else is in, or one joined with a mistyped + /// secret. The others keep whatever this device claimed. + #[arg(long)] + offline: bool, + }, +} + +#[derive(Debug, Args)] +struct WipeArgs { + #[command(flatten)] + paths: PathArgs, + + /// Control socket to check for a running agent. + #[arg(long)] + control_socket: Option, + + /// Actually remove it. Without this the command only says what it would. + #[arg(long)] + yes: bool, } #[derive(Debug, Args)] @@ -383,6 +435,8 @@ async fn run(command: Command) -> Result<(), Box> { Command::Up(args) => up(*args).await, Command::Status(args) => status(args).await, Command::Protocols => show_protocols(), + Command::Network(args) => network_command(args).await, + Command::Wipe(args) => wipe(args).await, } } @@ -908,6 +962,39 @@ impl tsunagi::ipc::unix::ReportSource for AgentControl { .map_err(|err| err.to_string()) }) } + + fn leave( + &self, + network_id: String, + ) -> tsunagi::BoxFuture<'_, Result> { + Box::pin(async move { + let wanted: tsunagi::NetworkId = network_id + .parse() + .map_err(|err| format!("`{network_id}` is not a network id: {err}"))?; + // The name is for the message the user reads, and it is only + // available while the network is still configured. + let name = self + .agent + .list_networks() + .await + .map_err(|err| err.to_string())? + .into_iter() + .find(|network| network.network_id == wanted) + .map(|network| network.name.as_str().to_string()) + .ok_or_else(|| format!("this agent is not in {wanted}"))?; + + let outcome = self + .agent + .leave_network(wanted) + .await + .map_err(|err| err.to_string())?; + Ok(tsunagi::ipc::LeftReport { + name, + announced: outcome.announced, + peers_told: outcome.peers_told as u32, + }) + }) + } } /// The networks this device has joined, read straight from the store. @@ -920,6 +1007,247 @@ fn stored_networks(paths: &StoragePaths) -> Vec .unwrap_or_default() } +/// Finds the one configured network whose id starts with `wanted`. +/// +/// A prefix, because the ids are 52 characters and `status` prints them +/// shortened; the name is deliberately not accepted, since two networks can +/// share one and choosing for the user is how the wrong network gets left. +fn resolve_network<'a>( + networks: &'a [tsunagi::storage::StoredNetwork], + wanted: &str, +) -> Result<&'a tsunagi::storage::StoredNetwork, String> { + let wanted = wanted.trim().trim_end_matches('…'); + if wanted.is_empty() { + return Err("name a network by its id; `tsunagi network` lists them".to_string()); + } + let matched: Vec<&tsunagi::storage::StoredNetwork> = networks + .iter() + .filter(|network| network.network_id.to_string().starts_with(wanted)) + .collect(); + match matched.as_slice() { + [one] => Ok(one), + [] => { + if networks + .iter() + .any(|network| network.name.as_str() == wanted) + { + return Err(format!( + "`{wanted}` is a network name, not an id. Two networks can share a name, \ + so this takes the id; `tsunagi network` lists them." + )); + } + Err(format!( + "no configured network has an id starting `{wanted}`; \ + `tsunagi network` lists them" + )) + } + several => Err(format!( + "`{wanted}` matches {} networks; use more of the id", + several.len() + )), + } +} + +/// `tsunagi network`: what this device belongs to, and leaving it. +async fn network_command(args: NetworkArgs) -> Result<(), Box> { + let paths = args.paths.resolve()?; + let socket = control_socket(&paths, args.control_socket.as_ref()); + match args.action { + None => show_networks(&paths, &socket).await, + Some(NetworkAction::Leave { network, offline }) => { + leave_network(&paths, &socket, &network, offline).await + } + } +} + +/// Every configured network, live where an agent can say so. +async fn show_networks( + paths: &StoragePaths, + socket: &std::path::Path, +) -> Result<(), Box> { + use report::{Health, Report, Row, Section}; + + let stored = stored_networks(paths); + if stored.is_empty() { + eprintln!("no network has been joined"); + return Ok(()); + } + + let observed = observe(paths, socket).await; + let live = match &observed { + Observed::Agent(report) => report.networks.clone(), + Observed::Stored { .. } => Vec::new(), + }; + + let mut out = Report::new(); + let mut section = Section::new("networks"); + for network in &stored { + let id = network.network_id.to_string(); + let running = live.iter().find(|other| other.network_id == id); + let state = match running { + Some(live) if live.active => { + match live.overlay.as_ref().and_then(|o| o.address.clone()) { + Some(address) => format!("running · {address}"), + None => "running · no address agreed yet".to_string(), + } + } + Some(_) => "configured, not running".to_string(), + None => "configured".to_string(), + }; + section.push( + Row::new( + Health::Info, + network.name.as_str().to_string(), + format!("{id} · {state}"), + ) + .with_note(format!( + "leave it with `tsunagi network leave {}`", + short(&id, 10) + )), + ); + } + out.push(section); + print_report("tsunagi networks", &out) +} + +/// Leaves one network, announcing it if there is anything to announce with. +async fn leave_network( + paths: &StoragePaths, + socket: &std::path::Path, + wanted: &str, + offline: bool, +) -> Result<(), Box> { + let networks = stored_networks(paths); + let network = resolve_network(&networks, wanted)?; + let id = network.network_id.to_string(); + let name = network.name.clone(); + + // The running agent does it, because only it can publish the release + // while its sessions are still up. + if socket.exists() { + let report = tsunagi::ipc::unix::leave_network(socket, &id).await?; + println!("left `{}` ({})", report.name, short(&id, 10)); + match (report.announced, report.peers_told) { + (true, 0) => eprintln!( + "\nNobody was connected, so nothing was told: the others keep the address \ + and name this device claimed until it says otherwise, and it no longer can." + ), + (true, peers) => eprintln!( + "\nThe release went to {peers} connected peer(s); they pass it on, so the \ + address and name are freed for the rest as they sync." + ), + (false, _) => eprintln!( + "\nThe network was not running, so nothing was announced: the others keep \ + the address and name this device claimed." + ), + } + return Ok(()); + } + + if !offline { + return Err(format!( + "no agent is running for this state directory, so nothing can announce that \ + `{name}` is being left. Start it and run this again to free the address and \ + name for the others, or pass --offline to drop the network locally and leave \ + them holding it." + ) + .into()); + } + + // Local removal. The storage lock makes sure no agent is using it. + let storage = tsunagi::storage::Storage::open(paths)?; + storage.remove_network(network.network_id).await?; + // What a protocol kept for it goes too; there is no plugin loaded here + // to be asked, so the one this build has is asked directly. + forget_protocol_state(paths, network.network_id); + storage.release_ownership_lock(); + println!("left `{name}` ({}) locally", short(&id, 10)); + eprintln!( + "\nNothing was announced: the others keep the address and name this device \ + claimed in it." + ); + Ok(()) +} + +/// Removes what the compiled-in protocols keep for a network. +/// +/// The offline path has no agent and so no plugins to ask. Each failure is +/// reported and none is fatal: the network is already gone from the state. +fn forget_protocol_state(paths: &StoragePaths, network: tsunagi::NetworkId) { + let store = tsunagi_wg_quic::WireguardConfig::new(paths.state_dir.join("wg-quic")); + match tsunagi_wg_quic::WgKeyStore::open(store.key_store_path()) { + Ok(store) => { + if let Err(err) = store.forget(network) { + eprintln!("warning: the wg-quic key for it could not be removed: {err}"); + } + } + // Never opened means never used, which is nothing to clean up. + Err(err) if !store.key_store_path().exists() => { + let _ = err; + } + Err(err) => eprintln!("warning: the wg-quic key store could not be opened: {err}"), + } +} + +/// `tsunagi wipe`: back to a device that has never joined anything. +async fn wipe(args: WipeArgs) -> Result<(), Box> { + let paths = args.paths.resolve()?; + let socket = control_socket(&paths, args.control_socket.as_ref()); + if socket.exists() { + return Err( + "stop the agent first: a wipe removes the state it is using, and leaving a \ + network properly needs it running anyway" + .into(), + ); + } + + let plan = tsunagi::storage::wipe_plan(&paths)?; + if plan.is_empty() { + println!("nothing stored: this device has never joined anything"); + return Ok(()); + } + + let networks = stored_networks(&paths); + if !args.yes { + println!("`tsunagi wipe --yes` would remove:\n"); + for entry in plan.entries() { + println!(" {}", entry.display()); + } + if !networks.is_empty() { + println!("\nand with it, membership of:\n"); + for network in &networks { + println!(" {} {}", network.name, network.network_id); + } + println!( + "\nNobody is told. Leave each network first — start the agent and run\n\ + `tsunagi network leave ` — to free the address and name it holds\n\ + for the others. Afterwards this device is a stranger: a new identity,\n\ + no networks, and no way to sign anything for the old ones." + ); + } + println!("\nNothing was removed."); + return Ok(()); + } + + let removed = tsunagi::storage::wipe(&paths)?; + // A socket file with nothing behind it is a leftover of the same kind. + if tokio::net::UnixStream::connect(&socket).await.is_err() { + let _ = std::fs::remove_file(&socket); + } + println!("removed {} item(s):", removed.entries().count()); + for entry in removed.entries() { + println!(" {}", entry.display()); + } + if !networks.is_empty() { + eprintln!( + "\nThis device left {} network(s) without telling anybody; they keep what it \ + claimed. The next start generates a new identity and knows nothing.", + networks.len() + ); + } + Ok(()) +} + /// `tsunagi id`: what this device is, and what changes it. async fn id(args: IdArgs) -> Result<(), Box> { let paths = args.paths.resolve()?; @@ -2917,3 +3245,83 @@ mod status_tests { assert!(text.contains("same secret"), "{text}"); } } + +#[cfg(test)] +mod network_tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use tsunagi::identity::{NetworkKeys, NetworkName, NetworkSecret}; + use tsunagi::storage::StoredNetwork; + + fn configured(name: &str, secret: &str) -> StoredNetwork { + let name = NetworkName::new(name).unwrap(); + let secret = + NetworkSecret::from_bytes(&[secret.as_bytes(), &[0u8; 32]].concat()[..32]).unwrap(); + let keys = NetworkKeys::derive(&name, &secret); + StoredNetwork { + network_id: keys.network_id(), + name, + secret, + auto_start: true, + } + } + + #[test] + fn a_prefix_of_the_id_is_enough_and_an_ambiguous_one_is_refused() { + // `status` prints ids shortened, so what a person has to hand is a + // prefix. Accepting it is the difference between leaving a network + // and copying 52 characters correctly. + let networks = vec![configured("lab", "one"), configured("lab", "two")]; + let full = networks[0].network_id.to_string(); + + let picked = resolve_network(&networks, &full[..10]).unwrap(); + assert_eq!(picked.network_id, networks[0].network_id); + // The ellipsis a person copies out of the report is not part of it. + let picked = resolve_network(&networks, &format!("{}…", &full[..10])).unwrap(); + assert_eq!(picked.network_id, networks[0].network_id); + + let err = resolve_network(&networks, "").unwrap_err(); + assert!(err.contains("by its id"), "{err}"); + } + + #[test] + fn a_name_is_refused_because_two_networks_can_share_one() { + // Exactly the situation this command exists for: two networks called + // `lab`, one of them joined with a mistyped secret. Choosing for the + // user here is how the wrong one gets left. + let networks = vec![configured("lab", "one"), configured("lab", "two")]; + let err = resolve_network(&networks, "lab").unwrap_err(); + assert!(err.contains("not an id"), "{err}"); + assert!( + err.contains("tsunagi network"), + "it says where to look: {err}" + ); + } + + #[test] + fn an_id_that_matches_nothing_says_so() { + let networks = vec![configured("lab", "one")]; + let err = resolve_network(&networks, "zzzzzz").unwrap_err(); + assert!(err.contains("no configured network"), "{err}"); + } + + #[test] + fn a_prefix_shared_by_two_networks_is_refused_rather_than_guessed() { + let networks = vec![configured("lab", "one"), configured("other", "two")]; + let shared = &networks[0].network_id.to_string()[..1]; + let both = networks + .iter() + .filter(|network| network.network_id.to_string().starts_with(shared)) + .count(); + if both < 2 { + // The two derived ids happen not to share a first character; + // the empty prefix is the same question with a certain answer. + let err = resolve_network(&networks, "").unwrap_err(); + assert!(err.contains("by its id"), "{err}"); + return; + } + let err = resolve_network(&networks, shared).unwrap_err(); + assert!(err.contains("use more of the id"), "{err}"); + } +} diff --git a/crates/tsunagi-wg-quic/src/plugin.rs b/crates/tsunagi-wg-quic/src/plugin.rs index f19da0c..e9ec2b4 100644 --- a/crates/tsunagi-wg-quic/src/plugin.rs +++ b/crates/tsunagi-wg-quic/src/plugin.rs @@ -792,6 +792,16 @@ impl IpPlugin for WireguardPlugin { self.nudge(Command::Teardown(network)); } + fn on_network_forgotten(&self, network: NetworkId) { + // The key is this protocol's identity in that network and nothing + // else's. Keeping it after leaving would keep a secret for a + // network this agent is no longer in — and hand back the same + // overlay address on a rejoin that everyone else has moved past. + if let Err(err) = self.worker.store.forget(network) { + tracing::warn!(%err, "cannot remove the WireGuard key of a network we left"); + } + } + fn shutdown<'a>(&'a self) -> BoxFuture<'a, ()> { Box::pin(async move { let (reply_tx, reply_rx) = oneshot::channel(); diff --git a/crates/tsunagi-wg-quic/tests/local_control.rs b/crates/tsunagi-wg-quic/tests/local_control.rs index e03619f..d735f1f 100644 --- a/crates/tsunagi-wg-quic/tests/local_control.rs +++ b/crates/tsunagi-wg-quic/tests/local_control.rs @@ -233,3 +233,82 @@ fn the_socket_path_is_derived_and_short_enough() { control_socket_path(&std::path::PathBuf::from("/somewhere/else")) ); } + +/// A source that can also leave, the way the binary's one does. +#[derive(Debug)] +struct Control(Agent); + +impl tsunagi::ipc::unix::ReportSource for Control { + fn report(&self) -> BoxFuture<'_, StatusReport> { + Box::pin(async move { StatusReport::default() }) + } + + fn leave(&self, network_id: String) -> BoxFuture<'_, Result> { + Box::pin(async move { + let wanted: tsunagi::NetworkId = network_id.parse().map_err(|_| "not an id")?; + let name = self + .0 + .list_networks() + .await + .map_err(|err| err.to_string())? + .into_iter() + .find(|network| network.network_id == wanted) + .map(|network| network.name.as_str().to_string()) + .ok_or("not a network this agent is in")?; + let outcome = self + .0 + .leave_network(wanted) + .await + .map_err(|err| err.to_string())?; + Ok(tsunagi::ipc::LeftReport { + name, + announced: outcome.announced, + peers_told: outcome.peers_told as u32, + }) + }) + } +} + +#[tokio::test] +async fn a_client_can_leave_a_network_through_the_running_agent() { + // The release can only be published by the agent that is running, and + // only while its sessions are up, so leaving goes over this socket + // rather than being done behind its back in the state store. + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("control-leave"); + + let dir = TempDir::new().unwrap(); + let agent = Agent::spawn(config_with(dir.path(), &discovery)) + .await + .unwrap(); + let other = TempDir::new().unwrap(); + let peer = Agent::spawn(config_with(other.path(), &discovery)) + .await + .unwrap(); + let network_id = agent.join_network(&name, &secret).await.unwrap(); + peer.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&agent, network_id, 1).await; + + let socket_path = dir.path().join("control.sock"); + let control = ControlSocket::bind(&socket_path, Arc::new(Control(agent.clone()))) + .await + .unwrap(); + + let report = tsunagi::ipc::unix::leave_network(&socket_path, &network_id.to_string()) + .await + .unwrap(); + assert_eq!(report.name, name.as_str()); + assert!(report.announced); + assert_eq!(report.peers_told, 1); + assert!(agent.list_networks().await.unwrap().is_empty()); + + // Asking again names the state it is in rather than failing obscurely. + let err = tsunagi::ipc::unix::leave_network(&socket_path, &network_id.to_string()) + .await + .unwrap_err(); + assert!(err.to_string().contains("not a network"), "{err}"); + + control.shutdown().await; + agent.shutdown().await; + peer.shutdown().await; +} diff --git a/crates/tsunagi-wg-quic/tests/wireguard.rs b/crates/tsunagi-wg-quic/tests/wireguard.rs index 792b223..46380ae 100644 --- a/crates/tsunagi-wg-quic/tests/wireguard.rs +++ b/crates/tsunagi-wg-quic/tests/wireguard.rs @@ -589,6 +589,34 @@ async fn an_interface_that_is_only_in_memory_is_not_reported_as_a_fault() { b.shutdown().await; } +#[tokio::test] +async fn leaving_a_network_takes_its_protocol_key_with_it() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-leave"); + + let agent = WgAgent::spawn(&discovery, "ta").await; + let network_id = agent.agent.join_network(&name, &secret).await.unwrap(); + let before = wait_until("the plugin has a key for it", || async { + Some(agent.plugin.overview(network_id)?.public_key) + }) + .await; + + agent.agent.leave_network(network_id).await.unwrap(); + + // Rejoining is joining, not resuming: the key was this agent's identity + // in a network it left, and everyone there was told to let the address + // it held go. Coming back with the same key would claim an identity the + // network has already released. + agent.agent.join_network(&name, &secret).await.unwrap(); + let after = wait_until("the plugin has a key again", || async { + Some(agent.plugin.overview(network_id)?.public_key) + }) + .await; + assert_ne!(before, after, "a fresh key, not the released one"); + + agent.shutdown().await; +} + #[tokio::test] async fn an_address_is_kept_across_a_restart() { let discovery = SharedMemoryDiscovery::new(); diff --git a/crates/tsunagi/src/agent/mod.rs b/crates/tsunagi/src/agent/mod.rs index 1989d12..fcb1de5 100644 --- a/crates/tsunagi/src/agent/mod.rs +++ b/crates/tsunagi/src/agent/mod.rs @@ -583,11 +583,59 @@ impl Agent { /// Deactivates a network if it is running and removes it from the state /// store together with its cached hints. + /// + /// Local and silent: nobody is told. [`Agent::leave_network`] is the one + /// that says goodbye first, and is what a person means by leaving. pub async fn forget_network(&self, network_id: NetworkId) -> Result<()> { if self.is_active(network_id).await { self.deactivate_network(network_id).await?; } - self.inner.storage.remove_network(network_id).await + self.inner.storage.remove_network(network_id).await?; + // Whatever a protocol kept for this network goes with it. The + // agent cannot know what that is, only that there is no longer any + // reason to hold it. + for plugin in &self.inner.config.plugins { + plugin.on_network_forgotten(network_id); + } + Ok(()) + } + + /// Leaves a network: gives up what was claimed, then forgets it. + /// + /// The order matters and cannot be improved on. A signed `Release` goes + /// out first, while there are still sessions to carry it, so the address + /// and name this agent held are freed for somebody else rather than + /// staying reserved to a member that has gone. Only then is the network + /// deactivated and removed. + /// + /// Reaching every member is not on offer and never could be: a member + /// that is away hears the tombstone from the ones that were here, the + /// same way it hears everything else. Leaving with nobody connected + /// tells nobody, and says so in the outcome rather than pretending. + /// + /// The author's version counter is deliberately **kept**. Rejoining the + /// same network with the same device key must continue from a higher + /// version than the release, or every replica would treat the new claim + /// as stale and ignore it. + pub async fn leave_network(&self, network_id: NetworkId) -> Result { + let mut outcome = LeaveOutcome::default(); + if self.is_active(network_id).await { + let (reply_tx, reply_rx) = oneshot::channel(); + self.command(network_id, NetCommand::Release { reply: reply_tx }) + .await?; + if let Ok(peers) = reply_rx.await { + outcome.announced = true; + outcome.peers_told = peers; + } + // The tombstone is queued on each session, not yet written to + // the wire. A short pause is the difference between peers + // hearing it now and hearing it from somebody else much later. + if outcome.peers_told > 0 { + tokio::time::sleep(RELEASE_FLUSH).await; + } + } + self.forget_network(network_id).await?; + Ok(outcome) } /// Whether a network is currently running. @@ -807,6 +855,29 @@ impl Agent { } } +/// How long a release is given to reach the sessions it was queued on. +/// +/// Short: it is one small message on a connection that is already open, and +/// the alternative to waiting at all is tearing the sessions down underneath +/// it. +const RELEASE_FLUSH: std::time::Duration = std::time::Duration::from_millis(300); + +/// What happened when an agent left a network. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct LeaveOutcome { + /// Whether a signed release was published at all. + /// + /// `false` when the network was not running: there was nothing to + /// publish it from, so this was a local removal only. + pub announced: bool, + /// How many connected peers it was sent to. + /// + /// Zero with `announced` true means the tombstone is in this agent's + /// own state and nowhere else, and it is leaving with it — so nobody + /// will learn of it. + pub peers_told: usize, +} + /// How long each plugin gets to tear itself down during agent shutdown. const PLUGIN_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10); diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 9af7ff1..9d23914 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -65,6 +65,15 @@ pub(crate) enum NetCommand { Reannounce, /// Answer to a different name from now on. SetHostname(String), + /// Give up everything this agent claimed here, and tell whoever is + /// listening. + /// + /// The reply is the number of peers the tombstone was queued to. They + /// pass it on, so a member that is away learns of it from them rather + /// than from an agent that has already gone. + Release { + reply: oneshot::Sender, + }, /// An IP plugin reported an error from one of its own tasks. PluginError { /// Plugin protocol id. @@ -87,6 +96,7 @@ impl std::fmt::Debug for NetCommand { NetCommand::Recheck => f.write_str("Recheck"), NetCommand::Reannounce => f.write_str("Reannounce"), NetCommand::SetHostname(_) => f.write_str("SetHostname"), + NetCommand::Release { .. } => f.write_str("Release"), NetCommand::PluginError { protocol, .. } => write!(f, "PluginError({protocol})"), } } @@ -357,6 +367,13 @@ impl Runtime { } NetCommand::Recheck => self.discovery_round().await, NetCommand::Reannounce => self.reannounce(), + NetCommand::Release { reply } => { + // A tombstone, so the address and the name this agent held + // are freed for somebody else instead of staying reserved + // to a member that has gone. + self.publish_record(RecordBody::Release).await; + let _ = reply.send(self.sessions.len()); + } NetCommand::SetHostname(hostname) => { if self.params.hostname != hostname { self.params.hostname = hostname; @@ -1419,11 +1436,16 @@ impl Runtime { candidates.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes())); // The durable roster. Every author of a signed record is a member, - // including this agent and including members that are not here. + // including this agent and including members that are not here — + // except one that has released everything. That record is a + // tombstone kept so the release cannot be undone by a replica that + // has not heard of it; the author is not a member any more, and + // listing it as one turns leaving into a peer that looks broken. let mut members: Vec = self .state .records() .into_iter() + .filter(|record| !matches!(record.body, RecordBody::Release)) .filter_map(|record| { let endpoint_id = record.author_id().ok()?; Some(MemberStatus { diff --git a/crates/tsunagi/src/dataplane/mod.rs b/crates/tsunagi/src/dataplane/mod.rs index 8d6a7fb..923e8e7 100644 --- a/crates/tsunagi/src/dataplane/mod.rs +++ b/crates/tsunagi/src/dataplane/mod.rs @@ -344,6 +344,19 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static { /// The plugin is expected to remove whatever it created for that network. fn on_network_deactivated(&self, network: NetworkId); + /// Called when this agent has left a network for good. + /// + /// Deactivation is temporary and keeps everything ready for next time; + /// this is the other one. Whatever the plugin holds *durably* for that + /// network — a key of its own, a file, a record — goes now, because the + /// agent is no longer a member and keeping it is keeping a secret for a + /// network it cannot rejoin without being told the secret again. + /// + /// Always preceded by [`IpPlugin::on_network_deactivated`], so this is + /// only about what outlives a session. Defaulted to nothing, for a + /// plugin that stores nothing. + fn on_network_forgotten(&self, _network: NetworkId) {} + /// Called once when the agent shuts down. /// /// The plugin removes the system objects it created and stops its tasks. diff --git a/crates/tsunagi/src/ipc/mod.rs b/crates/tsunagi/src/ipc/mod.rs index ed7aece..2d141a8 100644 --- a/crates/tsunagi/src/ipc/mod.rs +++ b/crates/tsunagi/src/ipc/mod.rs @@ -72,6 +72,12 @@ pub enum Request { /// the change takes effect and reaches peers immediately instead of /// waiting for a restart. SetHostname(String), + /// Leave a network: give up what was claimed, then forget it. + /// + /// Asked of the running agent rather than done behind its back, because + /// only it can publish the release, and only while its sessions are up. + /// The network is named by its id, in the text form `status` prints. + Leave(String), } /// What the agent answers. @@ -82,10 +88,29 @@ pub enum Response { Status(Box), /// The name the agent now answers to, after reducing it to canonical form. Hostname(String), + /// A network was left. + Left(LeftReport), /// The request could not be served. Error(String), } +/// What happened when a network was left. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct LeftReport { + /// The network's name, for the message the user reads. + pub name: String, + /// Whether a signed release was published. + /// + /// `false` when the network was not running: nothing could sign or send + /// it, so this was a local removal and the others keep the old claim. + pub announced: bool, + /// How many connected peers it was handed to. + /// + /// They pass it on, so this is not the number of members that will + /// learn of it — but zero means none of them will. + pub peers_told: u32, +} + /// Everything the agent is doing, in one snapshot. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct StatusReport { diff --git a/crates/tsunagi/src/ipc/unix.rs b/crates/tsunagi/src/ipc/unix.rs index caee30f..8aecd7b 100644 --- a/crates/tsunagi/src/ipc/unix.rs +++ b/crates/tsunagi/src/ipc/unix.rs @@ -16,7 +16,7 @@ use tokio::task::JoinHandle; use crate::BoxFuture; use crate::error::{Error, Result}; -use super::{MAX_MESSAGE_LEN, Request, Response, StatusReport}; +use super::{LeftReport, MAX_MESSAGE_LEN, Request, Response, StatusReport}; /// Builds the report that answers a status request. /// @@ -38,6 +38,14 @@ pub trait ReportSource: Send + Sync + 'static { ) -> BoxFuture<'_, std::result::Result> { Box::pin(async move { Err("this agent cannot change its hostname".to_string()) }) } + + /// Leaves a network, publishing a release first. + /// + /// Defaulted to a refusal for the same reason as the above: a source + /// that only reports says so plainly rather than appearing to do it. + fn leave(&self, _network_id: String) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot leave a network".to_string()) }) + } } impl ReportSource for F @@ -166,6 +174,10 @@ async fn handle(mut stream: UnixStream, source: Arc) -> Result Ok(accepted) => Response::Hostname(accepted), Err(reason) => Response::Error(reason), }, + Request::Leave(network_id) => match source.leave(network_id).await { + Ok(report) => Response::Left(report), + Err(reason) => Response::Error(reason), + }, }; write_message(&mut stream, &response).await } @@ -226,6 +238,20 @@ async fn exchange(path: &Path, request: &Request, within: Duration) -> Result, network_id: &str) -> Result { + let path = path.as_ref(); + let request = Request::Leave(network_id.to_string()); + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Left(report) => Ok(report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + /// Marks the wire format of the local control socket. /// /// `b"TSN"` followed by the version, so a mismatch is recognised as one @@ -237,7 +263,7 @@ async fn exchange(path: &Path, request: &Request, within: Duration) -> Result(stream: &mut UnixStream, value: &T) -> Result<()> { let encoded = postcard::to_stdvec(value) diff --git a/crates/tsunagi/src/storage/mod.rs b/crates/tsunagi/src/storage/mod.rs index a7fda18..66d95b6 100644 --- a/crates/tsunagi/src/storage/mod.rs +++ b/crates/tsunagi/src/storage/mod.rs @@ -350,3 +350,137 @@ impl Storage { guard.take(); } } + +/// What a wipe covers: every entry in the two directories. +/// +/// Reported before anything is removed, and again afterwards, because a +/// command whose whole job is to destroy state has to be able to say exactly +/// what it is about to destroy. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct WipePlan { + /// Entries in the state directory, which is the mandatory state and + /// anything a plugin keeps beside it. + pub state: Vec, + /// Entries in the cache directory. + pub cache: Vec, +} + +impl WipePlan { + /// Whether there is anything to remove at all. + pub fn is_empty(&self) -> bool { + self.state.is_empty() && self.cache.is_empty() + } + + /// Every entry, state first. + pub fn entries(&self) -> impl Iterator { + self.state + .iter() + .chain(self.cache.iter()) + .map(std::path::PathBuf::as_path) + } +} + +/// Lists what [`wipe`] would remove, without removing anything. +pub fn wipe_plan(paths: &StoragePaths) -> Result { + Ok(WipePlan { + state: removable(&paths.state_dir, &paths.state_db())?, + cache: removable(&paths.cache_dir, &paths.cache_db())?, + }) +} + +/// Removes everything in both directories, leaving this device a stranger. +/// +/// The device identity, every network it belongs to, every signed record and +/// everything a plugin kept beside them. The next start generates a new +/// identity and knows nothing, which is the point: after this there is no +/// membership left to be surprised by. +/// +/// What it is **not** is a goodbye. Other members keep the claims this device +/// signed, because signed state has no expiry and there is nobody left here +/// to sign a release. Leaving each network first with +/// [`crate::Agent::leave_network`] is what frees the address and the name. +/// +/// Refuses while an agent owns the state directory, and refuses a directory +/// with no `state.sqlite` in it — a wipe pointed at the wrong place by a +/// mistyped `--state-dir` would otherwise remove somebody's documents. +pub fn wipe(paths: &StoragePaths) -> Result { + let plan = wipe_plan(paths)?; + // Held across the removal, so an agent cannot start into a directory + // that is half gone. The lock file is itself removed at the end: on + // Unix the lock lives on the open handle, not on the name. + let lock = DirectoryLock::acquire(paths.lock_file())?; + + for entry in plan.entries() { + remove_entry(entry)?; + } + drop(lock); + // Created by acquiring the lock a moment ago, so it is ours to remove + // and nothing is left behind claiming the directory. + let _ = std::fs::remove_file(paths.lock_file()); + Ok(plan) +} + +/// Every entry of `dir`, once `marker` proves it is one of ours. +/// +/// An empty or missing directory is nothing to remove rather than an error: +/// wiping twice must be as ordinary as wiping once. +fn removable(dir: &Path, marker: &Path) -> Result> { + if !dir.exists() { + return Ok(Vec::new()); + } + let mut entries = Vec::new(); + let read = std::fs::read_dir(dir).map_err(|source| Error::Io { + path: dir.to_path_buf(), + source, + })?; + for entry in read { + let entry = entry.map_err(|source| Error::Io { + path: dir.to_path_buf(), + source, + })?; + entries.push(entry.path()); + } + if entries.is_empty() { + return Ok(entries); + } + + // The marker is what tells a state directory from a directory that + // merely got named as one. The lock file counts too: an agent that + // crashed before writing anything still leaves that. + let ours = marker.exists() || dir.join("state.lock").exists(); + if !ours { + return Err(Error::Storage(format!( + "{} does not look like a tsunagi directory: no {} in it. Nothing was removed.", + dir.display(), + marker.file_name().map_or_else( + || marker.display().to_string(), + |name| name.to_string_lossy().into_owned() + ) + ))); + } + entries.sort(); + Ok(entries) +} + +fn remove_entry(path: &Path) -> Result<()> { + let metadata = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + // Gone between the plan and the removal is the outcome asked for. + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(source) => { + return Err(Error::Io { + path: path.to_path_buf(), + source, + }); + } + }; + let removed = if metadata.is_dir() { + std::fs::remove_dir_all(path) + } else { + std::fs::remove_file(path) + }; + removed.map_err(|source| Error::Io { + path: path.to_path_buf(), + source, + }) +} diff --git a/crates/tsunagi/tests/cache_and_state.rs b/crates/tsunagi/tests/cache_and_state.rs index 5b58464..ba96d69 100644 --- a/crates/tsunagi/tests/cache_and_state.rs +++ b/crates/tsunagi/tests/cache_and_state.rs @@ -282,3 +282,85 @@ async fn secrets_never_appear_in_status_or_debug_output() { agent.agent.shutdown().await; } + +#[tokio::test] +async fn a_wipe_removes_everything_and_the_next_start_is_a_stranger() { + let dir = tempfile::tempdir().unwrap(); + let paths = StoragePaths::under(dir.path()); + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wiped"); + + let agent = Agent::spawn(config_with(dir.path(), &discovery)) + .await + .unwrap(); + let before = agent.endpoint_id(); + agent.join_network(&name, &secret).await.unwrap(); + // Something a plugin keeps beside the state, which a wipe must take too. + std::fs::create_dir_all(paths.state_dir.join("wireguard")).unwrap(); + std::fs::write(paths.state_dir.join("wireguard/keys.sqlite"), b"key").unwrap(); + + // Not while an agent owns the directory: a half-wiped state under a + // running agent is worse than no wipe at all. + let refused = tsunagi::storage::wipe(&paths).unwrap_err(); + assert!(matches!(refused, Error::StateLocked { .. }), "{refused}"); + agent.shutdown().await; + + let plan = tsunagi::storage::wipe_plan(&paths).unwrap(); + assert!( + plan.entries().any(|path| path.ends_with("state.sqlite")), + "{plan:?}" + ); + assert!( + plan.entries().any(|path| path.ends_with("wireguard")), + "what a plugin kept is state too: {plan:?}" + ); + + let wiped = tsunagi::storage::wipe(&paths).unwrap(); + assert_eq!(wiped, plan); + assert!(!paths.state_db().exists()); + assert!(!paths.state_dir.join("wireguard").exists()); + assert!(!paths.lock_file().exists(), "no lock is left claiming it"); + + // A stranger: new identity, no networks, nothing to be surprised by. + let fresh = Agent::spawn(config_with(dir.path(), &discovery)) + .await + .unwrap(); + assert_ne!(fresh.endpoint_id(), before); + assert!(fresh.list_networks().await.unwrap().is_empty()); + fresh.shutdown().await; +} + +#[tokio::test] +async fn wiping_twice_is_as_ordinary_as_wiping_once() { + let dir = tempfile::tempdir().unwrap(); + let paths = StoragePaths::under(dir.path()); + let discovery = SharedMemoryDiscovery::new(); + + let agent = Agent::spawn(config_with(dir.path(), &discovery)) + .await + .unwrap(); + agent.shutdown().await; + + assert!(!tsunagi::storage::wipe(&paths).unwrap().is_empty()); + let again = tsunagi::storage::wipe(&paths).unwrap(); + assert!(again.is_empty(), "nothing left to remove: {again:?}"); +} + +#[tokio::test] +async fn a_directory_that_is_not_ours_is_refused_rather_than_emptied() { + // The mistyped `--state-dir` that would otherwise remove somebody's + // documents. A marker decides, not the name of the directory. + let dir = tempfile::tempdir().unwrap(); + let paths = StoragePaths::new(dir.path(), dir.path().join("cache")); + std::fs::write(dir.path().join("thesis.txt"), b"years of work").unwrap(); + + let err = tsunagi::storage::wipe(&paths).unwrap_err(); + assert!( + err.to_string().contains("does not look like a tsunagi"), + "{err}" + ); + assert!( + dir.path().join("thesis.txt").exists(), + "nothing was removed" + ); +} diff --git a/crates/tsunagi/tests/leaving.rs b/crates/tsunagi/tests/leaving.rs new file mode 100644 index 0000000..563fca4 --- /dev/null +++ b/crates/tsunagi/tests/leaving.rs @@ -0,0 +1,171 @@ +//! Scenario 11: leaving a network, and what the others are left holding. +//! +//! Leaving is not deactivating and not forgetting. It is a signed statement +//! that this member gives up what it claimed, published while there is still +//! somebody to hear it, because signed state has no expiry and nothing else +//! will ever free the address. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use tsunagi::discovery::SharedMemoryDiscovery; +use tsunagi::testing::{TestAgent, network, wait_for_peers, wait_until}; + +#[tokio::test] +async fn leaving_frees_the_address_for_everyone_still_there() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("leaving-frees"); + + let leaver = TestAgent::spawn(&discovery).await.unwrap(); + let stayer = TestAgent::spawn(&discovery).await.unwrap(); + let network_id = leaver.agent.join_network(&name, &secret).await.unwrap(); + stayer.agent.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&stayer.agent, network_id, 1).await; + + let leaver_id = leaver.agent.endpoint_id(); + let address = wait_until("the one leaving holds an address", || async { + let status = stayer.agent.network_status(network_id).await.ok()?; + status + .members + .iter() + .find(|member| member.endpoint_id == leaver_id)? + .overlay_address_v4 + }) + .await; + + let outcome = leaver.agent.leave_network(network_id).await.unwrap(); + assert!(outcome.announced, "there was a session to announce it on"); + assert_eq!(outcome.peers_told, 1); + + // The one still there drops it from the roster. The tombstone stays in + // the record set — it has to, or a replica that never heard of it would + // reinstate the old claim — but a member that gave everything up is not + // a member, and listing it as one makes leaving look like a fault. + wait_until("the member is gone from the roster", || async { + let status = stayer.agent.network_status(network_id).await.ok()?; + status + .members + .iter() + .all(|member| member.endpoint_id != leaver_id) + .then_some(()) + }) + .await; + let taken = stayer.agent.network_status(network_id).await.unwrap(); + assert!( + !taken + .members + .iter() + .any(|member| member.overlay_address_v4 == Some(address)), + "the address is free for somebody else" + ); + + // And locally there is no membership left to be surprised by. + assert!( + leaver.agent.list_networks().await.unwrap().is_empty(), + "the network is gone from the store" + ); + assert!(!leaver.agent.is_active(network_id).await); + + leaver.agent.shutdown().await; + stayer.agent.shutdown().await; +} + +#[tokio::test] +async fn leaving_with_nobody_connected_says_so_rather_than_pretending() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("leaving-alone"); + + let agent = TestAgent::spawn(&discovery).await.unwrap(); + let network_id = agent.agent.join_network(&name, &secret).await.unwrap(); + + // Nobody is here, so the tombstone reaches nobody. The network is still + // left — the caller asked — but the outcome does not claim an audience + // there was not one for. + let outcome = agent.agent.leave_network(network_id).await.unwrap(); + assert!(outcome.announced, "it was published locally"); + assert_eq!(outcome.peers_told, 0); + assert!(agent.agent.list_networks().await.unwrap().is_empty()); + + agent.agent.shutdown().await; +} + +#[tokio::test] +async fn leaving_a_network_that_is_not_running_tells_nobody() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("leaving-inactive"); + + let agent = TestAgent::spawn(&discovery).await.unwrap(); + let network_id = agent.agent.join_network(&name, &secret).await.unwrap(); + agent.agent.deactivate_network(network_id).await.unwrap(); + + // There is no runtime to sign and send from, so nothing was announced + // and the outcome says exactly that instead of a quiet success. + let outcome = agent.agent.leave_network(network_id).await.unwrap(); + assert!(!outcome.announced); + assert_eq!(outcome.peers_told, 0); + assert!(agent.agent.list_networks().await.unwrap().is_empty()); + + agent.agent.shutdown().await; +} + +#[tokio::test] +async fn rejoining_after_leaving_is_not_taken_for_a_stale_record() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("leaving-rejoin"); + + let returner = TestAgent::spawn(&discovery).await.unwrap(); + let stayer = TestAgent::spawn(&discovery).await.unwrap(); + let network_id = returner.agent.join_network(&name, &secret).await.unwrap(); + stayer.agent.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&stayer.agent, network_id, 1).await; + + let returner_id = returner.agent.endpoint_id(); + wait_until("the one leaving holds an address", || async { + stayer + .agent + .network_status(network_id) + .await + .ok()? + .members + .iter() + .find(|member| member.endpoint_id == returner_id)? + .overlay_address_v4 + }) + .await; + + returner.agent.leave_network(network_id).await.unwrap(); + wait_until("the release reached the other one", || async { + let status = stayer.agent.network_status(network_id).await.ok()?; + status + .members + .iter() + .all(|member| member.endpoint_id != returner_id) + .then_some(()) + }) + .await; + + // The same device, the same key, the same network. The version counter + // survived leaving on purpose: a claim numbered below the release would + // be ignored by every replica that already has the release, and this + // member would be invisible for good. + returner.agent.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&stayer.agent, network_id, 1).await; + let again = wait_until("the returning member is seen again", || async { + stayer + .agent + .network_status(network_id) + .await + .ok()? + .members + .iter() + .find(|member| member.endpoint_id == returner_id)? + .overlay_address_v4 + }) + .await; + assert!( + tsunagi::state::DEFAULT_IPV4_RANGE.contains(again), + "an address in the network's range: {again}" + ); + + returner.agent.shutdown().await; + stayer.agent.shutdown().await; +} diff --git a/docs/testing.md b/docs/testing.md index a08e732..46e118b 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -40,6 +40,7 @@ of several system processes, and is not presented as one. | 8 | a dead candidate and a vanished peer do not block the others; retries are bounded and stop when the network is deactivated | `tests/resilience.rs` | | 9 | wrong version, a message before authentication, a proof replayed on another connection, an oversized frame and a `Hello` for an inactive network are all rejected without taking the agent down | `tests/authentication.rs` | | 10 | a second agent on the same state directory gets a clear error; after a clean stop the directory reopens; shutdown ends background tasks and refuses further work; independent agents coexist in one process | `tests/resilience.rs` | +| 11 | leaving a network frees the address for the others, says plainly when there was nobody to tell, and rejoining afterwards is not mistaken for a stale record; a wipe empties both directories and the next start is a stranger, while a directory that is not ours is refused | `tests/leaving.rs`, `tests/cache_and_state.rs` | `tests/wireguard.rs` drives the WireGuard data plane over real iroh connections. Everything is real except the packet interface: real agents, real