Compare commits

...
Author SHA1 Message Date
Aleksandr Bogomiakov 90ef5cefec Retire offline node claims with equal network-wide cleanup rights
Offline devices reserved names and IPv4 addresses indefinitely, leaving stale
members in the device history. Give every authenticated network participant
the same right to retire another device's observed signed claim without an
owner, quorum, permanent identity ban or secret change.

Add a separately signed retirement containing the target's original signed
claim as proof. Verify both signatures and network scopes, retain the strongest
proved cutoff per target with deterministic ties, and suppress claim versions
at or below it in membership, hostname ownership and IPv4 allocation. Keep the
old signed records and range agreement, and retain markers after a newer claim
returns so old replicas cannot resurrect obsolete reservations. A returning
device signs a newer claim, allocates against current holders and resets its
previous hostname priority rather than taking a reused name with an old time.

Commit local retirement and the requester's author counter atomically before
announcing it; receivers persist proofs before applying them. Bound history to
1024 targets by default, reject additional targets without evicting protection,
and treat corrupt or over-limit mandatory storage as an explicit error. Repeat
bounded marker batches with a per-session rotating cursor so queue pressure
cannot indefinitely starve later markers. Exclude retired claims and cached
hints from history-derived discovery without preventing a fresh authenticated
device from returning.

Expose Agent::forget_node through the shared local control API. CLI command
tsng network forget-node accepts exact network names, full ids or unique id
prefixes, and exact node hostnames, ids, prefixes or overlay IPv4 addresses.
Ambiguous selections change nothing. Recheck the observed version and refuse
self, online nodes, the selected exit node and absent or changed claims.
Add Forget with an identity confirmation to offline tray device rows. Widen
Show devices to 1440x640 with an 800x400 minimum to fit its contents.

Control ALPN/version advances to tsunagi/ctrl/7 and local IPC to 21; upgrade
all peers and restart the agent with matching CLI/tray clients. Data ALPN 5,
network identity derivation, handshake encoding and existing record signing
bytes remain unchanged. SQLite schema 9 adds retirement storage while retaining
identity, secrets, settings and signed claims. Document behavior and limits.

Validation on Linux through nix develop:
- cargo fmt --all -- --check
- cargo clippy --locked --workspace --all-targets -- -D warnings
- cargo test --locked --workspace --all-targets: 519 passed, 2 existing ignored
  tests; inherited TSUNAGI_CONTROL_SOCKET and TSUNAGI_CONTROL_GROUP were unset
- git diff --check and the built CLI forget-node help

Tests use real iroh authentication and SQLite. Cover restart, stale-replica
recovery, equal rights and network isolation, hostname reuse and return, twelve
marker batches through a two-message queue, malformed signatures and proofs,
invented future cutoffs, deterministic merge, capacity, migration and injected
counter-write rollback, selectors and cleanup through two real CLI processes.
Earlier iterations caught a private test API, a collapsible tray conditional
and a crash-based CLI test waiting for QUIC's idle timeout; corrected before
the final successful suite. Manual tray rendering, macOS/Windows, privileged
TUNs and public-network traversal were not run. The requested RC tag is
v0.1.0-rc.31; no push was requested.
2026-10-12 18:04:29 +01:00
Aleksandr Bogomiakov 13bdcaf98d Render protocol priority buttons with Phosphor icons
The tray protocol selection used ordinary Unicode up/down arrows for its
priority buttons. The default UI fonts lack these glyphs, so both buttons
rendered missing-glyph boxes. Use the regular Phosphor ARROW_UP and ARROW_DOWN
icons already installed in the application's proportional font fallback.

Validation on Linux through nix develop:
- cargo fmt --all -- --check
- cargo clippy --locked --workspace --all-targets -- -D warnings
- cargo test --locked --workspace --all-targets, with inherited control socket
  and group overrides removed: 509 passed, 2 existing ignored tests

No new tests were added for the two glyph substitutions. Manual tray rendering
and macOS/Windows checks were not performed. No release tag was requested for
this fix.
2026-10-12 17:19:53 +01:00
Aleksandr Bogomiakov d941cdac7a Change network protocol selections at runtime
Protocol selection previously belonged to tsng up and therefore to the
service command line. A blocked or degraded carrier could only be removed
by editing the service and restarting the whole agent. Make the ordered
selection a local per-network setting instead.

Add Agent::set_protocols and a local control request shared by CLI and tray.
Keep tsng protocols without arguments as the protocol catalog; naming a
network shows its selection and supplying a comma-separated list replaces
it. Accept exact names, full ids and unique prefixes using shared selector
rules; ambiguous selectors and invalid lists change nothing. An explicit
none selects control-only operation. Offline edits take the directory lock
and apply on the next start; live edits never write behind the daemon.

Register every built-in plugin and transport in the CLI. Treat up --protocol
as ordered initial defaults for networks without an override, allowing a
protocol absent from the unit to be enabled later. Retain the targeting of
unqualified startup options and allow explicitly scoped options to configure
initially disabled protocols. Persist NULL for inherited preferences and an
empty selection distinctly in SQLite schema 8; migration preserves existing
identities, secrets, signed records and counters.

Serialize changes with network lifecycle operations and commit preferences
before applying or announcing them. Retire only the affected network's data
links and E2E channels, cancel pending opens, and reject obsolete completion
results by attempt id. Clear peer data channels when authenticated protocol
capabilities change while retaining control sessions and topology. Replay
allocations and peer identities after asynchronously prepared plugins become
ready. A control-only start defers creation of the agent's single interface
until data is enabled; plugins receive a sink that can use it later. Automatic
carrier fallback continues to preserve E2E state; explicit policy changes
renegotiate it. Identity derivation and peer wire formats are unchanged.

Add a collapsible protocols row to each tray network tile, with a catalog
provided by the daemon, descriptions, immediately applied checkboxes and
preference-order buttons. Keep editing disabled until the worker refreshes
status so a second click cannot overwrite an edit using a stale snapshot.
Local IPC wire version is 20 for the new request and report fields; upgrade
and restart agent, CLI and tray together. Crate release versions remain unchanged; the requested RC tag is v0.1.0-rc.30.

Validation on Linux through nix develop:
- cargo fmt --all -- --check
- cargo clippy --locked --workspace --all-targets -- -D warnings
- cargo test --locked --workspace --all-targets: 509 passed, 2 existing ignored
  tests, with inherited TSUNAGI_CONTROL_SOCKET and TSUNAGI_CONTROL_GROUP removed

Tests use real iroh membership, SQLite, UDP, AWG and WireGuard encryption,
QUIC carriers and memory TUNs. Cover either endpoint-id role, full-size packet
delivery after switching, an unaffected second active network, saved selection
after restart, empty/invalid selections, first enablement without restart,
rapid changes racing old opens, live and offline CLI changes, startup-absent
protocols and preservation of signed state through migration. Earlier compile
and focused-test iterations caught initializer mistakes, a short test secret,
asynchronous plugin allocation readiness, and an inherited system socket
path; corrected before the successful final checks. Plain cargo was unavailable
in the host shell, so validation used the Nix toolchain. Manual tray interaction,
macOS/Windows, privileged host TUNs and public-network traversal were not run.
2026-10-12 16:19:18 +01:00
28 changed files with 3069 additions and 179 deletions

No files matched your search

+35
View File
@@ -112,6 +112,41 @@ fresh masking profile and outer keys for each pair connection attempt, including
reentry after fallback. Sustained data loss triggers switching independently of
the iroh control connection. [Configuration and limits](docs/amnezia.md).
Change a network's protocol selection without editing its service or restarting:
```sh
tsng protocols # available protocols and settings
tsng protocols home # this device's choice for home
tsng protocols home awg,wg-quic # prefer AWG, retain QUIC as fallback
tsng protocols home none # control plane only
```
Exact network names, full ids and unique id prefixes work here. The choice is
local to this device, saved per network and applied immediately; an offline
change applies on the next start. In the tray, expand **protocols** on a network
tile to toggle protocols or change their preference order. Both endpoints need
a common protocol at the same wire version. `tsng up --protocol` supplies the
initial defaults for networks without a saved choice; it does not restrict
which built-in protocols can be enabled later.
Remove an obsolete offline device from a network's shared history:
```sh
tsng network forget-node home old-laptop
tsng network forget-node home 10.13.37.143
```
Networks accept exact names, full ids or unique id prefixes; nodes accept those
id selectors, exact hostnames or overlay IPv4 addresses. Ambiguous selectors
change nothing. Start the network first if it is stopped. In **Show devices**,
click **Forget** beside an offline node and confirm its identity.
Every member has the same cleanup right. The signed cleanup frees the observed
claim's name and IPv4 reservation, survives restarts and reaches offline replicas
when they reconnect. It does not ban the device: a newer signed claim can return.
Online nodes and the currently selected exit node cannot be forgotten.
[State and limits](docs/sync-model.md#offline-node-cleanup).
## Platforms
| Platform | Build | CLI `tsng` | Tray | Exit node: offer / use | Packages |
+245 -2
View File
@@ -25,6 +25,7 @@ use tsunagi_wg_quic::WireguardConfig;
struct Node {
agent: Agent,
plugin: Arc<AwgPlugin>,
quic: Option<Arc<tsunagi_wg_quic::WireguardPlugin>>,
transport: Arc<AwgTransport>,
tuns: MemoryTunFactory,
loss: Arc<AtomicBool>,
@@ -47,6 +48,16 @@ impl Node {
interface: &str,
fallback: bool,
mtu: u32,
) -> Self {
Self::open_selected(root, discovery, interface, fallback, mtu, None).await
}
async fn open_selected(
root: &Path,
discovery: &SharedMemoryDiscovery,
interface: &str,
fallback: bool,
mtu: u32,
default_protocols: Option<Vec<String>>,
) -> Self {
let identity =
tsunagi::storage::StateStore::open(tsunagi::StoragePaths::under(root).state_db())
@@ -70,9 +81,9 @@ impl Node {
let tuns = MemoryTunFactory::new();
let mut config = config_with(root, discovery)
.with_interface(Arc::new(tuns.clone()), interface, mtu)
.with_overlay_ipv4_range(Some(tsunagi::state::DEFAULT_IPV4_RANGE))
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>)
.with_custom_transport(AWG_PROTOCOL, faults);
let mut quic_plugin = None;
if fallback {
let quic = tsunagi_wg_quic::WireguardPlugin::open(
WireguardConfig::new(root.join("wg-quic"))
@@ -81,8 +92,10 @@ impl Node {
)
.await
.unwrap();
config = config.with_plugin(quic);
config = config.with_plugin(quic.clone());
quic_plugin = Some(quic);
}
config.default_protocols = default_protocols;
let agent = Agent::spawn(config).await.unwrap();
let inbound_agent = agent.clone();
let inbound_transport = transport.clone();
@@ -96,6 +109,7 @@ impl Node {
Self {
agent,
plugin,
quic: quic_plugin,
transport,
tuns,
loss,
@@ -163,6 +177,235 @@ async fn exchange(a: &Node, b: &Node, id: NetworkId) {
);
}
}
#[tokio::test]
async fn a_control_only_agent_enables_real_data_without_restart_and_cancels_stale_attempts() {
let discovery = SharedMemoryDiscovery::new();
let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()];
let a = Node::open_selected(
roots[0].path(),
&discovery,
"initial-none-a",
true,
1280,
Some(vec![]),
)
.await;
let b = Node::open_selected(
roots[1].path(),
&discovery,
"initial-none-b",
true,
1280,
Some(vec![]),
)
.await;
let (name, secret) = network("enable-after-control-only");
let id = a.agent.join_network(&name, &secret).await.unwrap();
b.agent.join_network(&name, &secret).await.unwrap();
wait_for_peers(&a.agent, id, 1).await;
wait_for_peers(&b.agent, id, 1).await;
assert!(a.agent.overlay().is_none());
assert!(b.agent.overlay().is_none());
for node in [&a, &b] {
assert!(
node.agent
.network_status(id)
.await
.unwrap()
.protocols
.is_empty()
);
node.agent
.set_protocols(id, vec!["awg".into()])
.await
.unwrap();
assert!(node.agent.overlay().is_some());
}
a.ready(id, 1).await;
b.ready(id, 1).await;
exchange(&a, &b, id).await;
let before = a.agent.overlay().unwrap().interface;
// Changes race unfinished opens and prepared plugin callbacks; the last
// selection must win even if an earlier open has completed in its queue.
for selection in [
vec!["wg-quic".into()],
vec![],
vec!["awg".into()],
vec![],
vec!["awg".into()],
] {
a.agent.set_protocols(id, selection).await.unwrap();
}
a.ready(id, 1).await;
b.ready(id, 1).await;
exchange(&a, &b, id).await;
assert_eq!(a.agent.overlay().unwrap().interface, before);
for node in [&a, &b] {
let status = node.agent.network_status(id).await.unwrap();
assert_eq!(status.protocols, ["awg"]);
assert_eq!(status.metrics.sessions_established, 1);
assert_eq!(status.metrics.disconnects, 0);
}
a.agent.shutdown().await;
b.agent.shutdown().await;
}
#[tokio::test]
async fn runtime_protocol_selection_replaces_real_tunnels_and_survives_restart() {
let discovery = SharedMemoryDiscovery::new();
let roots = [TempDir::new().unwrap(), TempDir::new().unwrap()];
let a = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await;
let b = Node::open_with_fallback(roots[1].path(), &discovery, "runtime-b", true).await;
let (name, secret) = network("runtime-protocols");
let id = a.agent.join_network(&name, &secret).await.unwrap();
b.agent.join_network(&name, &secret).await.unwrap();
a.ready(id, 1).await;
b.ready(id, 1).await;
exchange(&a, &b, id).await;
let (other_name, other_secret, other_id) = loop {
let (name, secret) = network("unchanged-runtime-network");
let other_id = tsunagi::identity::NetworkKeys::derive(&name, &secret).network_id();
if tsunagi::state::derived_ipv4_range(id) != tsunagi::state::derived_ipv4_range(other_id) {
break (name, secret, other_id);
}
};
a.agent
.join_network(&other_name, &other_secret)
.await
.unwrap();
b.agent
.join_network(&other_name, &other_secret)
.await
.unwrap();
a.ready(other_id, 1).await;
b.ready(other_id, 1).await;
let other_before = a
.agent
.network_status(other_id)
.await
.unwrap()
.metrics
.data_links_established;
for node in [&a, &b] {
let before = node.agent.network_status(id).await.unwrap().protocols;
for invalid in [vec!["unknown".into()], vec!["awg".into(), "awg".into()]] {
assert!(node.agent.set_protocols(id, invalid).await.is_err());
assert_eq!(
node.agent.network_status(id).await.unwrap().protocols,
before
);
}
}
// Exercise updates initiated by either endpoint-id role, including an
// empty set. Membership and the control session stay alive throughout.
for node in [&a, &b] {
node.agent.set_protocols(id, Vec::new()).await.unwrap();
wait_until("peer accepts control-only selection", || async {
let other = if node.agent.endpoint_id() == a.agent.endpoint_id() {
&b
} else {
&a
};
let status = other.agent.network_status(id).await.ok()?;
(status.peers.len() == 1
&& status.peers[0].protocols.is_empty()
&& matches!(status.peers[0].data_path, tsunagi::agent::DataPath::None))
.then_some(())
})
.await;
assert_eq!(node.agent.network_status(id).await.unwrap().peers.len(), 1);
node.agent
.set_protocols(id, vec!["wg-quic".into(), "awg".into()])
.await
.unwrap();
}
a.agent
.set_protocols(id, vec!["wg-quic".into()])
.await
.unwrap();
wait_until("real QUIC replaces AWG", || async {
let a_status = a.agent.network_status(id).await.ok()?;
let b_status = b.agent.network_status(id).await.ok()?;
(a_status.peers[0].protocols == ["wg-quic"]
&& b_status.peers[0].protocols == ["wg-quic"]
&& !matches!(a_status.peers[0].data_path, tsunagi::agent::DataPath::None)
&& !matches!(b_status.peers[0].data_path, tsunagi::agent::DataPath::None)
&& a.transport.overview(id).iter().all(|p| p.closed)
&& b.transport.overview(id).iter().all(|p| p.closed))
.then_some(())
})
.await;
// Get addresses from the system allocation: the disabled AWG plugin no
// longer has a live view of this network.
for (source, dest) in [(&a, &b), (&b, &a)] {
wait_until("QUIC end-to-end handshake", || async {
source
.quic
.as_ref()?
.overview(id)?
.peers
.iter()
.any(|p| p.tunnel.as_ref().is_some_and(|t| t.health.is_up()))
.then_some(())
})
.await;
let addr = |n: &Node| {
n.quic
.as_ref()
.unwrap()
.overview(id)
.unwrap()
.overlay_address_v4
.unwrap()
};
let bytes = packet(addr(source), addr(dest), 1280);
source.tun().push_from_os(bytes.clone());
assert_eq!(
tokio::time::timeout(Duration::from_secs(5), dest.tun().pop_to_os())
.await
.unwrap()
.unwrap(),
bytes
);
}
assert_eq!(
a.agent
.network_status(other_id)
.await
.unwrap()
.metrics
.data_links_established,
other_before
);
exchange(&a, &b, other_id).await;
a.agent.shutdown().await;
let restarted = Node::open_with_fallback(roots[0].path(), &discovery, "runtime-a", true).await;
wait_for_peers(&restarted.agent, id, 1).await;
assert_eq!(
restarted.agent.network_status(id).await.unwrap().protocols,
["wg-quic"]
);
assert_eq!(
b.agent.network_status(other_id).await.unwrap().protocols,
["awg", "wg-quic"]
);
restarted.ready(other_id, 1).await;
b.ready(other_id, 1).await;
exchange(&restarted, &b, other_id).await;
restarted
.agent
.set_protocols(id, vec!["awg".into()])
.await
.unwrap();
restarted.ready(id, 1).await;
b.ready(id, 1).await;
exchange(&restarted, &b, id).await;
restarted.agent.shutdown().await;
b.agent.shutdown().await;
}
#[tokio::test]
async fn real_pairs_carry_full_size_packets_and_restart_with_fresh_profiles() {
tsunagi::testing::init_tracing();
+171 -29
View File
@@ -58,6 +58,8 @@ use shutdown::Shutdown;
/// Summary of a configured network, whether or not it is running.
#[derive(Debug, Clone)]
pub struct ConfiguredNetwork {
/// Enabled local data-plane protocols, in preference order.
pub protocols: Vec<String>,
/// Saved local broadcast participation.
pub broadcast: bool,
/// Public network identifier.
@@ -70,6 +72,17 @@ pub struct ConfiguredNetwork {
pub active: bool,
}
/// A durable version-scoped cleanup, not an identity ban.
#[derive(Debug, Clone)]
pub struct ForgetNodeOutcome {
/// Identity whose observed claim was retired.
pub target: EndpointId,
/// The retired claim version; future signed claims remain admissible.
pub record_version: u64,
/// Authenticated peers to which cleanup was queued immediately.
pub peers_told: usize,
}
/// A running agent.
///
/// Cloning gives another handle to the same agent.
@@ -104,6 +117,37 @@ struct Inner {
}
impl Inner {
/// Called under the lifecycle lock after startup: an initially
/// control-only agent creates its one interface on first enablement.
async fn ensure_interface(self: &Arc<Self>) {
if self.interface.get().is_some() {
return;
}
if let Some(factory) = self.config.tun_factory.clone() {
let carrier = Arc::new(Carrier(Arc::downgrade(self))) as Arc<dyn PacketCarrier>;
match crate::overlay::Interface::start(
factory,
self.config.interface_name.clone(),
self.config.interface_mtu,
Arc::clone(&self.routes),
carrier,
)
.await
{
Ok(interface) => {
let _ = self.interface.set(Arc::new(interface));
}
Err(err) => {
let _ = self.events.send(Event::PluginError {
network: NetworkId::from_bytes([0u8; 32]),
protocol: "overlay".into(),
reason: err.to_string(),
});
}
}
}
}
/// The name this agent currently answers to.
fn read_hostname(&self) -> String {
match self.hostname.read() {
@@ -265,12 +309,111 @@ impl TransportContext for TransportCtx {
}
impl Agent {
/// Retires one offline member's observed claim without revoking membership.
/// Every member has this right. The version must still match the selection.
pub async fn forget_node(
&self,
network: NetworkId,
peer: EndpointId,
expected_version: u64,
) -> Result<ForgetNodeOutcome> {
let _lifecycle = self.inner.lifecycle.lock().await;
let (reply, receive) = oneshot::channel();
self.command(
network,
NetCommand::ForgetNode {
peer,
expected_version,
reply,
},
)
.await?;
receive.await.map_err(|_| Error::Stopped)?
}
fn effective_protocols(&self, saved: Option<&Vec<String>>) -> Vec<String> {
saved
.or(self.inner.config.default_protocols.as_ref())
.cloned()
.unwrap_or_else(|| {
self.inner
.config
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect()
})
}
fn selected_plugins(&self, protocols: &[String]) -> Vec<crate::dataplane::SharedPlugin> {
protocols
.iter()
.filter_map(|name| {
self.inner
.config
.plugins
.iter()
.find(|p| p.protocol_id() == name)
.cloned()
})
.collect()
}
/// Saves and immediately applies a network's ordered local protocol set.
/// An empty set disables data traffic while retaining control sessions.
pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec<String>) -> Result<()> {
crate::dataplane::validate_protocols(&protocols, self.inner.limits.max_capabilities)?;
for name in &protocols {
if !self
.inner
.config
.plugins
.iter()
.any(|p| p.protocol_id() == name)
{
return Err(Error::InvalidProtocols(format!(
"this agent has no `{name}` plugin"
)));
}
}
let _lifecycle = self.inner.lifecycle.lock().await;
self.inner
.storage
.set_protocols(network_id, protocols.clone())
.await?;
if self.is_active(network_id).await {
let (reply, receive) = oneshot::channel();
if !protocols.is_empty() {
self.inner.ensure_interface().await;
}
self.command(
network_id,
NetCommand::SetProtocols {
protocols,
interface: self.inner.interface.get().cloned(),
reply,
},
)
.await?;
receive.await.map_err(|_| Error::Stopped)?;
}
Ok(())
}
/// Starts an agent.
///
/// Opens the state store (taking its ownership lock), restores the
/// persistent device identity, binds the iroh endpoint and activates every
/// configured network whose auto-start flag is set.
pub async fn spawn(config: AgentConfig) -> Result<Self> {
if let Some(protocols) = &config.default_protocols {
crate::dataplane::validate_protocols(protocols, config.limits.max_capabilities)?;
for name in protocols {
if !config.plugins.iter().any(|p| p.protocol_id() == name) {
return Err(Error::InvalidProtocols(format!(
"this agent has no `{name}` plugin"
)));
}
}
}
let storage = Storage::open(&config.paths)?;
let identity = storage.device_identity().await?;
let adapter = EndpointAdapter::bind(&config, &identity).await?;
@@ -303,31 +446,13 @@ impl Agent {
config,
});
// One interface for the agent, if it was given a way to make one.
// Failing to is reported and not fatal: the control plane works, and
// so do the protocols, they just have nowhere to put packets.
if let Some(factory) = inner.config.tun_factory.clone() {
let carrier = Arc::new(Carrier(Arc::downgrade(&inner))) as Arc<dyn PacketCarrier>;
match crate::overlay::Interface::start(
factory,
inner.config.interface_name.clone(),
inner.config.interface_mtu,
Arc::clone(&inner.routes),
carrier,
)
.await
{
Ok(interface) => {
let _ = inner.interface.set(Arc::new(interface));
}
Err(err) => {
let _ = inner.events.send(Event::PluginError {
network: NetworkId::from_bytes([0u8; 32]),
protocol: "overlay".into(),
reason: err.to_string(),
});
}
}
if inner
.config
.default_protocols
.as_ref()
.is_none_or(|names| !names.is_empty())
{
inner.ensure_interface().await;
}
// The data plane rides on iroh too, which is where it gets hole
@@ -364,8 +489,9 @@ impl Agent {
if !inner.config.plugins.is_empty() {
let (plugin_tx, plugin_rx) = mpsc::channel(64);
let sink = inner
.interface
.get()
.config
.tun_factory
.as_ref()
.map(|_| Arc::new(Sink(Arc::downgrade(&inner))) as Arc<dyn PacketSink>);
let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink);
for plugin in &inner.config.plugins {
@@ -784,6 +910,12 @@ impl Agent {
.await?
.into_iter()
.find(|stored| stored.network_id == network_id);
// Mandatory cleanup proofs must never be silently dropped on startup.
let retirements = self
.inner
.storage
.retirements(network_id, self.inner.limits.max_retirements)
.await?;
let range = self.desired_range(network_id).await?;
if let Some(range) = range {
let conflicts = self.inner.routes.overlapping_networks(network_id, range);
@@ -824,12 +956,19 @@ impl Agent {
};
let broadcast = stored.as_ref().is_none_or(|stored| stored.broadcast);
let protocols =
self.effective_protocols(stored.as_ref().and_then(|s| s.protocols.as_ref()));
let plugins = self.selected_plugins(&protocols);
if !plugins.is_empty() {
self.inner.ensure_interface().await;
}
let exit_offer = stored.as_ref().is_some_and(|stored| stored.exit_node);
let exit_via = stored
.as_ref()
.and_then(|stored| stored.exit_via)
.and_then(|bytes| EndpointId::from_bytes(&bytes).ok());
let handle = network::spawn(RuntimeParams {
retirements,
broadcast,
exit_offer,
exit_via,
@@ -842,7 +981,8 @@ impl Agent {
discovery,
discovery_interval: self.inner.config.discovery_interval,
discovery_policy: self.inner.config.discovery_policy.clone(),
plugins: self.inner.config.plugins.clone(),
available_plugins: self.inner.config.plugins.clone(),
plugins: plugins.clone(),
routes: Arc::clone(&self.inner.routes),
interface: self.inner.interface.get().cloned(),
hostname: self.inner.read_hostname(),
@@ -856,7 +996,7 @@ impl Agent {
networks.insert(network_id, handle);
drop(networks);
for plugin in &self.inner.config.plugins {
for plugin in &plugins {
plugin.on_network_activated(network_id);
}
@@ -997,6 +1137,7 @@ impl Agent {
.await?
.into_iter()
.map(|stored| ConfiguredNetwork {
protocols: self.effective_protocols(stored.protocols.as_ref()),
broadcast: stored.broadcast,
network_id: stored.network_id,
name: stored.name,
@@ -1075,6 +1216,7 @@ impl Agent {
let range = self.desired_range(stored.network_id).await?;
let range_conflicts = self.range_conflicts(stored.network_id, range).await?;
networks.push(NetworkStatus {
protocols: self.effective_protocols(stored.protocols.as_ref()),
auto_start: stored.auto_start,
broadcast: stored.broadcast,
exit_offer: stored.exit_node,
+390 -25
View File
@@ -26,7 +26,9 @@ use crate::net::{EndpointAdapter, PathAddr, snapshot_connection};
use crate::proto::handshake::{self, HandshakeOutcome, Role};
use crate::proto::message::{Announcement, ControlMessage, Envelope, encode, kind};
use crate::state::allocator::allocate;
use crate::state::{Ipv4Range, Merged, RecordBody, SignedRecord, StateSet};
use crate::state::{
Ipv4Range, MAX_RETIREMENT_LEN, Merged, RecordBody, SignedRecord, SignedRetirement, StateSet,
};
use crate::storage::Storage;
use super::events::Event;
@@ -48,6 +50,11 @@ pub(crate) struct InboundSession {
/// Commands accepted by a network runtime.
pub(crate) enum NetCommand {
ForgetNode {
peer: EndpointId,
expected_version: u64,
reply: oneshot::Sender<Result<super::ForgetNodeOutcome>>,
},
Inbound(Box<InboundSession>),
Send {
peer: EndpointId,
@@ -72,6 +79,11 @@ pub(crate) enum NetCommand {
enabled: bool,
reply: oneshot::Sender<()>,
},
SetProtocols {
interface: Option<Arc<crate::overlay::Interface>>,
protocols: Vec<String>,
reply: oneshot::Sender<()>,
},
/// Offer this agent as an exit node in this network, or stop.
SetExitOffer {
enabled: bool,
@@ -113,6 +125,7 @@ pub(crate) enum NetCommand {
impl std::fmt::Debug for NetCommand {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
NetCommand::ForgetNode { peer, .. } => write!(f, "ForgetNode({peer})"),
NetCommand::Inbound(_) => f.write_str("Inbound"),
NetCommand::InboundLink(link) => write!(f, "InboundLink({})", link.protocol),
NetCommand::Send { peer, message, .. } => {
@@ -122,6 +135,7 @@ impl std::fmt::Debug for NetCommand {
NetCommand::Status { .. } => f.write_str("Status"),
NetCommand::Recheck => f.write_str("Recheck"),
NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"),
NetCommand::SetProtocols { protocols, .. } => write!(f, "SetProtocols({protocols:?})"),
NetCommand::SetExitOffer { enabled, .. } => write!(f, "SetExitOffer({enabled})"),
NetCommand::SetExitVia { via, .. } => write!(f, "SetExitVia({})", via.is_some()),
NetCommand::Reannounce => f.write_str("Reannounce"),
@@ -156,6 +170,7 @@ impl NetworkHandle {
/// Everything a network runtime needs to run.
pub(crate) struct RuntimeParams {
pub(crate) retirements: Vec<SignedRetirement>,
pub(crate) broadcast: bool,
/// This agent offers itself as an exit node here.
pub(crate) exit_offer: bool,
@@ -171,6 +186,7 @@ pub(crate) struct RuntimeParams {
pub(crate) discovery_interval: Duration,
pub(crate) discovery_policy: DiscoveryPolicy,
pub(crate) plugins: Vec<SharedPlugin>,
pub(crate) available_plugins: Vec<SharedPlugin>,
/// Who holds which overlay address, shared with every other network.
pub(crate) routes: Arc<crate::overlay::RoutingTable>,
/// The one interface, when the agent has one.
@@ -192,6 +208,7 @@ pub(crate) struct RuntimeParams {
/// Outcome of one attempt to open a data plane link.
struct LinkOutcome {
generation: u64,
peer: EndpointId,
protocol: String,
result: Result<SharedLink, String>,
@@ -332,6 +349,9 @@ struct Runtime {
announced_reach: HashSet<crate::proto::message::ReachableLink>,
/// Links currently being opened, so we do not start two.
opening: HashSet<(EndpointId, String)>,
link_tasks: HashMap<(EndpointId, String), tokio::task::AbortHandle>,
link_attempts: HashMap<(EndpointId, String), u64>,
next_link_attempt: u64,
link_results_tx: mpsc::Sender<LinkOutcome>,
link_results_rx: mpsc::Receiver<LinkOutcome>,
/// Protocol dial failure count and backoff cooldown per peer and protocol.
@@ -355,6 +375,9 @@ struct Runtime {
state: StateSet,
/// Snapshots received while dispatching, handled on the next loop pass.
pending_state: Vec<(EndpointId, Vec<SignedRecord>)>,
pending_retirements: Vec<(EndpointId, Vec<SignedRetirement>)>,
/// Next cleanup batch per live session; prevents queue pressure starving a suffix.
retirement_cursor: HashMap<EndpointId, usize>,
/// The highest version this agent has ever published for this network.
own_version: u64,
/// Whether the exit node in use was reachable when last looked at, so
@@ -425,6 +448,9 @@ impl Runtime {
local_capabilities: Vec::new(),
announced_reach: HashSet::new(),
opening: HashSet::new(),
link_tasks: HashMap::new(),
link_attempts: HashMap::new(),
next_link_attempt: 0,
link_results_tx,
link_results_rx,
protocol_dial_states: HashMap::new(),
@@ -434,6 +460,8 @@ impl Runtime {
reported_name_taken: None,
state: StateSet::new(),
pending_state: Vec::new(),
pending_retirements: Vec::new(),
retirement_cursor: HashMap::new(),
own_version: 0,
exit_via_was_online: false,
announced_exit: false,
@@ -511,6 +539,10 @@ impl Runtime {
link.close();
}
self.links.clear();
for (_, task) in self.link_tasks.drain() {
task.abort();
}
self.opening.clear();
self.pending_health.clear();
self.last_failed_protocol.clear();
self.protocol_dial_states.clear();
@@ -529,6 +561,14 @@ impl Runtime {
async fn handle_command(&mut self, command: NetCommand) {
match command {
NetCommand::ForgetNode {
peer,
expected_version,
reply,
} => {
let result = self.forget_node(peer, expected_version).await;
let _ = reply.send(result);
}
NetCommand::Inbound(inbound) => {
self.install_session(*inbound).await;
}
@@ -553,6 +593,60 @@ impl Runtime {
NetCommand::Status { reply } => {
let _ = reply.send(Box::new(self.status()));
}
NetCommand::SetProtocols {
protocols,
interface,
reply,
} => {
self.params.interface = interface;
let current: Vec<_> = self
.params
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect();
if current != protocols {
let peers: Vec<_> = self.sessions.keys().copied().collect();
for peer in &peers {
self.reset_peer_data(*peer, "local protocol selection changed");
}
for plugin in &self.params.plugins {
if !protocols.iter().any(|name| name == plugin.protocol_id()) {
plugin.on_network_deactivated(self.network_id);
}
}
self.params.plugins = protocols
.iter()
.filter_map(|name| {
self.params
.available_plugins
.iter()
.find(|p| p.protocol_id() == name)
.cloned()
})
.collect();
for plugin in &self.params.plugins {
if !current.iter().any(|name| name == plugin.protocol_id()) {
plugin.on_network_activated(self.network_id);
}
}
self.hub.set_priority(protocols);
self.publish_allocations().await;
for peer in peers {
let caps = self
.sessions
.get(&peer)
.map(|s| s.capabilities.clone())
.unwrap_or_default();
self.dispatch_capabilities(peer, &caps);
}
self.reannounce();
self.ensure_links();
self.update_paths();
self.announce_reach(true);
}
let _ = reply.send(());
}
NetCommand::SetBroadcast { enabled, reply } => {
self.params.broadcast = enabled;
self.update_broadcast();
@@ -584,7 +678,22 @@ impl Runtime {
let _ = reply.send(outcome);
}
NetCommand::Recheck => self.discovery_round().await,
NetCommand::Reannounce => self.reannounce(),
NetCommand::Reannounce => {
// A newly enabled plugin may prepare its key asynchronously.
// Replay allocations and peer identities once it is ready.
self.publish_allocations().await;
let peers: Vec<_> = self
.sessions
.iter()
.map(|(peer, s)| (*peer, s.capabilities.clone()))
.collect();
for (peer, caps) in peers {
self.dispatch_capabilities(peer, &caps);
}
self.reannounce();
self.ensure_links();
self.update_paths();
}
NetCommand::Release { reply } => {
// A tombstone, so the address and the name this agent held
// are freed for somebody else instead of staying reserved
@@ -714,6 +823,7 @@ impl Runtime {
// the check is a comparison when nothing has changed.
self.ensure_own_claim().await;
self.report_hostname_taken();
self.broadcast_retirements();
// The same timer repairs the relay picture: reachability is soft
// state, repeated while it holds and forgotten when it stops.
self.update_paths();
@@ -727,7 +837,7 @@ impl Runtime {
// still dialable when the endpoint's own discovery can resolve
// one. This is what makes a network a mesh rather than a star
// around whoever was named on a command line.
for record in self.state.records() {
for record in self.state.active_records() {
if let Ok(author) = record.author_id()
&& author != self.local_id
{
@@ -744,6 +854,9 @@ impl Runtime {
let Ok(endpoint_id) = EndpointId::from_bytes(&hint.endpoint_id) else {
continue;
};
if self.state.is_retired(&endpoint_id) {
continue;
}
let Some(addr) = decode_hint(endpoint_id, &hint.addr) else {
continue;
};
@@ -949,6 +1062,22 @@ impl Runtime {
/// boundary, because a restored backup or a copied file could hold
/// anything.
async fn load_state(&mut self) {
for marker in std::mem::take(&mut self.params.retirements) {
// Storage verified these proofs and activation enforced capacity.
if let Err(err) = self.state.merge_retirement(
self.network_id,
marker,
self.params.limits.max_retirements,
) {
self.emit(Event::PluginError {
network: self.network_id,
protocol: "state".into(),
reason: err.to_string(),
});
self.shutdown.trigger();
return;
}
}
let stored = self
.params
.storage
@@ -964,7 +1093,8 @@ impl Runtime {
.storage
.own_record_version(self.network_id, self.local_id)
.await
.unwrap_or(0);
.unwrap_or(0)
.max(self.state.authored_version(&self.local_id));
for (id, hostname) in self
.params
.storage
@@ -1076,6 +1206,7 @@ impl Runtime {
let kept = self
.state
.get(&self.local_id)
.filter(|_| !self.state.is_retired(&self.local_id))
.filter(|record| record.body.hostname() == Some(name.as_str()))
.and_then(|record| record.body.hostname_since());
kept.unwrap_or_else(unix_millis)
@@ -1108,7 +1239,7 @@ impl Runtime {
if self
.state
.get(&self.local_id)
.is_some_and(|record| record.body == body)
.is_some_and(|record| record.body == body && !self.state.is_retired(&self.local_id))
{
return;
}
@@ -1116,13 +1247,187 @@ impl Runtime {
self.publish_record(body).await;
}
async fn forget_node(
&mut self,
peer: EndpointId,
expected_version: u64,
) -> Result<super::ForgetNodeOutcome> {
if peer == self.local_id {
return Err(Error::ForgetNode(
"use network leave for this device".into(),
));
}
if self.is_online(peer) {
return Err(Error::ForgetNode(
"the node is online; only offline history can be forgotten".into(),
));
}
if self.params.exit_via == Some(peer) {
return Err(Error::ForgetNode(
"stop using this exit node before forgetting it".into(),
));
}
let target = self
.state
.get(&peer)
.filter(|_| !self.state.is_retired(&peer))
.filter(|record| matches!(record.body, RecordBody::Claim { .. }))
.cloned()
.ok_or_else(|| Error::ForgetNode("no current signed claim for this node".into()))?;
if target.version != expected_version {
return Err(Error::ForgetNode(
"the node's claim changed; refresh and select it again".into(),
));
}
let version = self
.own_version
.max(self.state.authored_version(&self.local_id))
.checked_add(1)
.filter(|v| *v <= i64::MAX as u64)
.ok_or_else(|| Error::ForgetNode("author counter exhausted".into()))?;
let marker =
SignedRetirement::sign(&self.params.device_secret, self.network_id, version, target);
let mut candidate = self.state.clone();
candidate
.merge_retirement(
self.network_id,
marker.clone(),
self.params.limits.max_retirements,
)
.map_err(|err| Error::ForgetNode(err.to_string()))?;
// Signed cleanup and our author counter commit together before visibility.
self.params
.storage
.publish_own_retirement(marker.clone())
.await?;
self.own_version = version;
self.state = candidate;
self.known_hostnames.remove(&peer);
self.dial_states.remove(&peer);
self.candidate_addrs.remove(&peer);
self.publish_allocations().await;
self.report_hostname_taken();
let message = ControlMessage::Retirements {
markers: vec![marker],
};
let peers: Vec<_> = self.sessions.keys().copied().collect();
let peers_told = peers
.into_iter()
.filter(|peer| self.send_to(*peer, message.clone()).is_ok())
.count();
Ok(super::ForgetNodeOutcome {
target: peer,
record_version: expected_version,
peers_told,
})
}
fn broadcast_retirements(&mut self) -> usize {
let markers = self.state.retirements();
if markers.is_empty() {
return 0;
}
// Each proof has an encoded bound; reserve envelope/list overhead and
// chunk every marker, so a low-id prefix cannot starve later cleanup.
let batch = (self.params.limits.max_frame_len.saturating_sub(128) / MAX_RETIREMENT_LEN)
.max(1)
.min(self.params.limits.max_state_records.max(1));
let mut delivered = HashSet::new();
let peers: Vec<_> = self.sessions.keys().copied().collect();
let chunks: Vec<_> = markers.chunks(batch).collect();
for peer in peers {
let start = self.retirement_cursor.get(&peer).copied().unwrap_or(0) % chunks.len();
for offset in 0..chunks.len() {
let index = (start + offset) % chunks.len();
let message = ControlMessage::Retirements {
markers: chunks[index].to_vec(),
};
if self.send_to(peer, message).is_err() {
self.retirement_cursor.insert(peer, index);
break;
}
delivered.insert(peer);
self.retirement_cursor
.insert(peer, (index + 1) % chunks.len());
}
}
delivered.len()
}
async fn receive_retirements(&mut self, peer: EndpointId, markers: Vec<SignedRetirement>) {
let mut changed = false;
for marker in markers {
let mut candidate = self.state.clone();
match candidate.merge_retirement(
self.network_id,
marker.clone(),
self.params.limits.max_retirements,
) {
Ok(false) => continue,
Ok(true) => {}
Err(err) => {
self.metrics.protocol_violations += 1;
self.emit(Event::ProtocolViolation {
network: Some(self.network_id),
peer: Some(peer),
reason: format!("unusable cleanup proof: {err}"),
});
continue;
}
}
let target = marker.target.author_id().ok();
if let Err(err) = self.params.storage.put_retirement(marker).await {
self.emit(Event::PluginError {
network: self.network_id,
protocol: "state".into(),
reason: format!("cannot persist cleanup proof: {err}"),
});
continue;
}
self.state = candidate;
if let Some(target) = target
&& self.state.is_retired(&target)
&& !self.sessions.contains_key(&target)
{
self.known_hostnames.remove(&target);
self.dial_states.remove(&target);
self.candidate_addrs.remove(&target);
}
changed = true;
}
if changed {
// A live target answers cleanup with a fresh claim; its identity
// is still authenticated, and old hostname priority is not reused.
self.ensure_own_claim().await;
self.publish_allocations().await;
self.broadcast_retirements();
}
}
/// Signs, stores and announces one of this agent's own records.
///
/// Stored before it is announced, in one transaction with the version
/// counter, so a crash can never let us reuse a version we already put on
/// the wire.
async fn publish_record(&mut self, body: RecordBody) {
let version = self.own_version.saturating_add(1);
let observed = self.state.authored_version(&self.local_id).max(
self.state
.retirement_of(&self.local_id)
.map_or(0, |m| m.target.version),
);
let Some(version) = self
.own_version
.max(observed)
.checked_add(1)
.filter(|v| *v <= i64::MAX as u64)
else {
self.emit(Event::PluginError {
network: self.network_id,
protocol: "state".into(),
reason: "author counter exhausted".into(),
});
return;
};
let record = SignedRecord::sign(&self.params.device_secret, self.network_id, version, body);
if let Err(err) = self.params.storage.publish_own_record(record.clone()).await {
@@ -1147,6 +1452,9 @@ impl Runtime {
/// Handles snapshots collected while dispatching messages.
async fn drain_pending_state(&mut self) {
for (peer, markers) in std::mem::take(&mut self.pending_retirements) {
self.receive_retirements(peer, markers).await;
}
for (peer, records) in std::mem::take(&mut self.pending_state) {
self.receive_state(peer, records).await;
}
@@ -1155,6 +1463,7 @@ impl Runtime {
/// Sends everything we know to every peer.
fn broadcast_state(&mut self) {
self.broadcast_retirements();
let mut records = self.state.records();
records.truncate(self.params.limits.max_state_records);
if records.is_empty() {
@@ -1578,13 +1887,16 @@ impl Runtime {
if self.links.contains_key(&key) || self.opening.contains(&key) {
continue;
}
self.opening.insert(key);
self.opening.insert(key.clone());
let results = self.link_results_tx.clone();
let transport = Arc::clone(&transport);
let network = self.network_id;
let shutdown = self.shutdown.clone();
tokio::spawn(async move {
self.next_link_attempt = self.next_link_attempt.wrapping_add(1);
let generation = self.next_link_attempt;
self.link_attempts.insert(key.clone(), generation);
let task = tokio::spawn(async move {
let result = tokio::select! {
biased;
_ = shutdown.wait() => Err("network deactivated".to_string()),
@@ -1594,12 +1906,14 @@ impl Runtime {
};
let _ = results
.send(LinkOutcome {
generation,
peer,
protocol,
result,
})
.await;
});
self.link_tasks.insert(key, task.abort_handle());
}
}
@@ -1834,7 +2148,15 @@ impl Runtime {
fn handle_link_result(&mut self, outcome: LinkOutcome) {
let key = (outcome.peer, outcome.protocol.clone());
if self.link_attempts.get(&key) != Some(&outcome.generation) {
if let Ok(link) = outcome.result {
link.close();
}
return;
}
self.link_attempts.remove(&key);
self.opening.remove(&key);
self.link_tasks.remove(&key);
match outcome.result {
Ok(link) => {
self.record_protocol_success(outcome.peer, &outcome.protocol);
@@ -1903,12 +2225,17 @@ impl Runtime {
link.close();
return;
}
if !self
.params
.plugins
.iter()
.any(|plugin| plugin.protocol_id() == protocol)
{
if !self.params.plugins.iter().any(|plugin| {
plugin.protocol_id() == protocol
&& self.sessions.get(&peer).is_some_and(|s| {
s.capabilities.iter().any(|c| {
c.enabled
&& c.protocol == protocol
&& c.version == plugin.protocol_version()
})
})
}) {
link.close();
return;
}
@@ -1966,14 +2293,26 @@ impl Runtime {
/// Drops every link to a peer.
fn drop_links_for(&mut self, peer: EndpointId) {
// The peer itself is going, not just a path to it, so the
// protocol's link goes with it.
self.retirement_cursor.remove(&peer);
self.reset_peer_data(peer, "peer session ended");
self.hub.remove_peer(peer);
self.reachable.remove(&peer);
self.protocol_dial_states.retain(|(p, _), _| *p != peer);
self.last_failed_protocol.remove(&peer);
self.pending_health.retain(|(p, _), _| *p != peer);
let keys: Vec<(EndpointId, String)> = self
}
/// Retire only the data plane; authenticated membership and topology remain.
fn reset_peer_data(&mut self, peer: EndpointId, reason: &str) {
self.link_attempts.retain(|(id, _), _| *id != peer);
self.opening.retain(|(id, _)| *id != peer);
self.link_tasks.retain(|(id, _), task| {
if *id == peer {
task.abort();
false
} else {
true
}
});
self.hub.clear_peer_links(peer);
let keys: Vec<_> = self
.links
.keys()
.filter(|(id, _)| *id == peer)
@@ -1987,9 +2326,15 @@ impl Runtime {
network: self.network_id,
peer,
protocol: key.1,
reason: "peer session ended".into(),
reason: reason.into(),
});
}
self.protocol_dial_states.retain(|(id, _), _| *id != peer);
self.last_failed_protocol.remove(&peer);
self.pending_health.retain(|(id, _), _| *id != peer);
for plugin in &self.params.plugins {
plugin.on_peer_gone(self.network_id, peer);
}
}
// --------------------------------------------------------------- sessions
@@ -2182,9 +2527,6 @@ impl Runtime {
self.watch_exit_via();
self.update_paths();
self.announce_reach(false);
for plugin in &self.params.plugins {
plugin.on_peer_gone(self.network_id, peer);
}
// Retry promptly, then back off if it keeps failing.
let policy = &self.params.reconnect;
let state = self
@@ -2277,6 +2619,18 @@ impl Runtime {
}
ControlMessage::Announce(announcement) => {
let capabilities = announcement.capabilities.clone();
let changed = self.sessions.get(&peer).is_some_and(|session| {
let selection = |caps: &[PluginCapability]| {
caps.iter()
.filter(|c| c.enabled)
.map(|c| (c.protocol.clone(), c.version))
.collect::<Vec<_>>()
};
selection(&session.capabilities) != selection(&capabilities)
});
if changed {
self.reset_peer_data(peer, "peer protocol selection changed");
}
if let Some(session) = self.sessions.get_mut(&peer) {
session.hostname = Some(announcement.hostname.clone());
session.broadcast = announcement.broadcast;
@@ -2289,6 +2643,7 @@ impl Runtime {
self.dispatch_capabilities(peer, &capabilities);
self.ensure_links();
self.update_paths();
self.announce_reach(false);
}
ControlMessage::Ping { seq, payload } => {
let pong = ControlMessage::Pong {
@@ -2302,6 +2657,9 @@ impl Runtime {
ControlMessage::State { records } => {
self.pending_state.push((peer, records.clone()));
}
ControlMessage::Retirements { markers } => {
self.pending_retirements.push((peer, markers.clone()));
}
// Members somebody else knows of. Candidates, nothing more:
// each still has to pass the handshake, and each is tried the
// same way as one from any other source.
@@ -2499,12 +2857,13 @@ impl Runtime {
// listing it as one turns leaving into a peer that looks broken.
let mut members: Vec<MemberStatus> = self
.state
.records()
.active_records()
.into_iter()
.filter(|record| !matches!(record.body, RecordBody::Release))
.filter_map(|record| {
let endpoint_id = record.author_id().ok()?;
Some(MemberStatus {
record_version: record.version,
endpoint_id,
// Only uncontested claims are reported as held: two
// members may have claimed the same thing, and saying
@@ -2522,6 +2881,12 @@ impl Runtime {
members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id);
NetworkStatus {
protocols: self
.params
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect(),
auto_start: true,
broadcast: self.params.broadcast,
exit_offer: self.params.exit_offer,
+4
View File
@@ -108,6 +108,8 @@ pub struct HostnameConflict {
/// connected.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberStatus {
/// Current signed claim version, used to bind cleanup confirmation.
pub record_version: u64,
/// The member's device identity.
pub endpoint_id: EndpointId,
/// The IPv4 overlay address it claimed and signed for.
@@ -223,6 +225,8 @@ pub struct RangeConflict {
/// Status of one network.
#[derive(Debug, Clone)]
pub struct NetworkStatus {
/// Enabled local data-plane protocols, in preference order.
pub protocols: Vec<String>,
/// Whether this network is requested to start or resume automatically.
pub auto_start: bool,
/// Local broadcast participation in this network.
+7
View File
@@ -174,6 +174,8 @@ pub struct Limits {
pub max_reason_len: usize,
/// Largest number of signed records accepted in one snapshot.
pub max_state_records: usize,
/// Maximum durable retirement markers retained per network.
pub max_retirements: usize,
/// Deadline for the whole handshake.
pub handshake_timeout: Duration,
/// Deadline for one outbound dial attempt.
@@ -211,6 +213,7 @@ impl Default for Limits {
max_echo_payload_len: 4 * 1024,
max_reason_len: 256,
max_state_records: crate::state::MAX_RECORDS_PER_MESSAGE,
max_retirements: 1024,
handshake_timeout: Duration::from_secs(10),
dial_timeout: Duration::from_secs(10),
write_timeout: Duration::from_secs(30),
@@ -328,6 +331,9 @@ pub struct AgentConfig {
pub reconnect: ReconnectPolicy,
/// IP plugins whose capabilities are announced and dispatched.
pub plugins: Vec<SharedPlugin>,
/// Initial ordered selection for networks without a saved override.
/// `None` enables all registered plugins; an empty list is control-only.
pub default_protocols: Option<Vec<String>>,
/// Custom packet transports registered per protocol.
pub custom_transports:
std::collections::HashMap<String, Arc<dyn crate::dataplane::PacketTransport>>,
@@ -376,6 +382,7 @@ impl AgentConfig {
limits: Limits::default(),
reconnect: ReconnectPolicy::default(),
plugins: Vec::new(),
default_protocols: None,
custom_transports: std::collections::HashMap::new(),
tun_factory: None,
interface_name: "tsun0".to_string(),
+27
View File
@@ -40,6 +40,33 @@ pub use tunnel::{GenericTunnelPlugin, TunnelCodec};
/// Maximum length of a plugin protocol identifier.
pub const MAX_PROTOCOL_ID_LEN: usize = 32;
/// Validates bounded local protocol preferences before storing or applying them.
pub(crate) fn validate_protocols(protocols: &[String], max: usize) -> crate::error::Result<()> {
if protocols.len() > max {
return Err(crate::Error::InvalidProtocols(format!(
"at most {max} protocols are allowed"
)));
}
for (index, name) in protocols.iter().enumerate() {
if name.is_empty()
|| name.len() > MAX_PROTOCOL_ID_LEN
|| !name
.bytes()
.all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
{
return Err(crate::Error::InvalidProtocols(
"names must be 1–32 ASCII letters, digits, hyphens or underscores".into(),
));
}
if protocols[..index].contains(name) {
return Err(crate::Error::InvalidProtocols(format!(
"`{name}` is listed more than once"
)));
}
}
Ok(())
}
/// An announcement of one IP plugin's capability.
///
/// `data` is opaque to the core. Nothing in it may be interpreted as a shell
+10 -2
View File
@@ -492,11 +492,12 @@ impl RelayHub {
}
}
/// Revokes all links and topology involving a departed authenticated peer.
pub fn remove_peer(&self, peer: EndpointId) {
/// Retires data channels without removing an authenticated peer's topology.
pub fn clear_peer_links(&self, peer: EndpointId) {
let mut state = self.state();
if let Some(links) = state.raw.remove(peer.as_bytes()) {
for raw in links {
raw.link.close();
raw.reader.abort();
}
}
@@ -511,6 +512,13 @@ impl RelayHub {
link.close();
}
}
state.publish();
}
/// Revokes all links and topology involving a departed authenticated peer.
pub fn remove_peer(&self, peer: EndpointId) {
self.clear_peer_links(peer);
let mut state = self.state();
state.members.remove(peer.as_bytes());
state.graph.remove(peer.as_bytes());
for neighbors in state.graph.values_mut() {
+6
View File
@@ -15,6 +15,12 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum Error {
/// A node-history cleanup is unsafe, stale or cannot be performed.
#[error("cannot forget node: {0}")]
ForgetNode(String),
/// An ordered local data-plane protocol selection is invalid.
#[error("invalid protocol selection: {0}")]
InvalidProtocols(String),
/// The supplied network name does not satisfy the documented rules.
#[error("invalid network name: {0}")]
InvalidNetworkName(&'static str),
+236 -2
View File
@@ -171,6 +171,22 @@ pub fn default_control_socket_path() -> PathBuf {
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum Request {
/// Retire the selected offline member's observed claim.
ForgetNode {
/// Exact public network id.
network_id: String,
/// Exact device identity selected by the client.
peer: String,
/// Claim version displayed when the user selected the node.
expected_version: u64,
},
/// Replace one network's ordered local data-plane selection.
SetProtocols {
/// Public network identifier.
network_id: String,
/// Protocol names, best first; empty selects control only.
protocols: Vec<String>,
},
/// Report what the agent is doing.
Status,
/// Set local broadcast participation in one configured network.
@@ -255,6 +271,21 @@ pub enum Request {
impl std::fmt::Debug for Request {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Request::ForgetNode {
network_id,
peer,
expected_version,
} => write!(
f,
"ForgetNode {{ {network_id}, peer: {peer}, version: {expected_version} }}"
),
Request::SetProtocols {
network_id,
protocols,
} => write!(
f,
"SetProtocols {{ {network_id}, protocols: {protocols:?} }}"
),
Request::Status => f.write_str("Status"),
Request::SetBroadcast {
network_id,
@@ -286,6 +317,10 @@ impl std::fmt::Debug for Request {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum Response {
/// An observed offline claim was durably retired.
NodeForgotten(ForgottenNodeReport),
/// Accepted ordered protocol selection.
Protocols(Vec<String>),
/// A status report.
Status(Box<StatusReport>),
/// Accepted local broadcast participation.
@@ -312,6 +347,141 @@ pub enum Response {
Error(String),
}
/// The durable result of an offline-node cleanup.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ForgottenNodeReport {
/// Selected device identity.
pub peer: String,
/// The retired claim version, not a future-version ban.
pub record_version: u64,
/// Peers to which cleanup was immediately queued.
pub peers_told: u32,
}
/// Resolves a node by exact hostname, full id, unique id prefix or overlay IPv4.
/// Ambiguity is reported with names and full ids and never guessed.
pub fn resolve_node<'a>(
network: &'a NetworkReport,
selector: &str,
) -> std::result::Result<&'a MemberReport, String> {
let selector = selector.trim();
if selector.is_empty() {
return Err("select a node by name, id or IPv4 address".into());
}
if let Some(member) = network.members.iter().find(|m| m.endpoint_id == selector) {
return Ok(member);
}
let prefix = selector.trim_end_matches('…');
let matches: Vec<_> = network
.members
.iter()
.filter(|m| {
m.hostname.as_deref() == Some(selector)
|| m.overlay_address_v4.as_deref() == Some(selector)
|| (!prefix.is_empty() && m.endpoint_id.starts_with(prefix))
})
.collect();
match matches.as_slice() {
[one] => Ok(one),
[] => Err(format!(
"no signed node matches `{selector}` in `{}`",
network.name
)),
many => Err(format!(
"`{selector}` is ambiguous; use a full id or unique id prefix:\n{}",
many.iter()
.map(|m| format!(
" {} {} {}",
m.hostname.as_deref().unwrap_or("unnamed"),
m.endpoint_id,
m.overlay_address_v4.as_deref().unwrap_or("no IPv4")
))
.collect::<Vec<_>>()
.join("\n")
)),
}
}
/// Retires an exact selected claim; an online or changed target is rejected.
pub async fn forget_node(
socket: &Path,
network: &str,
peer: &str,
expected_version: u64,
) -> Result<ForgottenNodeReport> {
let request = Request::ForgetNode {
network_id: network.into(),
peer: peer.into(),
expected_version,
};
match exchange(socket, &request, EXCHANGE_TIMEOUT).await? {
Response::NodeForgotten(report) => Ok(report),
Response::Error(reason) => Err(Error::ForgetNode(reason)),
other => Err(Error::Storage(format!("unexpected response: {other:?}"))),
}
}
#[cfg(test)]
mod node_selector_tests {
use super::*;
#[test]
fn exact_names_addresses_and_unique_prefixes_select_the_same_signed_node() {
let node = MemberReport {
endpoint_id: "abcdef01".into(),
hostname: Some("laptop".into()),
overlay_address_v4: Some("10.2.0.3".into()),
record_version: 42,
..Default::default()
};
let network = NetworkReport {
name: "home".into(),
members: vec![node.clone()],
..Default::default()
};
for selector in ["laptop", "10.2.0.3", "abcdef01", "abc…"] {
assert_eq!(resolve_node(&network, selector), Ok(&node));
}
for selector in ["", "LAPTOP", "missing"] {
assert!(resolve_node(&network, selector).is_err());
}
}
#[test]
fn collisions_between_names_addresses_and_prefixes_require_an_explicit_identity() {
let first = MemberReport {
endpoint_id: "abcdef01".into(),
hostname: Some("duplicate".into()),
overlay_address_v4: Some("10.2.0.3".into()),
..Default::default()
};
let second = MemberReport {
endpoint_id: "abcdef02".into(),
hostname: Some("duplicate".into()),
overlay_address_v4: Some("10.2.0.3".into()),
..Default::default()
};
let third = MemberReport {
endpoint_id: "12345678".into(),
hostname: Some("abc".into()),
..Default::default()
};
let network = NetworkReport {
name: "home".into(),
members: vec![first.clone(), second, third],
..Default::default()
};
for selector in ["duplicate", "10.2.0.3", "abc"] {
let result = resolve_node(&network, selector);
assert!(result.is_err_and(|error| error.contains("ambiguous")
&& error.contains("abcdef01")
&& error.contains("abcdef02")
&& error.contains("duplicate")));
}
assert_eq!(resolve_node(&network, "abcdef01"), Ok(&first));
}
}
/// What happened when a network was stopped or started.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActiveReport {
@@ -377,6 +547,8 @@ pub struct LeftReport {
/// Everything the agent is doing, in one snapshot.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct StatusReport {
/// Data-plane protocols this agent can enable at runtime.
pub available_protocols: Vec<ProtocolReport>,
/// This device's persistent endpoint id.
pub endpoint_id: String,
/// Hostname announced to peers.
@@ -458,9 +630,20 @@ pub struct RangeConflictReport {
pub range: String,
}
/// A data-plane protocol available in the running agent.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtocolReport {
/// Stable protocol name used by the CLI and network negotiation.
pub name: String,
/// A short description of encryption and transport.
pub summary: String,
}
/// One network.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct NetworkReport {
/// Enabled local data-plane protocols, in preference order.
pub protocols: Vec<String>,
/// Whether the network starts or resumes automatically when its range is free.
#[serde(default)]
pub auto_start: bool,
@@ -551,6 +734,8 @@ pub struct HostnameConflictReport {
/// One member of the network, from signed state.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemberReport {
/// Signed claim version used to reject stale cleanup requests.
pub record_version: u64,
/// The member's device identity.
pub endpoint_id: String,
/// The IPv4 overlay address it claimed and signed for.
@@ -720,6 +905,23 @@ impl OverlayPeerReport {
/// running and what they can report or change. That is what keeps this module
/// free of any knowledge of them.
pub trait ReportSource: Send + Sync + 'static {
/// Persists and announces a version-scoped node cleanup.
fn forget_node(
&self,
_network: String,
_peer: String,
_expected_version: u64,
) -> BoxFuture<'_, std::result::Result<ForgottenNodeReport, String>> {
Box::pin(async { Err("this agent cannot forget node history".into()) })
}
/// Saves and applies a local ordered protocol set for one network.
fn set_protocols(
&self,
_network_id: String,
_protocols: Vec<String>,
) -> BoxFuture<'_, std::result::Result<Vec<String>, String>> {
Box::pin(async { Err("this agent cannot change protocols".into()) })
}
/// Produces a fresh report.
fn report(&self) -> BoxFuture<'_, StatusReport>;
@@ -849,7 +1051,7 @@ pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5);
///
/// Bump it whenever [`Request`], [`Response`] or anything they contain
/// changes shape.
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 19]);
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 21]);
/// Reads one request off an accepted stream, answers it, writes the response.
///
@@ -873,6 +1075,21 @@ where
}
};
let response = match request {
Request::ForgetNode {
network_id,
peer,
expected_version,
} => match source.forget_node(network_id, peer, expected_version).await {
Ok(report) => Response::NodeForgotten(report),
Err(reason) => Response::Error(reason),
},
Request::SetProtocols {
network_id,
protocols,
} => match source.set_protocols(network_id, protocols).await {
Ok(protocols) => Response::Protocols(protocols),
Err(reason) => Response::Error(reason),
},
Request::Status => Response::Status(Box::new(source.report().await)),
Request::SetBroadcast {
network_id,
@@ -1028,7 +1245,24 @@ pub async fn join_network_with_broadcast(
}
}
/// Changes one network's broadcast participation while the agent runs.
/// Replaces the enabled data-plane protocols of one configured network.
pub async fn set_protocols(
socket: &Path,
network_id: &str,
protocols: Vec<String>,
) -> Result<Vec<String>> {
let request = Request::SetProtocols {
network_id: network_id.to_owned(),
protocols,
};
match exchange(socket, &request, EXCHANGE_TIMEOUT).await? {
Response::Protocols(protocols) => Ok(protocols),
Response::Error(reason) => Err(Error::Storage(reason)),
other => Err(Error::Storage(format!("unexpected response: {other:?}"))),
}
}
/// Changes local broadcast participation in a network.
pub async fn set_broadcast(
path: impl AsRef<Path>,
network_id: &str,
+30 -2
View File
@@ -20,7 +20,7 @@ use crate::error::ProtocolError;
/// The version in the ALPN is the wire-compatibility version of the control
/// protocol. It is independent of the network identity scheme version, so
/// bumping it must not change any existing [`crate::NetworkId`].
pub const ALPN: &[u8] = b"tsunagi/ctrl/6";
pub const ALPN: &[u8] = b"tsunagi/ctrl/7";
/// ALPN of the tsunagi data plane.
///
@@ -43,7 +43,7 @@ pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
pub const MAX_SIGNATURE_LEN: usize = 64;
/// Control protocol version carried inside the handshake.
pub const PROTOCOL_VERSION: u16 = 6;
pub const PROTOCOL_VERSION: u16 = 7;
pub use crate::dataplane::transport::{MAX_TRANSPORT_PAYLOAD, TransportPayload};
@@ -181,6 +181,11 @@ pub enum ControlMessage {
/// Short free-text reason.
reason: String,
},
/// Signed cleanup markers, each proving the retired claim version.
Retirements {
/// Bounded by the per-message signed-state limit.
markers: Vec<crate::state::SignedRetirement>,
},
}
/// Somewhere a member has been seen.
@@ -273,6 +278,28 @@ pub fn validate_capability(
/// network, the other networks or the agent.
pub fn validate(message: &ControlMessage, limits: &Limits) -> Result<(), ProtocolError> {
match message {
ControlMessage::Retirements { markers } => {
check_len("retirements", markers.len(), limits.max_state_records)?;
for marker in markers {
check_len(
"retirement.signature",
marker.signature.len(),
MAX_SIGNATURE_LEN,
)?;
check_len(
"retirement.target.signature",
marker.target.signature.len(),
MAX_SIGNATURE_LEN,
)?;
if let Some(name) = marker.target.body.hostname() {
check_len(
"retirement.target.hostname",
name.len(),
limits.max_hostname_len,
)?;
}
}
}
ControlMessage::Transport { protocol, payload } => {
if protocol.is_empty() {
return Err(ProtocolError::Malformed("empty transport protocol"));
@@ -345,6 +372,7 @@ pub fn kind(message: &ControlMessage) -> &'static str {
ControlMessage::Ping { .. } => "ping",
ControlMessage::Pong { .. } => "pong",
ControlMessage::State { .. } => "state",
ControlMessage::Retirements { .. } => "retirements",
ControlMessage::Reachable { .. } => "reachable",
ControlMessage::Peers { .. } => "peers",
ControlMessage::Bye { .. } => "bye",
+84 -2
View File
@@ -28,16 +28,23 @@
//! replica that has the record needs nothing else to interpret it, and
//! recovery never depends on replaying a chain from the beginning.
//!
//! Any member may also sign a separate retirement of another author's proved
//! claim version. It suppresses stale reservations without changing that
//! author's record, and a newer claim remains valid.
//!
//! # What this slice does not do yet
//!
//! Compaction, revocation of a whole author, and snapshots covering more than
//! one record type. See `docs/sync-model.md` for the shape those take.
//! Compaction and permanent revocation of a whole author. See
//! `docs/sync-model.md` for the implemented cleanup semantics.
pub mod allocator;
use std::collections::HashMap;
use std::net::Ipv4Addr;
mod retirement;
pub use retirement::{MAX_RETIREMENT_LEN, SignedRetirement};
use iroh::{EndpointId, SecretKey, Signature};
use serde::{Deserialize, Serialize};
@@ -488,6 +495,7 @@ pub enum Merged {
#[derive(Debug, Clone, Default)]
pub struct StateSet {
records: HashMap<EndpointId, SignedRecord>,
retirements: HashMap<EndpointId, SignedRetirement>,
}
impl StateSet {
@@ -500,6 +508,7 @@ impl StateSet {
pub fn from_verified(records: impl IntoIterator<Item = (EndpointId, SignedRecord)>) -> Self {
Self {
records: records.into_iter().collect(),
retirements: HashMap::new(),
}
}
@@ -528,6 +537,52 @@ impl StateSet {
.collect()
}
/// Visible records, excluding versions retired by authenticated members.
pub fn active_records(&self) -> Vec<SignedRecord> {
self.records()
.into_iter()
.filter(|r| r.author_id().is_ok_and(|id| !self.is_retired(&id)))
.collect()
}
/// Whether the current known version is at or below a proved retirement.
pub fn is_retired(&self, author: &EndpointId) -> bool {
self.retirements
.get(author)
.is_some_and(|m| self.version_of(author) <= m.target.version)
}
/// Strongest retained cleanup proof for an identity.
pub fn retirement_of(&self, target: &EndpointId) -> Option<&SignedRetirement> {
self.retirements.get(target)
}
/// Durable markers in target-id order, including markers surpassed by a fresh claim.
pub fn retirements(&self) -> Vec<SignedRetirement> {
let mut markers: Vec<_> = self.retirements.values().cloned().collect();
markers.sort_by_key(|m| m.target.author);
markers
}
/// Accepts any member's signed cleanup after checking both signatures.
pub fn merge_retirement(
&mut self,
network: NetworkId,
marker: SignedRetirement,
max: usize,
) -> Result<bool, StateError> {
let target = marker.verify(network)?;
if let Some(old) = self.retirements.get(&target) {
if !marker.supersedes(old) {
return Ok(false);
}
} else if self.retirements.len() >= max {
return Err(StateError::Malformed("retirement capacity reached"));
}
self.retirements.insert(target, marker);
Ok(true)
}
/// Merges one record, verifying it first.
///
/// Merging is into the existing set, never a wholesale replacement, and an
@@ -591,6 +646,9 @@ impl StateSet {
pub fn address_holders(&self) -> HashMap<Ipv4Addr, EndpointId> {
let mut holders: HashMap<Ipv4Addr, EndpointId> = HashMap::new();
for (author, record) in &self.records {
if self.is_retired(author) {
continue;
}
let Some(address) = record.body.claimed_address() else {
continue;
};
@@ -618,6 +676,9 @@ impl StateSet {
pub fn hostname_holders(&self) -> HashMap<String, EndpointId> {
let mut holders: HashMap<String, (u64, EndpointId)> = HashMap::new();
for (author, record) in &self.records {
if self.is_retired(author) {
continue;
}
let Some(hostname) = record.body.hostname() else {
continue;
};
@@ -646,6 +707,9 @@ impl StateSet {
/// The name an author claimed and the member that holds it instead, when
/// somebody who claimed it earlier does.
pub fn hostname_taken(&self, author: &EndpointId) -> Option<(&str, EndpointId)> {
if self.is_retired(author) {
return None;
}
let hostname = self.records.get(author)?.body.hostname()?;
let holder = *self.hostname_holders().get(hostname)?;
(holder != *author).then_some((hostname, holder))
@@ -671,6 +735,12 @@ impl StateSet {
pub fn agreed_range(&self) -> Option<Ipv4Range> {
self.records
.iter()
.chain(
self.retirements
.iter()
.filter(|(author, _)| !self.records.contains_key(*author))
.map(|(author, marker)| (author, &marker.target)),
)
.filter_map(|(author, record)| {
let range = record.body.range()?;
let derived = derived_ipv4_range(NetworkId::from_bytes(record.network));
@@ -684,6 +754,18 @@ impl StateSet {
pub fn version_of(&self, author: &EndpointId) -> u64 {
self.records.get(author).map_or(0, |record| record.version)
}
/// Highest witnessed author counter across claims and authored cleanup.
/// This differs from the claim version used to decide retirement.
pub fn authored_version(&self, author: &EndpointId) -> u64 {
self.retirements
.values()
.filter(|m| m.author == *author.as_bytes())
.map(|m| m.version)
.chain(std::iter::once(self.version_of(author)))
.max()
.unwrap_or(0)
}
}
#[cfg(test)]
+219
View File
@@ -0,0 +1,219 @@
//! A member may retire an observed claim, never another identity's future.
use iroh::{EndpointId, SecretKey, Signature};
use serde::{Deserialize, Serialize};
use super::{RecordBody, SignedRecord, StateError, push_lp};
use crate::NetworkId;
/// Maximum encoded retirement size, including its bounded claim proof.
pub const MAX_RETIREMENT_LEN: usize = 1024;
/// A network member's signed retirement of a proved claim version.
///
/// The original author's signed claim proves the cutoff really existed.
/// Newer claims remain valid. This is neither a forged Release nor a ban.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SignedRetirement {
/// Device identity of the member requesting cleanup.
pub author: [u8; 32],
/// Network in which cleanup applies.
pub network: [u8; 32],
/// Requester's durable author counter, committed with this marker.
pub version: u64,
/// The target's original signed claim; bounds the retired version.
pub target: SignedRecord,
/// Requester's Ed25519 signature over this marker and its proof.
pub signature: Vec<u8>,
}
impl SignedRetirement {
fn canonical_bytes(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(768);
push_lp(&mut out, b"tsunagi-node-retirement/v1");
push_lp(&mut out, &self.network);
push_lp(&mut out, &self.author);
push_lp(&mut out, &self.version.to_be_bytes());
// Bind every field of the original signed claim, including its signature.
push_lp(&mut out, &self.target.network);
push_lp(&mut out, &self.target.author);
push_lp(&mut out, &self.target.version.to_be_bytes());
push_lp(&mut out, &self.target.body.canonical());
push_lp(&mut out, &self.target.signature);
out
}
/// Signs a marker with the requesting member's persistent device key.
pub fn sign(
secret: &SecretKey,
network: NetworkId,
version: u64,
target: SignedRecord,
) -> Self {
let mut marker = Self {
author: *secret.public().as_bytes(),
network: *network.as_bytes(),
version,
target,
signature: Vec::new(),
};
marker.signature = secret.sign(&marker.canonical_bytes()).to_bytes().to_vec();
marker
}
/// Verifies both authors, scope and a finite, genuinely signed cutoff.
pub fn verify(&self, network: NetworkId) -> Result<EndpointId, StateError> {
if self.network != *network.as_bytes() {
return Err(StateError::WrongNetwork);
}
if self.version == 0 || self.version > i64::MAX as u64 {
return Err(StateError::Malformed("retirement version is out of range"));
}
if !matches!(self.target.body, RecordBody::Claim { .. }) {
return Err(StateError::Malformed("only a claim can be retired"));
}
let target = self.target.verify(network)?;
let author = EndpointId::from_bytes(&self.author).map_err(|_| StateError::BadAuthor)?;
if author == target {
return Err(StateError::Malformed(
"use Release to retire your own claim",
));
}
let raw: [u8; Signature::LENGTH] =
self.signature
.as_slice()
.try_into()
.map_err(|_| StateError::BadSignatureLength {
expected: Signature::LENGTH,
})?;
author
.verify(&self.canonical_bytes(), &Signature::from_bytes(&raw))
.map_err(|_| StateError::BadSignature)?;
Ok(target)
}
/// A deterministic strongest marker per target, independent of arrival order.
pub(crate) fn supersedes(&self, other: &Self) -> bool {
self.target.version > other.target.version
|| (self.target.version == other.target.version
&& (self.canonical_bytes(), &self.signature)
< (other.canonical_bytes(), &other.signature))
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use super::*;
use crate::state::{Ipv4Range, StateSet};
fn claim(key: &SecretKey, id: NetworkId, version: u64) -> SignedRecord {
SignedRecord::sign(
key,
id,
version,
RecordBody::Claim {
address: Some("10.23.0.2".parse().unwrap()),
range: Some("10.23.0.0/24".parse::<Ipv4Range>().unwrap()),
hostname: Some("retired".into()),
hostname_since: Some(123),
},
)
}
#[test]
fn markers_release_claims_in_any_merge_order_and_older_snapshots_cannot_restore_them() {
let id = NetworkId::from_bytes([7; 32]);
let target = SecretKey::generate();
let cleaner = SecretKey::generate();
let record = claim(&target, id, 5);
let marker = SignedRetirement::sign(&cleaner, id, 8, record.clone());
for first in [true, false] {
let mut state = StateSet::new();
if first {
state.merge_retirement(id, marker.clone(), 16).unwrap();
}
state.merge(id, record.clone()).unwrap();
if !first {
state.merge_retirement(id, marker.clone(), 16).unwrap();
}
assert!(state.address_holders().is_empty());
assert!(state.hostname_holders().is_empty());
state.merge(id, claim(&target, id, 4)).unwrap();
assert!(state.is_retired(&target.public()));
assert!(
state.agreed_range().is_some(),
"cleanup does not renumber the network"
);
state.merge(id, claim(&target, id, 6)).unwrap();
assert!(!state.is_retired(&target.public()));
assert_eq!(state.hostname_of(&target.public()), Some("retired"));
}
}
#[test]
fn proof_signature_scope_tampering_future_cutoffs_and_capacity_are_checked() {
let id = NetworkId::from_bytes([7; 32]);
let target = SecretKey::generate();
let cleaner = SecretKey::generate();
let marker = SignedRetirement::sign(&cleaner, id, 8, claim(&target, id, 5));
assert!(marker.verify(NetworkId::from_bytes([8; 32])).is_err());
let mut forged = marker.clone();
forged.target.version = 99;
assert!(forged.verify(id).is_err());
let mut forged = marker.clone();
forged.signature[0] ^= 1;
assert!(forged.verify(id).is_err());
let mut proof = marker.target.clone();
proof.version = 99;
assert!(
SignedRetirement::sign(&cleaner, id, 8, proof)
.verify(id)
.is_err()
);
let mut state = StateSet::new();
assert!(state.merge_retirement(id, marker.clone(), 0).is_err());
assert!(state.merge_retirement(id, marker.clone(), 1).unwrap());
assert!(!state.merge_retirement(id, marker.clone(), 1).unwrap());
assert!(postcard::to_stdvec(&marker).unwrap().len() <= MAX_RETIREMENT_LEN);
let another = SecretKey::generate();
assert!(
state
.merge_retirement(
id,
SignedRetirement::sign(&cleaner, id, 9, claim(&another, id, 1)),
1
)
.is_err()
);
}
#[test]
fn competing_cleanup_markers_converge_and_keep_the_highest_observed_cutoff() {
let id = NetworkId::from_bytes([7; 32]);
let target = SecretKey::generate();
let a = SecretKey::generate();
let b = SecretKey::generate();
let markers = [
SignedRetirement::sign(&a, id, 1, claim(&target, id, 4)),
SignedRetirement::sign(&a, id, 2, claim(&target, id, 5)),
SignedRetirement::sign(&b, id, 3, claim(&target, id, 5)),
];
let mut forward = StateSet::new();
let mut reverse = StateSet::new();
for marker in &markers {
forward.merge_retirement(id, marker.clone(), 16).unwrap();
}
for marker in markers.iter().rev() {
reverse.merge_retirement(id, marker.clone(), 16).unwrap();
}
assert_eq!(forward.retirements(), reverse.retirements());
assert_eq!(
forward
.retirement_of(&target.public())
.unwrap()
.target
.version,
5
);
}
}
+32
View File
@@ -254,6 +254,12 @@ impl Storage {
.await
}
/// Persists the ordered protocol selection off the async executor.
pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec<String>) -> Result<()> {
self.with_state(move |state| state.set_protocols(network_id, &protocols))
.await
}
/// Persists whether this agent offers itself as an exit node in a network.
pub async fn set_exit_node(&self, network_id: NetworkId, enabled: bool) -> Result<()> {
self.with_state(move |state| state.set_exit_node(network_id, enabled))
@@ -304,6 +310,7 @@ impl Storage {
self.with_state(move |state| {
state.remove_network(network_id)?;
state.forget_signed_records(network_id)?;
state.forget_retirements(network_id)?;
state.forget_peer_hostnames(network_id)
})
.await?;
@@ -363,6 +370,31 @@ impl Storage {
.await
}
/// Loads verified durable cleanup proofs off the executor.
pub async fn retirements(
&self,
network: NetworkId,
max: usize,
) -> Result<Vec<crate::state::SignedRetirement>> {
self.with_state(move |state| state.retirements(network, max))
.await
}
/// Persists a received cleanup proof before applying it.
pub async fn put_retirement(&self, marker: crate::state::SignedRetirement) -> Result<()> {
self.with_state(move |state| state.put_retirement(&marker))
.await
}
/// Commits our cleanup and author counter together before announcement.
pub async fn publish_own_retirement(
&self,
marker: crate::state::SignedRetirement,
) -> Result<()> {
self.with_state(move |state| state.publish_own_retirement(&marker))
.await
}
/// Stores a record received from another replica.
pub async fn put_signed_record(&self, record: SignedRecord) -> Result<()> {
self.with_state(move |state| state.put_signed_record(&record))
+326 -4
View File
@@ -18,10 +18,10 @@ use rusqlite::{Connection, OptionalExtension, params};
use crate::error::{Error, Result};
use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret};
use crate::state::{RecordBody, SignedRecord};
use crate::state::{MAX_RETIREMENT_LEN, RecordBody, SignedRecord, SignedRetirement};
/// Schema version written by this build.
pub const SCHEMA_VERSION: i64 = 7;
pub const SCHEMA_VERSION: i64 = 9;
/// Key of the stored hostname setting.
const SETTING_HOSTNAME: &str = "hostname";
@@ -32,6 +32,8 @@ const SETTING_HOSTNAME: &str = "hostname";
/// and zeroizes on drop.
#[derive(Debug, Clone)]
pub struct StoredNetwork {
/// Ordered local protocol override; absent inherits the agent's defaults.
pub protocols: Option<Vec<String>>,
/// Derived public network identifier.
pub network_id: NetworkId,
/// Network name.
@@ -238,6 +240,29 @@ impl StateStore {
)
.map_err(|err| self.corrupt(format!("cannot migrate schema to 7: {err}")))?;
}
if found < 8 {
self.conn
.execute_batch(
"BEGIN;
ALTER TABLE networks ADD COLUMN protocols TEXT;
PRAGMA user_version = 8;
COMMIT;",
)
.map_err(|err| self.corrupt(format!("cannot migrate schema to 8: {err}")))?;
}
if found < 9 {
self.conn
.execute_batch(
"BEGIN;
CREATE TABLE IF NOT EXISTS retirements (
network_id BLOB NOT NULL, target BLOB NOT NULL, data BLOB NOT NULL,
PRIMARY KEY (network_id, target)
);
PRAGMA user_version = 9;
COMMIT;",
)
.map_err(|err| self.corrupt(format!("cannot migrate schema to 9: {err}")))?;
}
Ok(())
}
@@ -250,6 +275,7 @@ impl StateStore {
"settings",
"signed_records",
"peer_hostnames",
"retirements",
] {
let present: Option<String> = self
.conn
@@ -476,6 +502,25 @@ impl StateStore {
Ok(())
}
/// Saves a local ordered protocol selection without changing membership.
pub fn set_protocols(&self, network_id: NetworkId, protocols: &[String]) -> Result<()> {
crate::dataplane::validate_protocols(
protocols,
crate::config::Limits::default().max_capabilities,
)?;
let changed = self
.conn
.execute(
"UPDATE networks SET protocols = ?2 WHERE network_id = ?1",
params![network_id.as_bytes().as_slice(), protocols.join(",")],
)
.map_err(|err| Error::Storage(format!("cannot update protocols: {err}")))?;
if changed == 0 {
return Err(Error::NetworkUnknown(network_id));
}
Ok(())
}
/// Removes a network configuration entirely.
pub fn remove_network(&self, network_id: NetworkId) -> Result<()> {
self.conn
@@ -536,7 +581,7 @@ impl StateStore {
pub fn list_networks(&self) -> Result<Vec<StoredNetwork>> {
let mut stmt = self
.conn
.prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via FROM networks ORDER BY name")
.prepare("SELECT network_id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols FROM networks ORDER BY name")
.map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?;
let rows = stmt
.query_map([], |row| {
@@ -552,19 +597,37 @@ impl StateStore {
row.get::<_, bool>(4)?,
row.get::<_, bool>(5)?,
row.get::<_, Option<Vec<u8>>>(6)?,
row.get::<_, Option<String>>(7)?,
))
})
.map_err(|err| Error::Storage(format!("cannot list networks: {err}")))?;
let mut out = Vec::new();
for row in rows {
let (id, name, secret, auto_start, broadcast, exit_node, exit_via) =
let (id, name, secret, auto_start, broadcast, exit_node, exit_via, protocols) =
row.map_err(|err| Error::Storage(format!("cannot read network row: {err}")))?;
let protocols = protocols
.map(|text| {
let max = crate::config::Limits::default().max_capabilities;
if text.len() > max * (crate::dataplane::MAX_PROTOCOL_ID_LEN + 1) {
return Err(self.corrupt("stored protocol selection is too long"));
}
let names = if text.is_empty() {
Vec::new()
} else {
text.split(',').map(str::to_owned).collect()
};
crate::dataplane::validate_protocols(&names, max)
.map_err(|err| self.corrupt(err))?;
Ok(names)
})
.transpose()?;
let id: [u8; 32] = id
.as_slice()
.try_into()
.map_err(|_| self.corrupt("stored network id is not 32 bytes"))?;
out.push(StoredNetwork {
protocols,
network_id: NetworkId::from_bytes(id),
name: NetworkName::new(name)?,
secret: NetworkSecret::from_bytes(secret)?,
@@ -736,6 +799,112 @@ impl StateStore {
Ok(version.unwrap_or(0).max(0) as u64)
}
/// Loads durable cleanup markers without allocating oversized stored blobs.
pub fn retirements(&self, network: NetworkId, max: usize) -> Result<Vec<SignedRetirement>> {
let mut stmt = self
.conn
.prepare(
"SELECT CASE WHEN length(data) <= ?2 THEN data ELSE NULL END
FROM retirements WHERE network_id = ?1 ORDER BY target",
)
.map_err(|err| self.corrupt(err))?;
let rows = stmt
.query_map(
params![network.as_bytes().as_slice(), MAX_RETIREMENT_LEN as i64],
|row| row.get::<_, Option<Vec<u8>>>(0),
)
.map_err(|err| self.corrupt(err))?;
let mut out = Vec::new();
for row in rows {
if out.len() >= max {
return Err(Error::Storage(
"stored retirements exceed the configured limit".into(),
));
}
let bytes = row
.map_err(|err| self.corrupt(err))?
.ok_or_else(|| self.corrupt("oversized retirement"))?;
let marker: SignedRetirement =
postcard::from_bytes(&bytes).map_err(|err| self.corrupt(err))?;
marker.verify(network).map_err(|err| self.corrupt(err))?;
out.push(marker);
}
Ok(out)
}
/// Persists a strongest verified marker received from an authenticated session.
pub fn put_retirement(&self, marker: &SignedRetirement) -> Result<()> {
let network = NetworkId::from_bytes(marker.network);
marker
.verify(network)
.map_err(|err| Error::Storage(err.to_string()))?;
let old: Option<Option<Vec<u8>>> = self.conn.query_row(
"SELECT CASE WHEN length(data) <= ?3 THEN data ELSE NULL END FROM retirements WHERE network_id = ?1 AND target = ?2",
params![marker.network.as_slice(), marker.target.author.as_slice(), MAX_RETIREMENT_LEN as i64], |row| row.get(0))
.optional().map_err(|err| self.corrupt(err))?;
if let Some(bytes) = old {
let bytes = bytes.ok_or_else(|| self.corrupt("oversized retirement"))?;
let old: SignedRetirement =
postcard::from_bytes(&bytes).map_err(|err| self.corrupt(err))?;
old.verify(network).map_err(|err| self.corrupt(err))?;
if !marker.supersedes(&old) {
return Ok(());
}
}
let bytes = postcard::to_stdvec(marker).map_err(|err| Error::Storage(err.to_string()))?;
if bytes.len() > MAX_RETIREMENT_LEN {
return Err(Error::Storage(
"retirement exceeds its encoded bound".into(),
));
}
self.conn
.execute(
"INSERT INTO retirements (network_id, target, data) VALUES (?1, ?2, ?3)
ON CONFLICT(network_id, target) DO UPDATE SET data = excluded.data",
params![
marker.network.as_slice(),
marker.target.author.as_slice(),
bytes
],
)
.map_err(|err| Error::Storage(format!("cannot store retirement: {err}")))?;
Ok(())
}
/// Commits our signed cleanup marker and author counter in one transaction.
pub fn publish_own_retirement(&self, marker: &SignedRetirement) -> Result<()> {
let transaction = self
.conn
.unchecked_transaction()
.map_err(|err| Error::Storage(err.to_string()))?;
self.put_retirement(marker)?;
transaction
.execute(
"INSERT INTO own_record_version (network_id, author, version) VALUES (?1, ?2, ?3)
ON CONFLICT(network_id, author) DO UPDATE SET version = max(version, excluded.version)",
params![
marker.network.as_slice(),
marker.author.as_slice(),
marker.version as i64
],
)
.map_err(|err| Error::Storage(format!("cannot store retirement counter: {err}")))?;
transaction
.commit()
.map_err(|err| Error::Storage(format!("cannot commit retirement: {err}")))
}
/// Drops a forgotten network's local cleanup proofs.
pub fn forget_retirements(&self, network: NetworkId) -> Result<()> {
self.conn
.execute(
"DELETE FROM retirements WHERE network_id = ?1",
params![network.as_bytes().as_slice()],
)
.map_err(|err| Error::Storage(err.to_string()))?;
Ok(())
}
/// Forgets every record of a network.
pub fn forget_signed_records(&self, network_id: NetworkId) -> Result<()> {
self.conn
@@ -957,6 +1126,158 @@ mod broadcast_storage_tests {
assert!(store.peer_hostnames(id).unwrap().is_empty());
}
#[test]
fn retirement_migration_and_atomic_counter_commit_preserve_existing_state() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("state.sqlite");
let store = StateStore::open(&path).unwrap();
let identity = store.load_or_create_device_identity().unwrap();
let name = NetworkName::new("retirement-migration").unwrap();
let secret = NetworkSecret::from_bytes([6; 32]).unwrap();
let id = NetworkKeys::derive(&name, &secret).network_id();
store.upsert_network(id, &name, &secret, true).unwrap();
store.set_protocols(id, &["awg".into()]).unwrap();
let own = SignedRecord::sign(&identity.signing_key(), id, 8, RecordBody::Release);
store.publish_own_record(&own).unwrap();
store
.conn
.execute_batch("DROP TABLE retirements; PRAGMA user_version = 8;")
.unwrap();
drop(store);
let store = StateStore::open(&path).unwrap();
assert_eq!(
store.device_identity().unwrap().unwrap().endpoint_id(),
identity.endpoint_id()
);
assert_eq!(
store.list_networks().unwrap()[0].protocols,
Some(vec!["awg".into()])
);
assert_eq!(store.signed_records(id).unwrap(), vec![own]);
let target_key = iroh::SecretKey::generate();
let target = SignedRecord::sign(
&target_key,
id,
3,
RecordBody::Claim {
address: Some("10.13.37.2".parse().unwrap()),
range: Some(crate::state::DEFAULT_IPV4_RANGE),
hostname: Some("old-device".into()),
hostname_since: Some(123),
},
);
let marker = SignedRetirement::sign(&identity.signing_key(), id, 9, target);
store
.conn
.execute_batch(
"CREATE TRIGGER fail_retirement_counter BEFORE INSERT ON own_record_version
WHEN NEW.version = 9 BEGIN SELECT RAISE(ABORT, 'counter write refused'); END;",
)
.unwrap();
assert!(store.publish_own_retirement(&marker).is_err());
assert!(
store.retirements(id, 16).unwrap().is_empty(),
"marker must roll back with counter failure"
);
assert_eq!(
store
.own_record_version(id, identity.endpoint_id())
.unwrap(),
8
);
store
.conn
.execute_batch("DROP TRIGGER fail_retirement_counter;")
.unwrap();
store.publish_own_retirement(&marker).unwrap();
drop(store);
let store = StateStore::open(&path).unwrap();
assert_eq!(store.retirements(id, 16).unwrap(), vec![marker.clone()]);
assert_eq!(
store
.own_record_version(id, identity.endpoint_id())
.unwrap(),
9
);
assert!(
store.retirements(id, 0).is_err(),
"capacity changes must not silently drop markers"
);
store
.conn
.execute(
"UPDATE retirements SET data = ?1",
params![vec![0u8; MAX_RETIREMENT_LEN + 1]],
)
.unwrap();
assert!(matches!(
store.retirements(id, 16),
Err(Error::StateCorrupted { .. })
));
}
#[test]
fn v7_migration_keeps_signed_state_and_protocol_overrides_are_local_and_durable() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("state.sqlite");
let store = StateStore::open(&path).unwrap();
let identity = store.load_or_create_device_identity().unwrap();
let name = NetworkName::new("protocol-migration").unwrap();
let secret = NetworkSecret::from_bytes([4; 32]).unwrap();
let id = NetworkKeys::derive(&name, &secret).network_id();
store.upsert_network(id, &name, &secret, true).unwrap();
store.set_broadcast(id, false).unwrap();
let record = SignedRecord::sign(&identity.signing_key(), id, 17, RecordBody::Release);
store.publish_own_record(&record).unwrap();
store
.conn
.execute_batch("ALTER TABLE networks DROP COLUMN protocols; PRAGMA user_version=7;")
.unwrap();
drop(store);
let store = StateStore::open(&path).unwrap();
assert_eq!(
store
.load_or_create_device_identity()
.unwrap()
.endpoint_id(),
identity.endpoint_id()
);
assert_eq!(store.signed_records(id).unwrap(), vec![record]);
let saved = store.list_networks().unwrap();
assert!(saved[0].protocols.is_none());
assert!(!saved[0].broadcast);
assert!(saved[0].auto_start);
assert_eq!(saved[0].secret, secret);
let chosen = vec!["awg".into(), "wg-quic".into()];
store.set_protocols(id, &chosen).unwrap();
assert!(
store
.set_protocols(id, &["awg".into(), "awg".into()])
.is_err()
);
assert!(
store
.set_protocols(NetworkId::from_bytes([1; 32]), &[])
.is_err()
);
drop(store);
let store = StateStore::open(&path).unwrap();
assert_eq!(store.list_networks().unwrap()[0].protocols, Some(chosen));
store.upsert_network(id, &name, &secret, true).unwrap();
store.set_protocols(id, &[]).unwrap();
drop(store);
let store = StateStore::open(&path).unwrap();
assert_eq!(store.list_networks().unwrap()[0].protocols, Some(vec![]));
store
.conn
.execute("UPDATE networks SET protocols = 'awg,awg'", [])
.unwrap();
assert!(matches!(
store.list_networks(),
Err(Error::StateCorrupted { .. })
));
}
#[test]
fn v3_migration_preserves_identity_and_networks_and_defaults_broadcast_on() {
let dir = tempfile::tempdir().unwrap();
@@ -977,6 +1298,7 @@ mod broadcast_storage_tests {
"ALTER TABLE networks DROP COLUMN broadcast; \
ALTER TABLE networks DROP COLUMN exit_node; \
ALTER TABLE networks DROP COLUMN exit_via; \
ALTER TABLE networks DROP COLUMN protocols; \
PRAGMA user_version=3;",
)
.unwrap();
+358
View File
@@ -0,0 +1,358 @@
//! Equal-rights retirement uses real membership, signatures and persistent state.
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::testing::{TestAgent, config_with, network, wait_for_peers, wait_until};
use tsunagi::{Agent, NetworkId};
#[tokio::test]
async fn bounded_marker_batches_eventually_reach_the_tail_of_a_small_send_queue() {
use tsunagi::config::StoragePaths;
use tsunagi::state::{RecordBody, SignedRecord, SignedRetirement};
use tsunagi::storage::StateStore;
let discovery = SharedMemoryDiscovery::new();
let cleaner = TestAgent::spawn(&discovery).await.unwrap();
let replica = TestAgent::spawn(&discovery).await.unwrap();
let (name, secret) = network("forget-batched-history");
let id = cleaner.agent.join_network(&name, &secret).await.unwrap();
replica.agent.join_network(&name, &secret).await.unwrap();
let cleaner_dir = cleaner.stop().await;
let replica_dir = replica.stop().await;
let cleaner_path = StoragePaths::under(cleaner_dir.path()).state_db();
let replica_path = StoragePaths::under(replica_dir.path()).state_db();
// Seed genuine signed offline history in closed stores. Delivery still uses
// real agents, authenticated loopback sessions and the persistent stores.
let targets = tokio::task::spawn_blocking(move || {
let cleaner = StateStore::open(&cleaner_path).unwrap();
let replica = StateStore::open(&replica_path).unwrap();
let author = cleaner.device_identity().unwrap().unwrap();
let signing_key = iroh::SecretKey::from_bytes(&author.secret_bytes());
let mut targets = Vec::new();
for index in 0..12 {
let target = iroh::SecretKey::generate();
let record = SignedRecord::sign(
&target,
id,
1,
RecordBody::Claim {
address: None,
range: None,
hostname: Some(format!("offline-{index}")),
hostname_since: Some(1),
},
);
cleaner.put_signed_record(&record).unwrap();
replica.put_signed_record(&record).unwrap();
cleaner
.publish_own_retirement(&SignedRetirement::sign(
&signing_key,
id,
100 + index,
record,
))
.unwrap();
targets.push(target.public());
}
targets
})
.await
.unwrap();
let replica = Agent::spawn(config_with(replica_dir.path(), &discovery))
.await
.unwrap();
assert_eq!(
replica
.network_status(id)
.await
.unwrap()
.members
.iter()
.filter(|m| targets.contains(&m.endpoint_id))
.count(),
12
);
let mut config = config_with(cleaner_dir.path(), &discovery);
config.limits.max_state_records = 1;
config.limits.session_send_queue = 2;
let cleaner = Agent::spawn(config).await.unwrap();
wait_for_peers(&cleaner, id, 1).await;
wait_until(
"all cleanup batches arrive despite send queue pressure",
|| async {
replica
.network_status(id)
.await
.ok()?
.members
.iter()
.all(|m| !targets.contains(&m.endpoint_id))
.then_some(())
},
)
.await;
cleaner.shutdown().await;
replica.shutdown().await;
let replica_path = StoragePaths::under(replica_dir.path()).state_db();
tokio::task::spawn_blocking(move || {
let replica = StateStore::open(&replica_path).unwrap();
assert_eq!(replica.retirements(id, 16).unwrap().len(), 12);
})
.await
.unwrap();
}
async fn member(
agent: &Agent,
network: NetworkId,
peer: iroh::EndpointId,
) -> tsunagi::agent::MemberStatus {
wait_until("signed member", || async {
agent
.network_status(network)
.await
.ok()?
.members
.into_iter()
.find(|m| m.endpoint_id == peer)
})
.await
}
#[tokio::test]
async fn retirement_survives_restart_and_stale_replicas_but_allows_a_new_claim() {
let discovery = SharedMemoryDiscovery::new();
let cleaner = TestAgent::spawn_with(|c| c.with_hostname("cleaner"), &discovery)
.await
.unwrap();
let target = TestAgent::spawn_with(|c| c.with_hostname("old-device"), &discovery)
.await
.unwrap();
let replica = TestAgent::spawn_with(|c| c.with_hostname("replica"), &discovery)
.await
.unwrap();
let (name, secret) = network("forget-propagation");
let id = cleaner.agent.join_network(&name, &secret).await.unwrap();
target.agent.join_network(&name, &secret).await.unwrap();
replica.agent.join_network(&name, &secret).await.unwrap();
wait_for_peers(&cleaner.agent, id, 2).await;
let peer = target.agent.endpoint_id();
let original = wait_until("all three agents agree on allocated claims", || async {
let a = cleaner.agent.network_status(id).await.ok()?;
let b = target.agent.network_status(id).await.ok()?;
let c = replica.agent.network_status(id).await.ok()?;
for status in [&a, &b, &c] {
if status.members.len() != 3 {
return None;
}
let addresses: std::collections::HashSet<_> = status
.members
.iter()
.filter_map(|m| m.overlay_address_v4)
.collect();
if addresses.len() != 3 {
return None;
}
}
let record = a.members.iter().find(|m| m.endpoint_id == peer)?;
(b.members
.iter()
.find(|m| m.endpoint_id == peer)?
.record_version
== record.record_version
&& c.members
.iter()
.find(|m| m.endpoint_id == peer)?
.record_version
== record.record_version)
.then(|| record.clone())
})
.await;
assert!(
cleaner
.agent
.forget_node(id, peer, original.record_version)
.await
.is_err(),
"online nodes are not history"
);
assert!(
cleaner
.agent
.forget_node(id, cleaner.agent.endpoint_id(), 1)
.await
.is_err()
);
let replica_dir = replica.stop().await;
let target_dir = target.stop().await;
wait_until("target offline", || async {
cleaner
.agent
.network_status(id)
.await
.ok()?
.members
.iter()
.find(|m| m.endpoint_id == peer)
.filter(|m| !m.online)
.map(|_| ())
})
.await;
assert!(
cleaner
.agent
.forget_node(id, peer, original.record_version + 1)
.await
.is_err(),
"stale confirmation changes nothing"
);
let result = cleaner
.agent
.forget_node(id, peer, original.record_version)
.await
.unwrap();
assert_eq!(result.target, peer);
assert_eq!(result.record_version, original.record_version);
assert_eq!(result.peers_told, 0);
assert!(
!cleaner
.agent
.network_status(id)
.await
.unwrap()
.members
.iter()
.any(|m| m.endpoint_id == peer)
);
let cleaner_dir = cleaner.stop().await;
let cleaner =
Agent::spawn(config_with(cleaner_dir.path(), &discovery).with_hostname("cleaner"))
.await
.unwrap();
let replica =
Agent::spawn(config_with(replica_dir.path(), &discovery).with_hostname("replica"))
.await
.unwrap();
wait_for_peers(&cleaner, id, 1).await;
wait_until("stale replica learns durable retirement", || async {
replica
.network_status(id)
.await
.ok()?
.members
.iter()
.all(|m| m.endpoint_id != peer)
.then_some(())
})
.await;
assert!(
!cleaner
.network_status(id)
.await
.unwrap()
.members
.iter()
.any(|m| m.endpoint_id == peer)
);
replica.set_hostname("old-device").await.unwrap();
let replica_id = replica.endpoint_id();
wait_until("released hostname is reclaimed", || async {
cleaner
.network_status(id)
.await
.ok()?
.members
.iter()
.find(|m| m.endpoint_id == replica_id)
.filter(|m| m.hostname.as_deref() == Some("old-device"))?
.hostname_conflict
.is_none()
.then_some(())
})
.await;
let returned =
Agent::spawn(config_with(target_dir.path(), &discovery).with_hostname("old-device"))
.await
.unwrap();
wait_for_peers(&returned, id, 2).await;
wait_until("returning identity publishes a fresh version", || async {
let m = cleaner
.network_status(id)
.await
.ok()?
.members
.into_iter()
.find(|m| m.endpoint_id == peer)?;
(m.online && m.record_version > original.record_version && m.hostname_conflict.is_some())
.then_some(())
})
.await;
cleaner.shutdown().await;
replica.shutdown().await;
returned.shutdown().await;
}
#[tokio::test]
async fn forgetting_is_network_scoped_and_every_member_has_the_same_right() {
let discovery = SharedMemoryDiscovery::new();
let a = TestAgent::spawn(&discovery).await.unwrap();
let b = TestAgent::spawn(&discovery).await.unwrap();
let target = TestAgent::spawn(&discovery).await.unwrap();
let (name, secret) = network("forget-one-network");
let (other_name, other_secret) = network("keep-other-network");
let id = a.agent.join_network(&name, &secret).await.unwrap();
b.agent.join_network(&name, &secret).await.unwrap();
target.agent.join_network(&name, &secret).await.unwrap();
let other = a
.agent
.join_network(&other_name, &other_secret)
.await
.unwrap();
target
.agent
.join_network(&other_name, &other_secret)
.await
.unwrap();
let peer = target.agent.endpoint_id();
let record = member(&b.agent, id, peer).await;
member(&a.agent, other, peer).await;
target.agent.shutdown().await;
wait_until("offline on B", || async {
b.agent
.network_status(id)
.await
.ok()?
.members
.iter()
.find(|m| m.endpoint_id == peer)
.filter(|m| !m.online)
.map(|_| ())
})
.await;
b.agent
.forget_node(id, peer, record.record_version)
.await
.unwrap();
wait_until("A learns B's retirement", || async {
a.agent
.network_status(id)
.await
.ok()?
.members
.iter()
.all(|m| m.endpoint_id != peer)
.then_some(())
})
.await;
assert!(
a.agent
.network_status(other)
.await
.unwrap()
.members
.iter()
.any(|m| m.endpoint_id == peer)
);
a.agent.shutdown().await;
b.agent.shutdown().await;
}
+49 -2
View File
@@ -21,6 +21,17 @@ const POLL_INTERVAL: Duration = Duration::from_secs(2);
/// A request from the UI to the agent.
pub(crate) enum Command {
/// Retire the exact offline-node claim selected in the devices window.
ForgetNode {
network_id: String,
peer: String,
expected_version: u64,
},
/// Replace the network's enabled protocols in preference order.
SetProtocols {
network_id: String,
protocols: Vec<String>,
},
/// Stop serving a network, or start serving it again.
SetActive { network_id: String, active: bool },
/// Turn local broadcast participation on or off for a network.
@@ -106,7 +117,13 @@ impl AgentClient {
/// Sends a command; a dropped worker simply means nothing happens.
pub(crate) fn send(&self, command: Command) {
let _ = self.tx.send(command);
// Disable protocol editing until the refreshed snapshot arrives;
// otherwise a second checkbox click could replace the first edit
// using the same stale network selection.
lock(&self.shared).busy = true;
if self.tx.send(command).is_err() {
lock(&self.shared).busy = false;
}
}
/// A copy of the latest snapshot.
@@ -142,6 +159,8 @@ async fn worker(
let Some(command) = command else { return };
run(&resolve_socket(), command, &shared, &pending_copy, &ctx).await;
refresh(&shared, &ctx).await;
lock(&shared).busy = false;
ctx.request_repaint();
}
}
}
@@ -179,6 +198,35 @@ async fn run(
ctx.request_repaint();
let outcome: Result<String, String> = match command {
Command::ForgetNode {
network_id,
peer,
expected_version,
} => ipc::forget_node(socket, &network_id, &peer, expected_version)
.await
.map(|report| {
format!(
"forgot node claim v{}; cleanup queued to {} peers",
report.record_version, report.peers_told
)
})
.map_err(|err| err.to_string()),
Command::SetProtocols {
network_id,
protocols,
} => ipc::set_protocols(socket, &network_id, protocols)
.await
.map(|names| {
format!(
"protocols: {}",
if names.is_empty() {
"none".into()
} else {
names.join(", ")
}
)
})
.map_err(|err| err.to_string()),
Command::SetActive { network_id, active } => ipc::set_active(socket, &network_id, active)
.await
.map(|report| {
@@ -260,7 +308,6 @@ async fn run(
{
let mut guard = lock(shared);
guard.busy = false;
guard.last_action = if matches!(&outcome, Ok(message) if message.is_empty()) {
None
} else {
+4 -3
View File
@@ -116,20 +116,21 @@ impl App {
.with_title(devices::title(network))
.with_app_id(crate::icon::APP_ID)
.with_icon(crate::icon::window())
.with_inner_size([940.0, 540.0])
.with_min_inner_size([480.0, 320.0]);
.with_inner_size([1440.0, 640.0])
.with_min_inner_size([800.0, 400.0]);
let agent = &self.agent;
let traffic = &self.traffic;
// The window shares the single unit setting, so its toggle and the
// main window's stay in step.
let unit = &mut self.state.unit;
let confirm = &mut self.state.exit_confirm;
let forget_confirm = &mut self.state.forget_confirm;
let keep = ctx.show_viewport_immediate(viewport_id, builder, |vctx, _class| {
devices::show(
vctx,
agent,
unit,
confirm,
(confirm, forget_confirm),
network,
&report.endpoint_id,
traffic,
+95 -3
View File
@@ -27,6 +27,8 @@ pub(crate) fn title(network: &NetworkReport) -> String {
/// One row of the device table, built from a tunnel and/or a signed member.
struct Device<'a> {
/// Observed claim version, if this row came from signed membership.
record_version: Option<u64>,
/// The agent's own answer to whether it can be reached by any path.
online: bool,
/// How packets reach it.
@@ -65,6 +67,15 @@ pub(crate) struct ExitConfirm {
main: bool,
}
/// An exact offline claim selected for durable cleanup.
pub(crate) struct ForgetConfirm {
network_id: String,
peer: String,
version: u64,
name: String,
address: Option<String>,
}
impl ExitConfirm {
/// A question about stopping the exit node a network's tile shows.
pub(crate) fn stop_from_tile(network: &NetworkReport, peer_id: &str) -> Self {
@@ -121,11 +132,12 @@ pub(crate) fn show(
ctx: &egui::Context,
agent: &AgentClient,
unit: &mut Unit,
confirm: &mut Option<ExitConfirm>,
confirmations: (&mut Option<ExitConfirm>, &mut Option<ForgetConfirm>),
network: &NetworkReport,
own_id: &str,
traffic: &Traffic,
) {
let (confirm, forget_confirm) = confirmations;
egui::CentralPanel::default().show(ctx, |ui| {
ui.horizontal(|ui| {
ui.heading(&network.name);
@@ -158,10 +170,68 @@ pub(crate) fn show(
egui::ScrollArea::both()
.auto_shrink([false, false])
.show(ui, |ui| {
device_table(ui, unit, network, &devices, traffic, confirm);
device_table(
ui,
unit,
network,
&devices,
traffic,
confirm,
forget_confirm,
);
});
});
confirm_exit(ctx, agent, confirm, Some(&network.network_id));
confirm_forget(ctx, agent, forget_confirm, &network.network_id);
}
fn confirm_forget(
ctx: &egui::Context,
agent: &AgentClient,
confirm: &mut Option<ForgetConfirm>,
network: &str,
) {
let Some(pending) = confirm.as_ref().filter(|c| c.network_id == network) else {
return;
};
egui::Area::new(egui::Id::new("forget node backdrop"))
.order(egui::Order::Middle)
.fixed_pos(egui::Pos2::ZERO)
.show(ctx, |ui| {
let screen = ctx.screen_rect();
ui.allocate_response(screen.size(), egui::Sense::click_and_drag());
ui.painter()
.rect_filled(screen, 0.0, egui::Color32::from_black_alpha(140));
});
let (mut accepted, mut cancelled) = (false, false);
egui::Window::new("Forget offline node")
.order(egui::Order::Foreground)
.collapsible(false)
.resizable(false)
.anchor(egui::Align2::CENTER_CENTER, [0.0, 0.0])
.show(ctx, |ui| {
ui.set_max_width(440.0);
ui.label(format!("Forget {}?", pending.name));
ui.monospace(&pending.peer);
if let Some(address) = &pending.address {
ui.label(format!("IP: {address}"));
}
ui.label("The network will forget this claim and free its name and IP. A returning device can publish a new claim; this does not ban it.");
ui.horizontal(|ui| {
accepted = ui.add_enabled(!agent.snapshot().busy, egui::Button::new("Forget")).clicked();
cancelled = ui.button("Cancel").clicked();
});
});
if accepted {
agent.send(Command::ForgetNode {
network_id: pending.network_id.clone(),
peer: pending.peer.clone(),
expected_version: pending.version,
});
}
if accepted || cancelled || ctx.input(|i| i.key_pressed(egui::Key::Escape)) {
*confirm = None;
}
}
/// The confirmation for an exit-node choice, drawn in the window it was
@@ -354,6 +424,7 @@ fn devices<'a>(network: &'a NetworkReport, own_id: &str) -> Vec<Device<'a>> {
});
let hostname = hostnames.get(id).copied();
Device {
record_version: member.map(|m| m.record_version),
online,
path,
via_name,
@@ -493,6 +564,7 @@ fn device_table(
devices: &[Device<'_>],
traffic: &Traffic,
confirm: &mut Option<ExitConfirm>,
forget_confirm: &mut Option<ForgetConfirm>,
) {
let network_id = network.network_id.as_str();
if devices.is_empty() {
@@ -542,7 +614,26 @@ fn device_table(
ui.end_row();
for device in devices {
cell(ui, 0, &mut widths, |ui| name_cell(ui, device));
cell(ui, 0, &mut widths, |ui| {
ui.horizontal(|ui| {
name_cell(ui, device);
if !device.online
&& !device.in_use
&& let Some(version) = device.record_version
&& ui.add_enabled(network.active, egui::Button::new("Forget"))
.on_hover_text("Forget this offline claim across the network; returning devices may rejoin")
.clicked()
{
*forget_confirm = Some(ForgetConfirm {
network_id: network.network_id.clone(),
peer: device.id.into(),
version,
name: device.name(),
address: device.address.clone(),
});
}
});
});
cell(ui, 1, &mut widths, |ui| {
exit_cell(ui, device, network_id, confirm);
});
@@ -736,6 +827,7 @@ mod tests {
fn member(id: &str, online: bool, path: DataPathReport) -> MemberReport {
MemberReport {
record_version: 1,
endpoint_id: id.into(),
online,
path,
+89 -4
View File
@@ -45,6 +45,8 @@ pub(crate) struct UiState {
pub(crate) about_open: bool,
/// An exit-node choice waiting for the user to confirm it.
pub(crate) exit_confirm: Option<devices::ExitConfirm>,
/// A selected offline claim waiting for cleanup confirmation.
pub(crate) forget_confirm: Option<devices::ForgetConfirm>,
}
impl Default for UiState {
@@ -56,6 +58,7 @@ impl Default for UiState {
open_devices: BTreeSet::new(),
about_open: false,
exit_confirm: None,
forget_confirm: None,
}
}
}
@@ -123,7 +126,7 @@ pub(crate) fn draw(
ui.weak("No networks yet — join one below.");
}
for network in &report.networks {
draw_network(ui, agent, state, traffic, network, &report.endpoint_id);
draw_network(ui, agent, state, traffic, network, report);
}
});
}
@@ -202,7 +205,7 @@ fn draw_network(
state: &mut UiState,
traffic: &Traffic,
network: &NetworkReport,
own_id: &str,
report: &StatusReport,
) {
ui.add_space(6.0);
egui::Frame::group(ui.style()).show(ui, |ui| {
@@ -244,12 +247,14 @@ fn draw_network(
ui.weak("peers");
ui.label(format!(
"{} online · {} known",
devices::online_count(network, own_id),
devices::known_count(network, own_id)
devices::online_count(network, &report.endpoint_id),
devices::known_count(network, &report.endpoint_id)
));
ui.end_row();
});
draw_protocols(ui, agent, network, &report.available_protocols);
format::sparkline(ui, series, state.unit, 34.0);
if !network.range_conflicts.is_empty() {
@@ -345,6 +350,86 @@ fn draw_network(
});
}
fn draw_protocols(
ui: &mut egui::Ui,
agent: &AgentClient,
network: &NetworkReport,
available: &[tsunagi::ipc::ProtocolReport],
) {
let summary = if network.protocols.is_empty() {
"none".into()
} else {
network.protocols.join(", ")
};
egui::CollapsingHeader::new(format!("protocols {summary}"))
.id_salt(("protocols", &network.network_id))
.show(ui, |ui| {
ui.weak("Preference order; changes apply immediately and are saved.");
let mut selected = network.protocols.clone();
let mut changed = false;
ui.add_enabled_ui(!agent.snapshot().busy, |ui| {
// Enabled items follow the user's preference; unchecked items
// follow the agent's catalog. The worker refreshes after edits.
let ordered: Vec<_> = network
.protocols
.iter()
.filter_map(|name| available.iter().find(|p| &p.name == name))
.chain(
available
.iter()
.filter(|p| !network.protocols.contains(&p.name)),
)
.collect();
for protocol in ordered {
ui.horizontal(|ui| {
let mut on = selected.contains(&protocol.name);
if ui.checkbox(&mut on, &protocol.name).changed() {
if on {
selected.push(protocol.name.clone());
} else {
selected.retain(|p| p != &protocol.name);
}
changed = true;
}
if let Some(index) = selected.iter().position(|p| p == &protocol.name) {
if ui
.add_enabled(index > 0, egui::Button::new(icon::ARROW_UP))
.on_hover_text("prefer earlier")
.clicked()
{
selected.swap(index, index - 1);
changed = true;
}
if ui
.add_enabled(
index + 1 < selected.len(),
egui::Button::new(icon::ARROW_DOWN),
)
.on_hover_text("prefer later")
.clicked()
{
selected.swap(index, index + 1);
changed = true;
}
}
});
ui.indent(("description", &network.network_id, &protocol.name), |ui| {
ui.weak(&protocol.summary);
});
}
});
if selected.is_empty() {
ui.weak("Control only; no IP traffic in this network.");
}
if changed {
agent.send(Command::SetProtocols {
network_id: network.network_id.clone(),
protocols: selected,
});
}
});
}
/// A prominent line about an exit node. One that carries a command copies it
/// when clicked.
fn draw_exit_warning(ui: &mut egui::Ui, warning: &exit::Warning) {
+274 -61
View File
@@ -69,8 +69,8 @@ enum Command {
Up(Box<UpArgs>),
/// Reports this device, what the agent is doing, and what this host can do.
Status(StatusArgs),
/// Shows the protocols this build can carry packets with.
Protocols,
/// Lists protocols, or shows and changes one network's selection at runtime.
Protocols(ProtocolsArgs),
/// Makes a network, or joins one — the command that does the work.
///
/// The same as `tsng network join`, at the top level because it is
@@ -107,6 +107,23 @@ struct DnsArgs {
action: Option<DnsAction>,
}
#[derive(Debug, Args)]
#[command(
after_help = "Examples:\n tsng protocols\n tsng protocols home\n tsng protocols home awg,wg-quic\n tsng protocols home none\n\nThe selection is local to this device, saved per network and applied immediately.\nBoth endpoints need a common protocol at the same wire version."
)]
struct ProtocolsArgs {
#[command(flatten)]
paths: PathArgs,
/// Control socket to talk to; uses the running service by default.
#[arg(long, env = "TSUNAGI_CONTROL_SOCKET")]
control_socket: Option<PathBuf>,
/// Network's exact name, id or unique id prefix.
network: Option<String>,
/// Replace the selection, best first: awg,wg-quic; none disables data traffic.
#[arg(value_name = "LIST", requires = "network")]
protocols: Option<String>,
}
#[derive(Debug, Subcommand)]
enum DnsAction {
/// Starts serving, now and after every restart.
@@ -187,6 +204,13 @@ struct JoinArgs {
#[derive(Debug, Subcommand)]
enum NetworkAction {
/// Forgets an offline node's observed claim on every replica; newer claims may return.
ForgetNode {
/// Exact network name, full id or unique id prefix.
network: String,
/// Exact node hostname, full endpoint id, unique id prefix or overlay IPv4.
node: String,
},
/// Shows or changes this network's local LAN broadcast participation.
Broadcast {
/// Exact network name, full id or unique id prefix.
@@ -465,11 +489,13 @@ struct UpArgs {
#[arg(long, help_heading = "System")]
no_tun: bool,
/// Protocols to carry packets with, best first.
/// Initial protocol selection for networks without a saved choice, best first.
///
/// A pair of peers uses one they both have at the same wire version. A
/// peer with none in common keeps its control plane and gets no data
/// plane. `none` runs the control plane alone.
/// plane. `none` runs the control plane alone. `tsng protocols NETWORK LIST`
/// changes and saves each network's choice while this agent runs; any
/// built-in protocol can be enabled, even if absent from this initial list.
#[arg(
long = "protocol",
value_name = "LIST",
@@ -897,7 +923,7 @@ async fn run(command: Command) -> Result<(), Box<dyn std::error::Error>> {
Command::Id(args) => id(args).await,
Command::Up(args) => up(*args).await,
Command::Status(args) => status(args).await,
Command::Protocols => show_protocols(),
Command::Protocols(args) => protocols_command(args).await,
Command::Join(args) => join_command(args).await,
Command::Network(args) => network_command(args).await,
Command::Dns(args) => dns_command(args).await,
@@ -1454,6 +1480,7 @@ const PROTOCOLS: &[ProtocolSpec] = &[
];
/// One `-o` setting, and the protocol it was aimed at.
#[derive(Clone)]
struct Setting {
/// `Some` when written as `protocol:key=value`.
protocol: Option<String>,
@@ -1550,6 +1577,31 @@ struct AgentControl {
}
impl tsunagi::ipc::ReportSource for AgentControl {
fn forget_node(
&self,
network: String,
peer: String,
expected_version: u64,
) -> tsunagi::BoxFuture<'_, Result<tsunagi::ipc::ForgottenNodeReport, String>> {
Box::pin(async move {
let network = network
.parse()
.map_err(|err| format!("invalid network id: {err}"))?;
let peer = peer
.parse()
.map_err(|err| format!("invalid endpoint id: {err}"))?;
let result = self
.agent
.forget_node(network, peer, expected_version)
.await
.map_err(|err| err.to_string())?;
Ok(tsunagi::ipc::ForgottenNodeReport {
peer: result.target.to_string(),
record_version: result.record_version,
peers_told: result.peers_told as u32,
})
})
}
fn report(&self) -> tsunagi::BoxFuture<'_, tsunagi::ipc::StatusReport> {
Box::pin(async move {
let dns = self.dns_state().await;
@@ -1704,6 +1756,23 @@ impl tsunagi::ipc::ReportSource for AgentControl {
})
}
fn set_protocols(
&self,
network_id: String,
protocols: Vec<String>,
) -> tsunagi::BoxFuture<'_, Result<Vec<String>, String>> {
Box::pin(async move {
let id = network_id
.parse()
.map_err(|err| format!("invalid network id: {err}"))?;
self.agent
.set_protocols(id, protocols.clone())
.await
.map_err(|err| err.to_string())?;
Ok(protocols)
})
}
fn set_exit_offer(
&self,
network_id: String,
@@ -1866,45 +1935,178 @@ fn resolve_network<'a>(
networks: &'a [tsunagi::storage::StoredNetwork],
wanted: &str,
) -> Result<&'a tsunagi::storage::StoredNetwork, String> {
let entries: Vec<_> = networks
.iter()
.map(|n| (n.network_id.to_string(), n.name.as_str().to_owned()))
.collect();
resolve_network_index(&entries, wanted).map(|i| &networks[i])
}
/// Shared selector rules for stored networks and secret-free live reports.
fn resolve_network_index(networks: &[(String, String)], wanted: &str) -> Result<usize, String> {
let wanted = wanted.trim_matches(|c: char| c.is_ascii_whitespace());
if wanted.is_empty() {
return Err("select a network by name or id; `tsng network` lists them".to_string());
return Err("select a network by name or id; `tsng network` lists them".into());
}
if let Some(network) = networks
.iter()
.find(|network| network.network_id.to_string() == wanted)
{
return Ok(network);
if let Some(index) = networks.iter().position(|(id, _)| id == wanted) {
return Ok(index);
}
// Only id prefixes lose a copied report ellipsis; names are used verbatim.
let prefix = wanted.trim_end_matches('…');
let matched: Vec<&tsunagi::storage::StoredNetwork> = networks
let matches: Vec<_> = networks
.iter()
.filter(|network| {
network.name.as_str() == wanted
|| (!prefix.is_empty() && network.network_id.to_string().starts_with(prefix))
})
.enumerate()
.filter(|(_, (id, name))| name == wanted || (!prefix.is_empty() && id.starts_with(prefix)))
.collect();
match matched.as_slice() {
[one] => Ok(one),
match matches.as_slice() {
[(index, _)] => Ok(*index),
[] => Err(format!(
"no configured network matches name or id `{wanted}`; \
`tsng network` lists them"
"no configured network matches name or id `{wanted}`; `tsng network` lists them"
)),
several => {
let candidates = several
several => Err(format!(
"`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{}",
several.len(),
several
.iter()
.map(|network| format!(" `{}` {}", network.name, network.network_id))
.map(|(_, (id, name))| format!(" `{name}` {id}"))
.collect::<Vec<_>>()
.join("\n");
Err(format!(
"`{wanted}` matches {} networks; use the full id or a unique id prefix:\n{candidates}",
several.len()
))
}
.join("\n")
)),
}
}
fn parse_protocol_list(text: &str) -> Result<Vec<String>, String> {
if text.eq_ignore_ascii_case("none") {
return Ok(Vec::new());
}
let mut names = Vec::new();
for name in text.split(',') {
let name = name.trim();
let name = if name.eq_ignore_ascii_case("wireguard") {
WG_PROTOCOL
} else {
name
};
if !PROTOCOLS.iter().any(|spec| spec.name == name) {
return Err(format!(
"unknown protocol `{name}`; use wg,awg,wg-quic,tcp-tls or `none` alone. Run `tsng protocols` for details"
));
}
if names.iter().any(|n| n == name) {
return Err(format!("protocol `{name}` is listed more than once"));
}
names.push(name.to_owned());
}
Ok(names)
}
async fn protocols_command(args: ProtocolsArgs) -> Result<(), Box<dyn std::error::Error>> {
let Some(selector) = args.network else {
return show_protocols();
};
let requested = args
.protocols
.as_deref()
.map(parse_protocol_list)
.transpose()?;
let custom = args.paths.state_dir.is_some();
let paths = args.paths.resolve()?;
let socket = control_socket(&paths, args.control_socket.as_ref(), custom);
if tsunagi::ipc::is_serving(&socket).await {
let report = tsunagi::ipc::request_status(&socket).await?;
let entries: Vec<_> = report
.networks
.iter()
.map(|n| (n.network_id.clone(), n.name.clone()))
.collect();
let network = &report.networks[resolve_network_index(&entries, &selector)?];
let protocols = match requested {
Some(protocols) => {
tsunagi::ipc::set_protocols(&socket, &network.network_id, protocols).await?
}
None => network.protocols.clone(),
};
println!(
"protocols {} for `{}` ({}) · preference order",
if protocols.is_empty() {
"none".into()
} else {
protocols.join(",")
},
network.name,
network.network_id
);
} else {
// Take the directory ownership lock even for an offline edit. Never
// write behind a live agent whose socket cannot be reached.
let storage = tsunagi::storage::Storage::open(&paths)?;
let configured = storage.list_networks().await?;
let network = resolve_network(&configured, &selector)?;
let protocols = match requested {
Some(protocols) => {
storage
.set_protocols(network.network_id, protocols.clone())
.await?;
Some(protocols)
}
None => network.protocols.clone(),
};
let shown = protocols
.map(|names| {
if names.is_empty() {
"none".into()
} else {
names.join(",")
}
})
.unwrap_or_else(|| "defaults from tsng up --protocol".into());
println!(
"protocols {shown} for `{}` ({}) · agent stopped; applies on next start",
network.name, network.network_id
);
}
Ok(())
}
async fn forget_node_command(
socket: &std::path::Path,
network: &str,
node: &str,
) -> Result<(), Box<dyn std::error::Error>> {
let report = tsunagi::ipc::request_status(socket)
.await
.map_err(|err| format!("a reachable running agent is required to forget nodes: {err}"))?;
let entries: Vec<_> = report
.networks
.iter()
.map(|n| (n.network_id.clone(), n.name.clone()))
.collect();
let network = &report.networks[resolve_network_index(&entries, network)?];
if !network.active {
return Err(format!(
"start `{}` first: tsng network start {}",
network.name, network.network_id
)
.into());
}
let member = tsunagi::ipc::resolve_node(network, node)?;
let result = tsunagi::ipc::forget_node(
socket,
&network.network_id,
&member.endpoint_id,
member.record_version,
)
.await?;
println!(
"forgot {} ({}) claim v{} in `{}`; cleanup queued to {} peers. A newer signed claim can return.",
member.hostname.as_deref().unwrap_or("unnamed"),
result.peer,
result.record_version,
network.name,
result.peers_told
);
Ok(())
}
/// `tsng 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()?;
@@ -1912,6 +2114,9 @@ async fn network_command(args: NetworkArgs) -> Result<(), Box<dyn std::error::Er
let socket = control_socket(&paths, args.control_socket.as_ref(), is_custom_state);
let paths = agent_paths(paths, &socket, is_custom_state).await;
match args.action {
Some(NetworkAction::ForgetNode { network, node }) => {
forget_node_command(&socket, &network, &node).await
}
None => show_networks(&paths, &socket).await,
Some(NetworkAction::Broadcast { network, choice }) => {
broadcast_command(&paths, &socket, &network, choice).await
@@ -4445,43 +4650,24 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
// What the user asked for is checked first, before anything that could
// fail for a reason of its own: a misspelled protocol or setting is
// their mistake to see, not something to bury under a privilege error.
let wanted: Vec<&ProtocolSpec> = {
let mut wanted = Vec::new();
for name in args
.protocols
.iter()
.filter(|name| !name.eq_ignore_ascii_case("none"))
{
// Accept "wireguard" as a convenience alias for the canonical "wg".
let canonical = if name.eq_ignore_ascii_case("wireguard") {
WG_PROTOCOL
} else {
name.as_str()
};
let Some(spec) = PROTOCOLS.iter().find(|spec| spec.name == canonical) else {
let known: Vec<&str> = PROTOCOLS.iter().map(|spec| spec.name).collect();
return Err(format!(
"this build has no protocol called `{name}`; it has {}. \
Run `tsng protocols` to see what each one takes.",
known.join(", ")
)
.into());
};
wanted.push(spec);
}
wanted
};
let initial = parse_protocol_list(&args.protocols.join(","))?;
let wanted: Vec<&ProtocolSpec> = initial
.iter()
.filter_map(|name| PROTOCOLS.iter().find(|spec| spec.name == name))
.collect();
config.default_protocols = Some(initial);
let settings = parse_settings(&args.protocol_options)?;
// A setting nobody takes is a mistake, not a preference: one that was
// silently dropped looks exactly like one that did not work.
for setting in &settings {
if !wanted
if !PROTOCOLS
.iter()
.filter(|spec| setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name))
.any(|spec| !settings_for(spec, std::slice::from_ref(setting)).is_empty())
{
return Err(format!(
"no selected protocol takes `{}`; run `tsng protocols` to see what they do",
"no configured protocol takes `{}`; qualify disabled protocol settings as protocol:key=value. Run `tsng protocols` for details",
setting.key
)
.into());
@@ -4497,7 +4683,7 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
let mut tcp_tls_plugin = None;
let mut pure_wg_plugin: Option<Arc<PureWgPlugin>> = None;
let mut pure_wg_transport: Option<Arc<PureWgTransport>> = None;
if !wanted.is_empty() {
if !PROTOCOLS.is_empty() {
let tun_factory: Arc<dyn TunFactory> = if args.no_tun {
Arc::new(MemoryTunFactory::new())
} else {
@@ -4507,8 +4693,17 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
config = config.with_interface(tun_factory, args.interface.clone(), mtu);
}
for spec in &wanted {
let options = settings_for(spec, &settings);
for spec in PROTOCOLS {
// Unqualified options keep targeting the initial selection; explicitly
// scoped options also configure protocols that will be enabled later.
let scoped: Vec<_> = settings
.iter()
.filter(|setting| {
setting.protocol.is_some() || wanted.iter().any(|p| p.name == spec.name)
})
.cloned()
.collect();
let options = settings_for(spec, &scoped);
match spec.name {
AWG_PROTOCOL => {
let mut wg = WireguardConfig::new(paths.state_dir.join("awg"));
@@ -5112,6 +5307,7 @@ async fn build_report(
};
NetworkReport {
protocols: network.protocols.clone(),
auto_start: network.auto_start,
exit: exit_report(network),
broadcast: network.broadcast,
@@ -5138,6 +5334,7 @@ async fn build_report(
.members
.iter()
.map(|member| MemberReport {
record_version: member.record_version,
endpoint_id: member.endpoint_id.to_string(),
overlay_address_v4: member.overlay_address_v4.map(|addr| addr.to_string()),
// What this agent is currently experiencing trying to
@@ -5182,6 +5379,13 @@ async fn build_report(
.collect();
StatusReport {
available_protocols: PROTOCOLS
.iter()
.map(|p| tsunagi::ipc::ProtocolReport {
name: p.name.into(),
summary: p.summary.into(),
})
.collect(),
endpoint_id: status.endpoint_id.to_string(),
hostname: status.hostname.clone(),
bound_sockets: status
@@ -5536,6 +5740,7 @@ mod status_tests {
/// The situation that prompted this: one peer left and came back.
fn network_after_a_peer_returned() -> NetworkReport {
NetworkReport {
protocols: vec!["wg-quic".into()],
auto_start: true,
exit: Default::default(),
hostname_conflict: None,
@@ -5559,6 +5764,7 @@ mod status_tests {
}],
members: vec![
MemberReport {
record_version: 1,
endpoint_id: OWN.into(),
overlay_address_v4: Some("10.13.37.69".into()),
failed_dials: 0,
@@ -5566,6 +5772,7 @@ mod status_tests {
..Default::default()
},
MemberReport {
record_version: 1,
endpoint_id: ONLINE.into(),
overlay_address_v4: Some("10.13.37.237".into()),
failed_dials: 0,
@@ -5842,6 +6049,7 @@ mod status_tests {
fn offline_members_are_listed_after_online_ones() {
let mut network = network_after_a_peer_returned();
network.members.push(MemberReport {
record_version: 1,
endpoint_id: AWAY.into(),
overlay_address_v4: Some("10.13.37.99".into()),
failed_dials: 9,
@@ -5963,6 +6171,7 @@ mod status_tests {
fn an_offline_member_keeps_the_last_name_it_announced() {
let mut network = network_after_a_peer_returned();
network.members.push(MemberReport {
record_version: 1,
endpoint_id: AWAY.into(),
overlay_address_v4: Some("10.13.37.99".into()),
failed_dials: 0,
@@ -5984,6 +6193,7 @@ mod status_tests {
// are attributed to it instead of floating free as a counter.
let mut network = network_after_a_peer_returned();
network.members.push(MemberReport {
record_version: 1,
endpoint_id: AWAY.into(),
overlay_address_v4: Some("10.13.37.99".into()),
failed_dials: 9,
@@ -6008,6 +6218,7 @@ mod status_tests {
// and said to be, rather than being shown as away or as a fault.
let mut network = network_after_a_peer_returned();
network.members.push(MemberReport {
record_version: 1,
endpoint_id: AWAY.into(),
overlay_address_v4: Some("10.13.37.99".into()),
hostname: Some("laptop".into()),
@@ -6219,6 +6430,7 @@ mod network_tests {
NetworkSecret::from_bytes(&[secret.as_bytes(), &[0u8; 32]].concat()[..32]).unwrap();
let keys = NetworkKeys::derive(&name, &secret);
StoredNetwork {
protocols: None,
exit_node: false,
exit_via: None,
broadcast: true,
@@ -6442,6 +6654,7 @@ mod network_context_tests {
let secret = NetworkSecret::from_bytes([seed; 32]).unwrap();
let keys = NetworkKeys::derive(&name, &secret);
StoredNetwork {
protocols: None,
exit_node: false,
exit_via: None,
broadcast: true,
+267
View File
@@ -54,6 +54,19 @@ fn start_bare(port: u16) -> Running {
}
fn start_bare_range(port: u16, range: Option<&str>) -> Running {
start_bare_config(port, range, None)
}
fn start_bare_config(port: u16, range: Option<&str>, protocols: Option<&str>) -> Running {
start_bare_peers(port, range, protocols, &[])
}
fn start_bare_peers(
port: u16,
range: Option<&str>,
protocols: Option<&str>,
peers: &[String],
) -> Running {
let dir = TempDir::new().unwrap();
let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_tsng"));
command
@@ -71,6 +84,12 @@ fn start_bare_range(port: u16, range: Option<&str>) -> Running {
if let Some(range) = range {
command.args(["--ipv4-range", range]);
}
if let Some(protocols) = protocols {
command.args(["--protocol", protocols]);
}
for peer in peers {
command.args(["--peer", peer]);
}
let child = command.spawn().expect("the agent binary starts");
let mut agent = Running { child, dir };
let runtime = tokio::runtime::Builder::new_current_thread()
@@ -112,6 +131,254 @@ fn start(network: &str, port: u16) -> Running {
agent
}
#[test]
fn cli_forgets_only_the_selected_offline_claim_over_real_control() {
let cleaner = start("forget-cli", 0);
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let socket = tsunagi::ipc::control_socket_path(&cleaner.dir.path().join("state"));
let read = || {
runtime
.block_on(tsunagi::ipc::request_status(&socket))
.unwrap()
};
let before = read();
let bootstrap = format!("{}@{}", before.endpoint_id, before.bound_sockets[0]);
let target = start_bare_peers(0, None, Some("none"), &[bootstrap]);
assert!(
target
.run(&["id", "hostname", "old-cli-node"])
.status
.success()
);
assert!(
target
.run(&[
"join",
"--network",
"forget-cli",
"--secret",
"a-secret-for-the-cli-test"
])
.status
.success()
);
runtime.block_on(async {
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let status = tsunagi::ipc::request_status(&socket).await.unwrap();
if status.networks[0]
.members
.iter()
.any(|m| m.hostname.as_deref() == Some("old-cli-node") && m.online)
{
break;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap();
});
assert!(
!cleaner
.run(&["network", "forget-node", "forget-cli", "old-cli-node"])
.status
.success()
);
assert!(
!cleaner
.run(&["network", "forget-node", "forget-cli", &before.endpoint_id])
.status
.success()
);
assert!(
target
.run(&["network", "stop", "forget-cli"])
.status
.success()
);
runtime.block_on(async {
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let status = tsunagi::ipc::request_status(&socket).await.unwrap();
if status.networks[0]
.members
.iter()
.any(|m| m.hostname.as_deref() == Some("old-cli-node") && !m.online)
{
break;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap();
});
let forgotten = cleaner.run(&["network", "forget-node", "forget-cli", "old-cli-node"]);
assert!(
forgotten.status.success(),
"{}",
String::from_utf8_lossy(&forgotten.stderr)
);
assert!(String::from_utf8_lossy(&forgotten.stdout).contains("newer signed claim can return"));
assert!(
!read().networks[0]
.members
.iter()
.any(|m| m.hostname.as_deref() == Some("old-cli-node"))
);
assert!(
!cleaner
.run(&["network", "forget-node", "forget-cli", "old-cli-node"])
.status
.success()
);
}
#[test]
fn protocols_absent_from_startup_can_be_enabled_and_changed_offline() {
let mut agent = start_bare_config(0, None, Some("none"));
assert!(
agent
.run(&[
"join",
"--network",
"from-control-only",
"--secret",
"a-control-only-test-secret"
])
.status
.success()
);
let initial = agent.run(&["protocols", "from-control-only"]);
assert!(initial.status.success());
assert!(String::from_utf8_lossy(&initial.stdout).contains("protocols none"));
let enabled = agent.run(&["protocols", "from-control-only", "awg"]);
assert!(
enabled.status.success(),
"{}",
String::from_utf8_lossy(&enabled.stderr)
);
assert_eq!(
read_networks(&agent.dir)[0].protocols,
Some(vec!["awg".into()])
);
agent.child.kill().unwrap();
agent.child.wait().unwrap();
let saved = agent.run(&["protocols", "from-control-only", "wg-quic,awg"]);
assert!(
saved.status.success(),
"{}",
String::from_utf8_lossy(&saved.stderr)
);
assert!(String::from_utf8_lossy(&saved.stdout).contains("agent stopped"));
assert_eq!(
read_networks(&agent.dir)[0].protocols,
Some(vec!["wg-quic".into(), "awg".into()])
);
}
#[test]
fn protocols_are_changed_live_per_network_and_saved_without_changing_the_service() {
let agent = start("protocol-choice", 0);
assert!(
agent
.run(&[
"join",
"--network",
"other-choice",
"--secret",
"other-secret-for-cli-test"
])
.status
.success()
);
let before = read_networks(&agent.dir);
let other_id = before
.iter()
.find(|n| n.name.as_str() == "other-choice")
.unwrap()
.network_id;
let changed = agent.run(&["protocols", "protocol-choice", "awg,wg-quic"]);
assert!(
changed.status.success(),
"{}",
String::from_utf8_lossy(&changed.stderr)
);
let shown = agent.run(&["protocols", "protocol-choice"]);
assert!(shown.status.success());
assert!(String::from_utf8_lossy(&shown.stdout).contains("awg,wg-quic"));
let saved = read_networks(&agent.dir);
assert_eq!(
saved
.iter()
.find(|n| n.name.as_str() == "protocol-choice")
.unwrap()
.protocols,
Some(vec!["awg".into(), "wg-quic".into()])
);
assert_eq!(
saved
.iter()
.find(|n| n.network_id == other_id)
.unwrap()
.protocols,
before
.iter()
.find(|n| n.network_id == other_id)
.unwrap()
.protocols
);
for invalid in ["unknown", "awg,awg", "none,awg"] {
assert!(
!agent
.run(&["protocols", "protocol-choice", invalid])
.status
.success()
);
}
assert!(!agent.run(&["protocols", "missing", "awg"]).status.success());
let disabled = agent.run(&["protocols", "protocol-choice", "none"]);
assert!(
disabled.status.success(),
"{}",
String::from_utf8_lossy(&disabled.stderr)
);
assert_eq!(
read_networks(&agent.dir)
.iter()
.find(|n| n.name.as_str() == "protocol-choice")
.unwrap()
.protocols,
Some(Vec::new())
);
assert!(
agent
.run(&["network", "stop", "protocol-choice"])
.status
.success()
);
assert!(
agent
.run(&["protocols", "protocol-choice", "wg-quic"])
.status
.success()
);
assert!(
agent
.run(&["network", "start", "protocol-choice"])
.status
.success()
);
assert!(
String::from_utf8_lossy(&agent.run(&["protocols", "protocol-choice"]).stdout)
.contains("wg-quic")
);
}
#[test]
fn joining_with_no_secret_makes_one_and_prints_what_to_send() {
// The hurry case: a network with somebody, for as long as it is needed,
+15
View File
@@ -8,6 +8,21 @@ tsng up --protocol awg
tsng up --protocol wg,awg,wg-quic,tcp-tls
```
The startup list is the initial per-network preference. A saved network choice
overrides it and can enable any built-in protocol at runtime:
```sh
tsng protocols home awg,wg-quic
tsng protocols home awg
```
Changes are local to this device, persist across restarts, and retire existing
data links and end-to-end channels for that network before renegotiation.
Authenticated control sessions, membership, addresses and other networks stay
live. Removing every protocol (`none` in the CLI, all checkboxes off in the tray)
leaves the network control-only. This is different from automatic carrier
fallback, which preserves the selected end-to-end channel.
`awg` is Tsunagi AWG wire version 1. Its packet masking engine is adapted from
[RunaRoo/AmneziaWG-RS](https://github.com/RunaRoo/AmneziaWG-RS), with the pinned
source and license recorded in
+9 -4
View File
@@ -9,14 +9,18 @@ Two versions exist and are independent:
- **Identity scheme**, `tsunagi-network-id-v1`. Frozen. Changing it creates a
different network space for the same name and secret.
- **Control protocol**, ALPN `tsunagi/ctrl/6`, `PROTOCOL_VERSION = 6`.
- **Control protocol**, ALPN `tsunagi/ctrl/7`, `PROTOCOL_VERSION = 7`.
Upgrading the crate or bumping the control protocol must never change an
existing `NetworkId`.
Control version 6 adds private transport negotiation for pairwise AWG and data
delivery checks. Data ALPN remains `tsunagi/data/5`. All members must upgrade
together; network ids and stored state are unchanged. See [amnezia.md](amnezia.md).
Control version 7 adds signed offline-node retirement messages to version 6's
private AWG transport negotiation and delivery checks. Data ALPN remains
`tsunagi/data/5`. All members must upgrade together; network ids, handshake
encoding and existing record signatures are unchanged. Local CLI/tray IPC is
version 21, so restart the agent with matching clients. SQLite schema 9 adds
retirement storage while preserving existing identities, settings and records.
See [sync-model.md](sync-model.md#offline-node-cleanup) and [amnezia.md](amnezia.md).
## Canonical encoding
@@ -238,6 +242,7 @@ not affect other networks.
| `Ping { seq, payload }` | small request used to verify the exchange |
| `Pong { seq, payload }` | the echoed reply |
| `State { records }` | a snapshot of signed records, merged into what the receiver holds |
| `Retirements { markers }` | signed cleanup markers with the target's original signed claim as proof |
| `Peers { peers }` | unverified member address hints, requiring their own handshake |
| `Reachable { links }` | sender's current direct data links, by kind of transport |
| `Bye { reason }` | graceful goodbye; not a revocation of anything |
+1 -1
View File
@@ -124,5 +124,5 @@ counts sends without storing frames. Results are CPU forwarding cost, **not**
end-to-end network latency or VPN throughput; encryption, fragmentation,
sockets, congestion and scheduling contribute separately.
Wire compatibility: control ALPN `tsunagi/ctrl/6`, data ALPN `tsunagi/data/5`.
Wire compatibility: control ALPN `tsunagi/ctrl/7`, data ALPN `tsunagi/data/5`.
Upgrade every participant together; saved identities and network state persist.
+46 -27
View File
@@ -1,18 +1,9 @@
# Planned state synchronisation
# State synchronisation
**The first slice of this model is now implemented**, in `crates/tsunagi-core/src/state/`, and is
used for one thing: IPv4 overlay addresses. What follows describes the whole
model; the section at the end says exactly which parts exist.
The proof of concept still exchanges hostname and capability announcements
over live sessions and keeps no replicated history for those, which is why
WireGuard peer *membership* remains session-scoped even though a peer's
*address* no longer is. This file records the intended direction so the module
boundaries in [architecture.md](architecture.md) stay compatible with it, and so
nobody mistakes the current announcements for synchronisation.
There is no fake "ready CRDT" here and no snapshots that are not actually
verified.
Signed claims in `crates/tsunagi-core/src/state/` preserve IPv4 reservations and
hostnames across an absence. Signed releases and offline-node cleanup markers
free those reservations durably. Capabilities, links and authenticated sessions
remain volatile; a stored claim never authenticates a live connection.
## The problem
@@ -75,6 +66,39 @@ Turning a computer off is not a revocation of its hostname and does not remove
the participant. Revocations are not dropped merely because they are old, and
no acknowledgement from offline peers is required to keep working.
## Offline-node cleanup
Any authenticated member can sign a `SignedRetirement` for another member's
observed `Claim`. It includes the original signed claim as proof and uses the
separate `tsunagi-node-retirement/v1` signing domain. Both signatures and both
network scopes are checked. No owner, vote or quorum is involved, and no member
can fabricate the target's release or invent a future version cutoff.
The strongest observed target version wins; ties use a deterministic canonical
byte order. Claims at or below that version do not occupy an address, hostname
or visible member row. Original signed records remain stored, including their
range, so cleanup does not renumber the network. Markers remain even after a
newer claim returns: absence from a snapshot is never deletion and replaying an
old snapshot cannot restore the retired claim.
The local `Agent::forget_node` API checks the exact observed version again and
refuses self, online nodes, the selected exit node and missing/changed claims.
CLI and tray share this API. A returning live node can sign a newer claim and
allocate against the current holders; it resets the old hostname priority, so
it cannot reclaim a reused name merely by keeping its old timestamp. Cleanup
is not a permanent identity ban or a change of secret.
The marker and requester's own author counter commit in one SQLite transaction
before publication. Receivers persist a verified marker before applying it.
Invalid proofs are rejected individually. Periodic bounded marker batches
rotate through each peer's send queue, including when a queue fills, so a
replica that missed an update can catch up without the requester being online.
One strongest marker per target is retained, with a default limit of 1024 per
network and 1024 bytes per encoded marker. New targets beyond that limit are
refused rather than evicting resurrection protection. Corrupt mandatory marker
storage or a configured limit smaller than the stored set is an explicit error.
## Storage requirement this creates
When signed records land, writing the event and bumping the author's own
@@ -104,9 +128,9 @@ Implemented, in `crates/tsunagi-core/src/state/`:
* the merge rules above: higher version wins, an older version never rolls
back a newer one, duplicates are idempotent, a same-version conflict is
resolved identically on every replica and reported;
* a release tombstone, which merges correctly and is not undone by a replica
that has not heard of it — though nothing emits one yet, so freeing an
address still means forgetting the network;
* a release tombstone, emitted when leaving or replacing the device key and
not undone by a replica that has not heard of it;
* durable hostnames and equal-rights offline-node retirement described above;
* persistence in `state.sqlite`, with the record and the author's own version
counter committed in **one transaction before the record is announced**;
* distribution as a `State` control message, merged into what the receiver
@@ -114,21 +138,16 @@ Implemented, in `crates/tsunagi-core/src/state/`:
* allocation of a free IPv4 address against what everybody else holds, which
is what makes an address stable across an absence.
Deliberately not implemented: compaction, revoking a whole author, record
types beyond addressing, and any bound on how large a snapshot may grow
beyond the per-message limit.
Deliberately not implemented: compaction and permanent author revocation.
Claims have a per-message bound; retirement history also has a per-network bound.
## Future tests
These are **not implemented and must not be reported as passing**:
- snapshot merge against the rules above, including conflicting same-version
records;
- revocation propagation and resistance to resurrection after compaction;
- resistance to resurrection after compaction (compaction is not implemented);
- long network partitions and rejoin after an extended absence;
- hostname rename with atomic revoke-and-announce;
- recovery of a returning participant from a replica that is not the author;
- NAT traversal and hole punching between real hosts;
- relay fallback behaviour against a self-hosted relay;
- multi-process and multi-host deployment, as opposed to several library
instances inside one test process.
- multi-host deployment. CLI tests do run separate local agent processes;
library tests run several independent instances inside one process.
+28 -4
View File
@@ -164,10 +164,9 @@ paused clock. The public Mainline/relay smoke test is ignored by default.
## Not covered, and not claimed to be
Listed in [sync-model.md](sync-model.md#future-tests): snapshots, revocations,
long partitions, hostname renames, recovery of a returning participant, NAT
traversal, relay fallback, and multi-process or multi-host deployment. None of
these are implemented, and none are marked as passing.
Listed in [sync-model.md](sync-model.md#future-tests): compaction, long partitions,
NAT traversal, relay fallback and multi-host deployment. These scenarios are
not established by the default loopback suite.
## Debugging a test
@@ -184,3 +183,28 @@ without substituting iroh, membership, encryption or SQLite. It checks runtime
loss detection, fallback and fresh profiles, in addition to restart and input
validation. Tests are offline; public STUN/DPI/NAT effectiveness needs separate
real-network validation.
Runtime protocol selection tests switch real AWG and iroh/QUIC carriers and
end-to-end WireGuard tunnels from either endpoint-id role, carry full-size
packets, keep another active network's tunnels unchanged, disable and re-enable
the data plane without losing control sessions, and preserve selection after
restart. CLI tests use a real agent process and control socket to exercise
selection by network name, stopped networks, invalid lists, and persisted
control-only configuration, offline updates, and enabling a protocol absent
from the startup list. A real AWG test starts without a TUN, creates the single
memory interface on enablement, and races rapid changes against pending opens.
SQLite tests migrate schema 7 to 8 without changing
identity, secrets or signed records, preserve order and explicit empty choices,
and reject malformed stored selections. These checks use memory TUNs and do
not establish GUI rendering or behavior on an actual OS interface.
`tests/forget_node.rs` uses real agents, handshakes and SQLite to check equal
cleanup rights, network isolation, online/self/stale-version refusal, restart,
recovery of a stale replica, reuse of a hostname and a returning identity's
newer claim. Bounded batch delivery is checked with a two-message send queue
and twelve retained markers. Retirement unit tests check signatures, proof tampering, invented
future cutoffs, bounds, deterministic merge and resistance to stale snapshots.
SQLite tests migrate schema 8 to 9 and inject counter-write failure to prove
atomic rollback. The CLI test starts two real agent processes and exercises
offline cleanup by network and hostname through the control socket.
Tray rendering and physical screen sizing are not exercised automatically.
+12 -2
View File
@@ -38,6 +38,14 @@ Read this before relying on anything here. The protocol is in
arbitrarily many identities, flood the network with records and collide with
other participants' names. This is why a majority is not a root of trust.
Signatures protect authorship; they do not make a participant honest.
- **Cleanup rights are equal.** Any member can retire an observed signed claim
of another device, freeing its hostname and IPv4 reservation. A valid original
claim bounds the cutoff; a member cannot invent a future cutoff or forge the
target's release. The local API refuses online nodes, but a remote signature
cannot prove that a target was offline. A live target may publish a newer
claim; a malicious member can repeatedly disrupt reservations. Cleanup is
not expulsion or secret revocation. See
[sync-model.md](sync-model.md#offline-node-cleanup).
- **Weak secrets.** This targets high-entropy secrets. There is no PAKE, so a
short human passphrase can be guessed offline by anyone who can reach the
handshake. Use `NetworkSecret::generate()`.
@@ -55,8 +63,10 @@ Read this before relying on anything here. The protocol is in
having iroh under it does not make it so.
- **A cloned state directory is a cloned identity.** `state.sqlite` holds the
device secret key and the network secrets. Copying it copies the participant.
Restoring an old backup rolls the agent's state back, which — once signed
records exist — can resurrect revoked information or replay stale versions.
Restoring an old backup rolls the agent's state back and can replay stale
versions until it meets a replica holding the newer signed state. Retained
cleanup markers prevent old claims from resurrecting after that merge; they
cannot help if every copy of a marker has been lost.
- **A compromised host.** The secret is on disk to survive restarts. File
permissions are owner-only where the platform supports it, and the state
directory takes an ownership lock, but neither defends against a user who can