ab/tsunagi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Let a device leave a network, and start over

Browse Source 
Joining was one command and leaving was nothing at all: a network went into
`state.sqlite` on the first `up` and stayed there, so a mistyped secret
left a second network beside the working one with no way to remove it but
editing the database by hand.

`tsunagi network` lists what this device belongs to. `tsunagi network leave
<id>` publishes a signed release first — while the agent is running and its
sessions are up — and only then deactivates the network and removes it. The
order is the whole point: signed state has no expiry, so the tombstone is
the only thing that ever frees the address and the name for the others, and
after the network is gone there is nothing left here to sign one with.
Peers pass it on, so a member that was away hears it from them rather than
from an agent that has already left.

With no agent running nothing can sign or send, and the command says so
instead of quietly succeeding: `--offline` drops the network locally and
says plainly that the others keep the old claim. The outcome always
distinguishes "published to nobody" from "not published at all", because
they leave the network in different states.

A network is named by its id, and a unique prefix will do. The name is
refused on purpose: two networks can share one — that is exactly the
situation this command exists for — and picking between them for the user
is how the wrong one gets left.

The author's version counter deliberately survives. Rejoining the same
network with the same key must continue above the release, or every replica
that holds the release would treat the new claim as stale and the returning
member would be invisible for good. The protocol key does not survive:
rejoining is joining, not resuming, and coming back with a key the network
was told to let go claims an identity nobody holds any more. Plugins learn
about it through a new `on_network_forgotten`, which is about what outlives
a session rather than what a deactivation tears down.

A released member also drops out of the roster `status` prints. The
tombstone stays in the record set — a replica that never heard of it would
otherwise reinstate the old claim — but listing an author that gave
everything up as a member made leaving look like a peer that had broken.

`tsunagi wipe` is the other half: it empties both directories, so the
device identity, every network, every signed record and everything a
protocol kept beside them go at once and the next start is a stranger. It
refuses while an agent holds the directory, 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 prints what it would remove and
what membership would be lost. It is not a goodbye and says so: leaving the
networks first is what frees their addresses.

The local control protocol is 8 — the socket carries a `Leave` request now,
since only the running agent can publish the release.

Exercised end to end against real agents: leaving by prefix released the
address to a connected peer, leaving by name was refused, `--offline` was
refused until asked for explicitly, wipe was refused while the agent ran,
and the directory afterwards had no identity in it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
2026-09-21 21:54:05 +01:00
co-authored by Claude Opus 5
parent fae3816bbf
commit 41604225ba
14 changed files with 1105 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tsunagi-cli/src/main.rs
+408
View File
@@ -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<PathBuf>,
#[command(subcommand)]
action: Option<NetworkAction>,
}
#[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<PathBuf>,
/// 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<dyn std::error::Error>> {
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<tsunagi::ipc::LeftReport, String>> {
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<tsunagi::storage::StoredNetwork>
.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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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 <id>` — 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<dyn std::error::Error>> {
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}");
}
}
crates/tsunagi-wg-quic/src/plugin.rs
+10
View File
@@ -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();
crates/tsunagi-wg-quic/tests/local_control.rs
+79
View File
@@ -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<tsunagi::ipc::LeftReport, String>> {
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;
}
crates/tsunagi-wg-quic/tests/wireguard.rs
+28
View File
@@ -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();
crates/tsunagi/src/agent/mod.rs
+72 -1
View File
@@ -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<LeaveOutcome> {
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);
crates/tsunagi/src/agent/network.rs
+23 -1
View File
@@ -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<usize>,
},
/// 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<MemberStatus> = self
.state
.records()
.into_iter()
.filter(|record| !matches!(record.body, RecordBody::Release))
.filter_map(|record| {
let endpoint_id = record.author_id().ok()?;
Some(MemberStatus {
crates/tsunagi/src/dataplane/mod.rs
+13
View File
@@ -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.
crates/tsunagi/src/ipc/mod.rs
+25
View File
@@ -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<StatusReport>),
/// 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 {
crates/tsunagi/src/ipc/unix.rs
+28 -2
View File
@@ -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<String, String>> {
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<LeftReport, String>> {
Box::pin(async move { Err("this agent cannot leave a network".to_string()) })
}
}
impl<F> ReportSource for F
@@ -166,6 +174,10 @@ async fn handle(mut stream: UnixStream, source: Arc<dyn ReportSource>) -> 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<Re
}
}
/// Asks a running agent to leave a network.
///
/// The agent publishes the release and removes the network; this only
/// carries the request and the outcome.
pub async fn leave_network(path: impl AsRef<Path>, network_id: &str) -> Result<LeftReport> {
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<Re
///
/// 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', 7]);
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 8]);
async fn write_message<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> Result<()> {
let encoded = postcard::to_stdvec(value)
crates/tsunagi/src/storage/mod.rs
+134
View File
@@ -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<std::path::PathBuf>,
/// Entries in the cache directory.
pub cache: Vec<std::path::PathBuf>,
}
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<Item = &std::path::Path> {
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<WipePlan> {
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<WipePlan> {
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<Vec<std::path::PathBuf>> {
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,
})
}
crates/tsunagi/tests/cache_and_state.rs
+82
View File
@@ -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"
);
}
crates/tsunagi/tests/leaving.rs
+171
View File
@@ -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;
}