diff --git a/AGENTS.md b/AGENTS.md index 694fb69..b86cdae 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,6 +36,14 @@ Keep these separate. Crossing them is the main thing to review for. key. Outbound packets are routed to the owner of the destination address; inbound packets are dropped unless their source is the address derived for the peer that sent them. Never trust an address a peer announces. +- **Signed state is the only durable agreement.** A fact that must survive a + participant being away goes in `src/state/` as a record signed by its + author, never in a session. Merging is deterministic, an older version never + rolls back a newer one, and absence from a snapshot is not deletion. Never + add a vote or a quorum: a majority is not a trust root here, and it would + stall with one peer online. +- **A record and the author's own version counter commit together**, in one + transaction, **before** the record is announced. - **Plugins own their system objects.** A plugin creates and removes its own interface and nothing else. Never touch routing, DNS or firewall settings. - **Device identity vs network identity.** The iroh endpoint id is the device's @@ -97,6 +105,7 @@ Keep these separate. Crossing them is the main thing to review for. | `src/proto/` | framing, message formats, membership handshake | | `src/net.rs` | iroh endpoint adapter and observability snapshots | | `src/agent/` | agent lifecycle, per-network runtimes, sessions, events, status | +| `src/state/` | signed records that outlive a session, their merge rules and address allocation | | `src/dataplane/` | the plugin contract, the packet transport, and the WireGuard plugin | | `src/bin/tsunagi.rs`| the command line agent; the only place that owns a runtime, a logger and signals | | `tests/` | integration tests; `tests/common/` is the shared harness | diff --git a/README.md b/README.md index f60204a..23567d4 100644 --- a/README.md +++ b/README.md @@ -129,33 +129,21 @@ ping 100.65.243.53 IPv6 works out of the box: each member's address is derived from the network id and collides with essentially nothing. -**IPv4 is opt-in**, because no IPv4 range is free on every host — -`100.64.0.0/10` is Tailscale's, `10.0.0.0/8` and `192.168.0.0/16` are -everywhere, `172.17.0.0/16` is Docker. Name one you know is unused, the same -one on every member: +**IPv4 addresses are allocated and then remembered.** The default range is +`10.13.37.0/24`; the first member to join settles it and later members adopt +what they find, so `--ipv4-range` only matters for whoever starts the network: ```bash -tsunagi up --network lab --secret "$SECRET" --wireguard --ipv4-range 10.77.0.0/16 +tsunagi up --network lab --secret "$SECRET" --wireguard --ipv4-range 10.44.0.0/16 +tsunagi up --network lab --secret "$SECRET" --wireguard --ipv4-range none # IPv6 only ``` -The range is part of how addresses are derived, so members configured -differently would misroute. It travels in the announcement purely so a -mismatch is reported instead: the offending peer gets no IPv4 and keeps -working over IPv6. See -[docs/wireguard.md](docs/wireguard.md#ipv4-alongside-ipv6). - -Notes: - -- Only one side needs `--peer`; the link is bidirectional. -- The default `--transport relay` uses iroh's public address lookup and relays, - so two machines behind NAT find each other. `--transport local` keeps everything - on the local network. See *How peers find each other* below — it is worth - understanding what gets published. -- Without a network interface, add `--no-tun`: the mesh, the data links and the - WireGuard handshakes all still run and are visible in the status output, only - traffic does not reach the operating system. That is the quickest way to - confirm the network forms. -- For real traffic, see *Running unprivileged* below. +An address is claimed with a record signed by that member's persistent device +key, stored, and merged between every replica. A member that disappears for a +month comes back to the same address, because the claim outlived the session. +No vote is involved — see +[docs/wireguard.md](docs/wireguard.md#ipv4-allocated-signed-and-kept) and +[docs/sync-model.md](docs/sync-model.md). ## Running unprivileged diff --git a/docs/architecture.md b/docs/architecture.md index 7a4422c..67456bb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -45,6 +45,7 @@ and the agent stays manageable. | `proto` | message format, handshake, membership proof, protocol limits | | `agent` | agent and per-network lifecycle, reconnect, in-process message routing | | `storage` | mandatory state and the separately recoverable cache | +| `state` | signed records that outlive a session, merged between replicas | | `dataplane::transport` | authenticated datagram links to peers; where reachability lives | | `dataplane` | the contract IP plugins implement, plus the WireGuard plugin | diff --git a/docs/protocol.md b/docs/protocol.md index dacd64f..628921e 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -172,8 +172,15 @@ not affect other networks. | `Announce { hostname, capabilities }` | this agent's hostname and IP-plugin capabilities | | `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 | | `Bye { reason }` | graceful goodbye; not a revocation of anything | +A `State` snapshot is merged, never substituted: an author missing from it is +left untouched. Each record carries its own signature, so a peer forwarding +somebody else's record cannot alter it, and a record that fails verification +is dropped without affecting the rest of the batch. See +[sync-model.md](sync-model.md). + `PluginCapability { protocol, version, enabled, data }` is opaque to the core: `data` is bounded and handed to the matching plugin unparsed. Nothing in it is ever treated as a shell command, filesystem path or OS setting. diff --git a/docs/sync-model.md b/docs/sync-model.md index 5e26c6e..3cc03d7 100644 --- a/docs/sync-model.md +++ b/docs/sync-model.md @@ -1,10 +1,13 @@ # Planned state synchronisation -**Nothing in this document is implemented.** The proof of concept exchanges -hostname and capability announcements over live sessions and keeps no -replicated history. That is also why WireGuard peer membership is -session-scoped today: a peer leaves the overlay when its control session ends, -because there is no agreed durable state to keep it. This file records the intended direction so the module +**The first slice of this model is now implemented**, in `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. @@ -72,6 +75,31 @@ migrations for this. - Anyone who knows the secret can author records, so a majority of records is not evidence of anything. +## What exists today + +Implemented, in `src/state/`: + +* signed records, one per author per network, each holding that author's + complete current statement rather than a delta; +* signing and verification with the persistent iroh device key, over a + length-prefixed canonical encoding; +* 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; +* 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 + already holds rather than replacing it; +* 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. + ## Future tests These are **not implemented and must not be reported as passing**: diff --git a/docs/testing.md b/docs/testing.md index dd2646c..f11008f 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -50,6 +50,15 @@ keeping the WireGuard identity, shutdown removing every interface, a forged overlay claim being rejected, and the core carrying the payload without interpreting it. +Unit tests in `src/state/` cover the signed record model directly: tampering +with any field breaks verification, a newer version wins while an older one +never rolls back, two authors claiming one address resolve the same way no +matter the merge order, one key used in two places is reported rather than +silently merged, a release survives a late-arriving old claim, a bad record in +a batch does not stop the rest, and allocation is deterministic, spread out, +walks past everything taken and reports a full range instead of handing out a +duplicate. + `tests/local_control.rs` covers the local control socket end to end: a client asking a running agent for status over a real Unix socket, a leftover socket file being replaced while a live one is not, and the derived socket path diff --git a/docs/wireguard.md b/docs/wireguard.md index b5913f9..6ca99d1 100644 --- a/docs/wireguard.md +++ b/docs/wireguard.md @@ -83,51 +83,52 @@ Two consequences matter: * a member's address is bound to its WireGuard public key, so address ownership can be checked locally rather than believed. -## IPv4 alongside IPv6 +## IPv4: allocated, signed, and kept -The overlay carries IPv4 as well, but **it is off unless you name a range**, -and every member must name the same one: - -```bash -tsunagi up --network lab --secret "$SECRET" --wireguard --ipv4-range 10.77.0.0/16 -``` - -Two reasons it has no default. - -**There is no IPv4 range that is free everywhere.** `100.64.0.0/10` is -Tailscale's and carrier-grade NAT's, `10.0.0.0/8` and `192.168.0.0/16` are on -half the networks in the world, `172.17.0.0/16` is Docker. Picking one -requires knowing what is already in use on every machine that will join, which -is the operator's knowledge, not ours. IPv6 needs none of this: a ULA derived -from the network id collides with essentially nothing. - -**The range is an input to the derivation.** Each agent computes every peer's -address itself, so two members configured with different ranges would derive -different addresses for each other and IPv4 would silently misroute. The range -therefore travels in the announcement — not as a request, and never trusted, -but so that a mismatch is *detected*. When it happens, the offending peer gets -no IPv4 address here, keeps working over IPv6, and the reason is reported: +IPv6 addresses are *derived*: a 64 bit interface identifier makes a collision +impossible in practice, so nobody has to agree on anything. IPv4 has nothing +like that room, so deriving would collide. Instead an address is **allocated +and then recorded as a signed fact**, using the model in +[sync-model.md](sync-model.md). ```text -! wireguard: peer SDsEb/WF is configured with the IPv4 overlay range - 10.81.0.0/16 but this agent uses 10.80.0.0/16; every member must use the - same one. That peer has no IPv4 address here and is reachable over IPv6 only. +default range 10.13.37.0/24 (override with --ipv4-range) +who decides the first member to claim; later ones adopt what they find +who signs the claiming member, with its persistent device key +where it is kept state.sqlite, and every replica that has seen it +what a return costs nothing: the old address is reclaimed ``` -**IPv4 addresses can also collide with each other.** A 64 bit interface -identifier makes an IPv6 collision impossible in practice; IPv4 has nothing -like that room. In a `/16` with 50 members the chance that two members derive -the same address is roughly 2%. A mesh with no coordinator cannot allocate -around it, so the collision is resolved instead: the member whose WireGuard -public key sorts lower keeps the address, a rule every member computes -identically and therefore agrees on without exchanging anything. The other -member has no IPv4 address and remains reachable over IPv6. Pick a roomy -range — a `/16` for a handful of machines, larger for more — and the odds stay -small. +How it works: -The honest summary: **IPv6 always works. IPv4 is opt-in, needs agreement, and -degrades predictably when it does not get it.** Allocating IPv4 properly needs -the agreed state described in [sync-model.md](sync-model.md). +1. On joining, an agent reads back the records it already had and learns more + from its peers. +2. If it already holds an address, it keeps it. **That is the whole point**: a + participant that was away for a month comes back to the address it signed + for, because the claim outlived the session. +3. Otherwise it picks a free one — starting from a position derived from its + own identity, so two newcomers rarely start in the same place — signs the + claim, commits it together with its version counter, and only then + announces it. +4. Every replica merges what it receives into what it has. An author missing + from a snapshot is left alone: absence is not deletion. + +**No vote is involved, deliberately.** Anyone who knows the network secret can +mint identities, so a majority proves nothing, and a quorum would stall with +one participant online and diverge across a partition. Two members who claim +the same address at the same moment are resolved by a rule both compute +identically — the lower endpoint id keeps it — and the loser simply allocates +again with a higher version. + +**The range is agreed, not configured per member.** `--ipv4-range` says what +this agent would use; a network that has already settled on something else +wins, and the agent adopts it. So the flag matters for whoever starts the +network and is harmless afterwards. Pass `--ipv4-range none` for an IPv6-only +overlay. + +A release tombstone exists in the record type and merges correctly, but +nothing emits one yet, so an address stays claimed until the network is +forgotten. ## Address ownership is enforced, not announced @@ -246,8 +247,10 @@ async fn main() -> Result<()> { * **Full mesh only.** Every member runs a tunnel to every other member. Routing through an intermediate participant is not implemented. -* **IPv4 is opt-in, must be agreed, and can collide.** See above. A proper - allocator needs agreed state. +* **Nothing frees an address yet.** The release record exists and merges, but + no command emits one. +* **A snapshot grows with the number of members ever seen**, and is capped per + message rather than compacted. * **No routes, DNS or firewall rules.** The plugin creates its interface and nothing else. Anything beyond the overlay `/64` is the operator's business. * **Membership is session-scoped.** A peer leaves the overlay when its control diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 3f07005..a86f8bb 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -184,7 +184,7 @@ impl Agent { // bound; overflow drops the request rather than stalling the plugin. if !inner.config.plugins.is_empty() { let (plugin_tx, plugin_rx) = mpsc::channel(64); - let context = PluginContext::new(plugin_tx); + let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id()); for plugin in &inner.config.plugins { plugin.attach(context.clone()); } @@ -311,6 +311,8 @@ impl Agent { plugins: self.inner.config.plugins.clone(), hostname: self.inner.hostname.clone(), transport: self.inner.transport.get().cloned(), + device_secret: self.inner.identity.signing_key(), + ipv4_range: self.inner.config.overlay_ipv4_range, }); networks.insert(network_id, handle); drop(networks); diff --git a/src/agent/network.rs b/src/agent/network.rs index 35be599..062218f 100644 --- a/src/agent/network.rs +++ b/src/agent/network.rs @@ -23,6 +23,8 @@ use crate::identity::{NetworkId, NetworkKeys}; 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::storage::Storage; use super::events::Event; @@ -114,6 +116,11 @@ pub(crate) struct RuntimeParams { pub(crate) discovery_interval: Duration, pub(crate) plugins: Vec, pub(crate) hostname: String, + /// Signing key for this agent's own records. + pub(crate) device_secret: iroh::SecretKey, + /// The IPv4 overlay range this agent would use, if the network has not + /// already settled on another one. + pub(crate) ipv4_range: Option, /// How data plane links are opened. `None` disables the data plane. pub(crate) transport: Option>, } @@ -194,6 +201,12 @@ struct Runtime { opening: HashSet<(EndpointId, String)>, link_results_tx: mpsc::Sender, link_results_rx: mpsc::Receiver, + /// Signed records, merged from every replica we have talked to. + state: StateSet, + /// Snapshots received while dispatching, handled on the next loop pass. + pending_state: Vec<(EndpointId, Vec)>, + /// The highest version this agent has ever published for this network. + own_version: u64, } impl Runtime { @@ -220,6 +233,9 @@ impl Runtime { opening: HashSet::new(), link_results_tx, link_results_rx, + state: StateSet::new(), + pending_state: Vec::new(), + own_version: 0, } } @@ -229,6 +245,10 @@ impl Runtime { } async fn run(&mut self, mut commands: mpsc::Receiver) { + // Everything this agent knew before it restarted, including the + // address it holds. + self.load_state().await; + let mut ticker = tokio::time::interval(self.params.discovery_interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -243,6 +263,7 @@ impl Runtime { event = self.session_events_rx.recv() => { if let Some(event) = event { self.handle_session_event(event).await; + self.drain_pending_state().await; } } result = self.dial_results_rx.recv() => { @@ -550,6 +571,219 @@ impl Runtime { } } + // ----------------------------------------------------------- agreed state + + /// Reads back what this agent already knew before it restarted. + /// + /// Records are verified again on load: the database is not a trust + /// boundary, because a restored backup or a copied file could hold + /// anything. + async fn load_state(&mut self) { + let stored = self + .params + .storage + .signed_records(self.network_id) + .await + .unwrap_or_default(); + let (_, errors) = self.state.merge_all(self.network_id, stored); + for err in errors { + tracing::warn!(%err, "discarding an unusable stored record"); + } + self.own_version = self + .params + .storage + .own_record_version(self.network_id) + .await + .unwrap_or(0); + + self.ensure_own_claim().await; + self.publish_allocations(); + } + + /// The range this network uses: whatever it has already settled on, else + /// what this agent was configured with. + /// + /// Adopting the agreed one is what lets a participant join without being + /// told the range out of band. + fn effective_range(&self) -> Option { + self.state.agreed_range().or(self.params.ipv4_range) + } + + /// Makes sure this agent holds an address, claiming one if it does not. + /// + /// Called after anything that could change the picture: startup, and + /// every time another replica's records arrive. + async fn ensure_own_claim(&mut self) { + let Some(range) = self.effective_range() else { + return; + }; + let holders = self.state.address_holders(); + let mine = self.state.address_of(&self.local_id); + + // An address we still hold is kept; this is what makes a returning + // participant get its old address back. + if let Some(mine) = mine + && range.contains(mine) + { + return; + } + + let taken: std::collections::HashSet = holders + .iter() + .filter(|(_, holder)| **holder != self.local_id) + .map(|(address, _)| *address) + .collect(); + + let wanted = match allocate( + self.network_id, + self.local_id, + range, + &taken, + self.state + .get(&self.local_id) + .and_then(|record| record.body.claimed_address()), + ) { + Ok(address) => address, + Err(err) => { + self.metrics.plugin_errors += 1; + self.emit(Event::PluginError { + network: self.network_id, + protocol: "overlay".into(), + reason: err.to_string(), + }); + return; + } + }; + + self.publish_record(RecordBody::Ipv4Claim { + address: wanted, + range, + }) + .await; + } + + /// 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 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 { + self.emit(Event::PluginError { + network: self.network_id, + protocol: "overlay".into(), + reason: format!("cannot store our own record: {err}"), + }); + return; + } + self.own_version = version; + + match self.state.merge(self.network_id, record) { + Ok(_) => {} + Err(err) => { + tracing::error!(%err, "our own record did not verify"); + return; + } + } + self.broadcast_state(); + } + + /// Handles snapshots collected while dispatching messages. + async fn drain_pending_state(&mut self) { + for (peer, records) in std::mem::take(&mut self.pending_state) { + self.receive_state(peer, records).await; + } + } + + /// Sends everything we know to every peer. + fn broadcast_state(&mut self) { + let mut records = self.state.records(); + records.truncate(self.params.limits.max_state_records); + if records.is_empty() { + return; + } + let message = ControlMessage::State { records }; + let peers: Vec = self.sessions.keys().copied().collect(); + for peer in peers { + if let Err(err) = self.send_to(peer, message.clone()) { + tracing::debug!(%err, "could not queue a state snapshot"); + } + } + } + + /// Merges a snapshot from a peer. + async fn receive_state(&mut self, peer: EndpointId, records: Vec) { + let before = self.state.records(); + let (outcomes, errors) = self.state.merge_all(self.network_id, records); + + for err in errors { + self.metrics.protocol_violations += 1; + self.emit(Event::ProtocolViolation { + network: Some(self.network_id), + peer: Some(peer), + reason: format!("unusable signed record: {err}"), + }); + } + for outcome in &outcomes { + if *outcome == Merged::Conflicted { + self.emit(Event::PluginError { + network: self.network_id, + protocol: "overlay".into(), + reason: "two different records from one author at the same version; \ + a device key appears to be in use in two places" + .into(), + }); + } + } + + let changed = outcomes.iter().any(|outcome| { + matches!( + outcome, + Merged::Added | Merged::Updated | Merged::Conflicted + ) + }); + if !changed { + return; + } + + for record in self.state.records() { + if record.author == *self.local_id.as_bytes() { + continue; + } + if let Err(err) = self.params.storage.put_signed_record(record).await { + tracing::debug!(%err, "cannot persist a record"); + } + } + + // Somebody may have taken the address we were using. + self.ensure_own_claim().await; + self.publish_allocations(); + if self.state.records() != before { + self.broadcast_state(); + } + } + + /// Tells the plugins who holds which overlay address. + fn publish_allocations(&mut self) { + let Some(range) = self.effective_range() else { + return; + }; + let mut allocations: Vec<(EndpointId, std::net::Ipv4Addr)> = self + .state + .address_holders() + .into_iter() + .map(|(address, holder)| (holder, address)) + .collect(); + allocations.sort_by_key(|(holder, _)| *holder.as_bytes()); + + for plugin in &self.params.plugins { + plugin.on_address_allocation(self.network_id, range, &allocations); + } + } + // ------------------------------------------------------------ data plane /// Protocol ids this agent has a plugin for. @@ -771,6 +1005,8 @@ impl Runtime { tracing::debug!(%err, "could not queue initial announcement"); } + self.broadcast_state(); + self.emit(Event::PeerConnected { network: self.network_id, peer, @@ -914,6 +1150,9 @@ impl Runtime { tracing::debug!(%err, "could not queue pong"); } } + ControlMessage::State { records } => { + self.pending_state.push((peer, records.clone())); + } ControlMessage::Pong { .. } | ControlMessage::Bye { .. } => {} } diff --git a/src/bin/tsunagi.rs b/src/bin/tsunagi.rs index f995822..3ddf544 100644 --- a/src/bin/tsunagi.rs +++ b/src/bin/tsunagi.rs @@ -16,11 +16,12 @@ use tsunagi::agent::Event; use tsunagi::config::{AgentConfig, StoragePaths, TransportPolicy}; use tsunagi::dataplane::IpPlugin; use tsunagi::dataplane::wireguard::{ - Ipv4Range, MemoryTunFactory, TunFactory, WireguardConfig, WireguardPlugin, + MemoryTunFactory, TunFactory, WireguardConfig, WireguardPlugin, }; use tsunagi::discovery::{CompositeDiscovery, NetworkDiscovery, StaticBootstrap}; use tsunagi::identity::{NetworkName, NetworkSecret}; use tsunagi::iroh_types::EndpointAddr; +use tsunagi::state::Ipv4Range; use tsunagi::{Agent, NetworkId}; /// A small agent for private mesh networks. @@ -104,24 +105,20 @@ struct TunSetupArgs { ipv4_range: Option, } -/// Strips the error type's own prefix, which is about peers rather than flags. -fn plain_reason(err: &tsunagi::dataplane::PluginError) -> String { - let text = err.to_string(); - text.split_once(": ") - .map(|(_, rest)| rest.to_string()) - .unwrap_or(text) -} - /// Resolves the IPv4 overlay range from the flag. +/// +/// Absent means the built-in default. A network that already settled on +/// another range wins over both. fn resolve_ipv4_range( range: Option<&String>, ) -> Result, Box> { match range { - Some(text) => Ok(Some(text.parse::().map_err(|err| { - // The underlying error type is about peers; reword it for a flag. - format!("--ipv4-range {text}: {}", plain_reason(&err)) - })?)), - None => Ok(None), + Some(text) if text.eq_ignore_ascii_case("none") => Ok(None), + Some(text) => Ok(Some( + text.parse::() + .map_err(|err| format!("--ipv4-range {text}: {err}"))?, + )), + None => Ok(Some(tsunagi::state::DEFAULT_IPV4_RANGE)), } } @@ -237,13 +234,13 @@ struct UpArgs { #[arg(long)] wg_mtu: Option, - /// Also run an IPv4 overlay in this range, as `address/prefix`. + /// IPv4 overlay range, as `address/prefix`, or `none` to disable IPv4. /// - /// Off unless given: no IPv4 range is free on every host. Pick one you - /// know is unused everywhere — not 100.64.0.0/10, which is Tailscale's - /// and carrier-grade NAT's. Every member must pass the same range; a - /// mismatch is detected and reported rather than silently misrouted. - /// IPv6 needs none of this and is always on. + /// Defaults to 10.13.37.0/24. Only the first member to join decides: + /// a network that has already settled on a range wins, and a joining + /// agent adopts what it finds. Addresses are allocated from it and + /// recorded in signed state, so each member keeps its own across + /// restarts and long absences. #[arg(long, value_name = "CIDR")] ipv4_range: Option, @@ -541,9 +538,6 @@ async fn up(args: UpArgs) -> Result<(), Box> { // Parsed up front so a typo is reported immediately, and so the option is // never silently ignored when the data plane is off. let ipv4_range = resolve_ipv4_range(args.ipv4_range.as_ref())?; - if ipv4_range.is_some() && !args.wireguard { - return Err("--ipv4-range only applies together with --wireguard".into()); - } let mut bootstrap: Vec = Vec::new(); for peer in &args.peers { @@ -555,6 +549,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { ])); let mut config = AgentConfig::new(paths.clone()) + .with_overlay_ipv4_range(ipv4_range) .with_transport(args.transport.into()) .with_discovery(discovery) .with_discovery_interval(Duration::from_secs(5)); @@ -573,8 +568,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { system_tun_factory()? }; let mut wg = WireguardConfig::new(paths.state_dir.join("wireguard")) - .with_interface_prefix(args.wg_prefix.clone()) - .with_ipv4_range(ipv4_range); + .with_interface_prefix(args.wg_prefix.clone()); if let Some(mtu) = args.wg_mtu { wg = wg.with_mtu(mtu); } diff --git a/src/config.rs b/src/config.rs index bcba814..a203fda 100644 --- a/src/config.rs +++ b/src/config.rs @@ -123,6 +123,8 @@ pub struct Limits { pub max_echo_payload_len: usize, /// Largest accepted free-text reason string, in bytes. pub max_reason_len: usize, + /// Largest number of signed records accepted in one snapshot. + pub max_state_records: usize, /// Deadline for the whole handshake. pub handshake_timeout: Duration, /// Deadline for one outbound dial attempt. @@ -156,6 +158,7 @@ impl Default for Limits { max_capability_data_len: 4 * 1024, max_echo_payload_len: 4 * 1024, max_reason_len: 256, + max_state_records: crate::state::MAX_RECORDS_PER_MESSAGE, handshake_timeout: Duration::from_secs(10), dial_timeout: Duration::from_secs(10), write_timeout: Duration::from_secs(30), @@ -234,6 +237,13 @@ pub struct AgentConfig { pub reconnect: ReconnectPolicy, /// IP plugins whose capabilities are announced and dispatched. pub plugins: Vec, + /// The IPv4 overlay range this agent proposes. + /// + /// Addresses are allocated from it and recorded in signed state, so a + /// participant keeps the same one across restarts. A network that has + /// already settled on another range wins: a joining agent adopts what it + /// finds rather than imposing this. + pub overlay_ipv4_range: Option, } impl AgentConfig { @@ -249,6 +259,7 @@ impl AgentConfig { limits: Limits::default(), reconnect: ReconnectPolicy::default(), plugins: Vec::new(), + overlay_ipv4_range: Some(crate::state::DEFAULT_IPV4_RANGE), } } @@ -291,6 +302,12 @@ impl AgentConfig { self } + /// Sets the IPv4 overlay range this agent proposes, or disables IPv4. + pub fn with_overlay_ipv4_range(mut self, range: Option) -> Self { + self.overlay_ipv4_range = range; + self + } + /// Registers an IP plugin. pub fn with_plugin(mut self, plugin: SharedPlugin) -> Self { self.plugins.push(plugin); diff --git a/src/dataplane/mod.rs b/src/dataplane/mod.rs index a9b5998..b357a06 100644 --- a/src/dataplane/mod.rs +++ b/src/dataplane/mod.rs @@ -94,18 +94,30 @@ pub(crate) enum PluginRequest { #[derive(Clone)] pub struct PluginContext { sender: Option>, + local: Option, } impl PluginContext { - pub(crate) fn new(sender: mpsc::Sender) -> Self { + pub(crate) fn new(sender: mpsc::Sender, local: EndpointId) -> Self { Self { sender: Some(sender), + local: Some(local), } } /// A context that discards everything, for plugins used outside an agent. pub fn detached() -> Self { - Self { sender: None } + Self { + sender: None, + local: None, + } + } + + /// This agent's own endpoint id, when the context is attached. + /// + /// A plugin needs it to find itself in the agreed allocation. + pub fn local_endpoint_id(&self) -> Option { + self.local } fn send(&self, request: PluginRequest) { @@ -148,6 +160,7 @@ impl std::fmt::Debug for PluginContext { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("PluginContext") .field("attached", &self.sender.is_some()) + .field("local", &self.local.map(|id| id.fmt_short().to_string())) .finish() } } @@ -197,6 +210,20 @@ pub trait IpPlugin: Send + Sync + std::fmt::Debug + 'static { capability: &PluginCapability, ) -> std::result::Result<(), PluginError>; + /// The overlay addresses the network has agreed on. + /// + /// Allocated rather than derived, and backed by the signed records in + /// [`crate::state`], so a participant keeps its address across restarts + /// and long absences. Called whenever the agreed picture changes. + fn on_address_allocation( + &self, + network: NetworkId, + range: crate::state::Ipv4Range, + allocations: &[(EndpointId, std::net::Ipv4Addr)], + ) { + let _ = (network, range, allocations); + } + /// A data plane link to a peer is available for this plugin's protocol. /// /// The plugin moves its packets over this link and never learns how the diff --git a/src/dataplane/wireguard/announcement.rs b/src/dataplane/wireguard/announcement.rs index e28f73c..6447a88 100644 --- a/src/dataplane/wireguard/announcement.rs +++ b/src/dataplane/wireguard/announcement.rs @@ -18,14 +18,15 @@ use crate::dataplane::PluginError; use crate::identity::NetworkId; use super::keys::WgPublicKey; -use super::overlay::{Ipv4Range, overlay_address}; +use super::overlay::overlay_address; /// Version of the announcement format. /// -/// Bumped to 2 when the IPv4 overlay range was added. postcard is not -/// self-describing, so an older peer cannot read a newer announcement; the -/// mismatch is reported rather than misparsed. -pub const ANNOUNCEMENT_VERSION: u16 = 2; +/// Version 3 dropped the IPv4 range again: overlay addressing moved to the +/// signed records in [`crate::state`], which carry the range and survive a +/// participant being away. postcard is not self-describing, so an older peer +/// cannot read a newer announcement; the mismatch is reported, not misparsed. +pub const ANNOUNCEMENT_VERSION: u16 = 3; /// What one participant advertises for the WireGuard data plane. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -39,12 +40,6 @@ pub struct WgAnnouncement { /// Carried for diagnostics and cross-checking only. Addresses are always /// derived locally, never taken from this field. pub overlay_address: Ipv6Addr, - /// The IPv4 overlay range this peer is configured with, if any. - /// - /// Not a request and not trusted: it exists so that two members who were - /// configured differently find out, instead of silently deriving - /// different addresses for each other and misrouting IPv4. - pub ipv4_range: Option, } /// A peer announcement that has been validated against a specific network. @@ -54,22 +49,15 @@ pub struct ValidatedAnnouncement { pub public_key: WgPublicKey, /// The overlay address derived locally for this key. Authoritative. pub overlay_address: Ipv6Addr, - /// The IPv4 overlay range the peer is configured with. - pub ipv4_range: Option, } impl WgAnnouncement { /// Builds this agent's announcement. - pub fn new( - network: NetworkId, - public_key: &WgPublicKey, - ipv4_range: Option, - ) -> Self { + pub fn new(network: NetworkId, public_key: &WgPublicKey) -> Self { Self { version: ANNOUNCEMENT_VERSION, public_key: *public_key.as_bytes(), overlay_address: overlay_address(network, public_key), - ipv4_range, } } @@ -127,18 +115,9 @@ impl WgAnnouncement { )); } - if let Some(range) = self.ipv4_range - && range.prefix_len > 30 - { - return Err(PluginError::Rejected(format!( - "announced IPv4 range {range} has no room for hosts" - ))); - } - Ok(ValidatedAnnouncement { public_key, overlay_address: derived, - ipv4_range: self.ipv4_range, }) } } @@ -166,7 +145,7 @@ mod tests { let peer = WgSecretKey::generate().public(); let local = WgSecretKey::generate().public(); - let payload = WgAnnouncement::new(id, &peer, None).encode().unwrap(); + let payload = WgAnnouncement::new(id, &peer).encode().unwrap(); let validated = WgAnnouncement::decode_and_validate(&payload, id, &local).unwrap(); assert_eq!(validated.public_key, peer); @@ -179,7 +158,7 @@ mod tests { // carried here, so there is nothing for a peer to lie about. let id = network("identity-only"); let peer = WgSecretKey::generate().public(); - let payload = WgAnnouncement::new(id, &peer, None).encode().unwrap(); + let payload = WgAnnouncement::new(id, &peer).encode().unwrap(); assert!( payload.len() < 80, "the announcement should stay tiny, got {} bytes", @@ -195,7 +174,7 @@ mod tests { let local = WgSecretKey::generate().public(); // An attacker claims the victim's overlay address with its own key. - let mut forged = WgAnnouncement::new(id, &attacker, None); + let mut forged = WgAnnouncement::new(id, &attacker); forged.overlay_address = overlay_address(id, &victim); let result = WgAnnouncement::decode_and_validate(&forged.encode().unwrap(), id, &local); @@ -212,7 +191,7 @@ mod tests { let peer = WgSecretKey::generate().public(); let local = WgSecretKey::generate().public(); - let payload = WgAnnouncement::new(there, &peer, None).encode().unwrap(); + let payload = WgAnnouncement::new(there, &peer).encode().unwrap(); assert!(WgAnnouncement::decode_and_validate(&payload, here, &local).is_err()); } @@ -227,7 +206,7 @@ mod tests { let wrong_version = WgAnnouncement { version: ANNOUNCEMENT_VERSION + 1, - ..WgAnnouncement::new(id, &peer, None) + ..WgAnnouncement::new(id, &peer) }; assert!( WgAnnouncement::decode_and_validate(&wrong_version.encode().unwrap(), id, &local) @@ -236,38 +215,18 @@ mod tests { let zero_key = WgAnnouncement { public_key: [0u8; 32], - ..WgAnnouncement::new(id, &peer, None) + ..WgAnnouncement::new(id, &peer) }; assert!( WgAnnouncement::decode_and_validate(&zero_key.encode().unwrap(), id, &local).is_err() ); } - #[test] - fn the_ipv4_range_travels_so_a_mismatch_can_be_seen() { - let id = network("ranges"); - let peer = WgSecretKey::generate().public(); - let local = WgSecretKey::generate().public(); - let range = Some(Ipv4Range::new("10.9.0.0".parse().unwrap(), 16).unwrap()); - - let payload = WgAnnouncement::new(id, &peer, range).encode().unwrap(); - let validated = WgAnnouncement::decode_and_validate(&payload, id, &local).unwrap(); - assert_eq!(validated.ipv4_range, range); - - // A range with no usable hosts is nonsense and is refused. - let mut bad = WgAnnouncement::new(id, &peer, range); - bad.ipv4_range = Some(Ipv4Range { - base: "10.9.0.0".parse().unwrap(), - prefix_len: 31, - }); - assert!(WgAnnouncement::decode_and_validate(&bad.encode().unwrap(), id, &local).is_err()); - } - #[test] fn a_peer_cannot_claim_our_own_key() { let id = network("self"); let local = WgSecretKey::generate().public(); - let payload = WgAnnouncement::new(id, &local, None).encode().unwrap(); + let payload = WgAnnouncement::new(id, &local).encode().unwrap(); assert!(WgAnnouncement::decode_and_validate(&payload, id, &local).is_err()); } @@ -275,7 +234,7 @@ mod tests { fn announcements_stay_well_under_the_capability_payload_limit() { let id = network("size"); let peer = WgSecretKey::generate().public(); - let payload = WgAnnouncement::new(id, &peer, None).encode().unwrap(); + let payload = WgAnnouncement::new(id, &peer).encode().unwrap(); assert!( payload.len() < crate::config::Limits::default().max_capability_data_len, "announcement is {} bytes", diff --git a/src/dataplane/wireguard/device.rs b/src/dataplane/wireguard/device.rs index d48b107..a11aae1 100644 --- a/src/dataplane/wireguard/device.rs +++ b/src/dataplane/wireguard/device.rs @@ -42,9 +42,10 @@ use crate::dataplane::transport::{SharedLink, TransportError}; use crate::identity::NetworkId; use super::keys::{WgPublicKey, WgSecretKey}; -use super::overlay::{Ipv4Range, overlay_address}; +use super::overlay::overlay_address; use super::packet::IpHeader; use super::tun::TunDevice; +use crate::state::Ipv4Range; /// How often WireGuard's own timers are driven. /// diff --git a/src/dataplane/wireguard/mod.rs b/src/dataplane/wireguard/mod.rs index 243a9d4..9530003 100644 --- a/src/dataplane/wireguard/mod.rs +++ b/src/dataplane/wireguard/mod.rs @@ -50,14 +50,12 @@ pub mod plugin; pub mod store; pub mod tun; +pub use crate::state::Ipv4Range; pub use announcement::{ValidatedAnnouncement, WgAnnouncement}; pub use config::{Cidr, DEFAULT_INTERFACE_PREFIX, MAX_INTERFACE_NAME_LEN, interface_name}; pub use device::{PeerHealth, PeerStats, PeerSummary, WireguardDevice}; pub use keys::{WgPublicKey, WgSecretKey}; -pub use overlay::{ - Ipv4Range, OVERLAY_PREFIX_LEN, RFC6598_SHARED_RANGE, overlay_address, overlay_address_v4, - overlay_prefix, -}; +pub use overlay::{OVERLAY_PREFIX_LEN, overlay_address, overlay_address_v4, overlay_prefix}; pub use packet::IpHeader; pub use plugin::{ DEFAULT_MTU, MIN_MTU, NetworkOverview, PeerOverview, WIREGUARD_OVERHEAD, WIREGUARD_PROTOCOL, diff --git a/src/dataplane/wireguard/overlay.rs b/src/dataplane/wireguard/overlay.rs index 8a6942a..ec75066 100644 --- a/src/dataplane/wireguard/overlay.rs +++ b/src/dataplane/wireguard/overlay.rs @@ -22,10 +22,9 @@ use std::net::{Ipv4Addr, Ipv6Addr}; -use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use crate::dataplane::PluginError; +use crate::state::Ipv4Range; use crate::identity::NetworkId; @@ -40,77 +39,6 @@ pub const OVERLAY_PREFIX_LEN: u8 = 64; /// Prefix length of one member's address inside the overlay. pub const OVERLAY_HOST_PREFIX_LEN: u8 = 128; -/// An IPv4 range the overlay can be derived into. -/// -/// Every member of a network must be configured with the same one, because -/// addresses are derived from it. See [`crate::dataplane::wireguard::plugin::WireguardConfig::ipv4_range`]. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub struct Ipv4Range { - /// Base address of the range. - pub base: Ipv4Addr, - /// Prefix length, at most 30 so there is room for hosts. - pub prefix_len: u8, -} - -impl Ipv4Range { - /// Builds a range, rejecting one with no room for hosts. - pub fn new(base: Ipv4Addr, prefix_len: u8) -> Result { - if prefix_len > 30 { - return Err(PluginError::Rejected(format!( - "a /{prefix_len} has no room for hosts; use /30 or larger" - ))); - } - Ok(Self { base, prefix_len }) - } - - /// Whether an address falls inside the range. - pub fn contains(&self, address: Ipv4Addr) -> bool { - let host_bits = 32 - u32::from(self.prefix_len); - let mask = if host_bits >= 32 { - 0 - } else { - u32::MAX << host_bits - }; - u32::from(address) & mask == u32::from(self.base) & mask - } -} - -impl std::fmt::Display for Ipv4Range { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}/{}", self.base, self.prefix_len) - } -} - -impl std::str::FromStr for Ipv4Range { - type Err = PluginError; - - fn from_str(text: &str) -> Result { - let (base, prefix) = text.split_once('/').ok_or_else(|| { - PluginError::Rejected(format!( - "`{text}` is not an address with a prefix, for example 10.77.0.0/16" - )) - })?; - let base = base.parse().map_err(|err| { - PluginError::Rejected(format!("`{base}` is not an IPv4 address: {err}")) - })?; - let prefix_len = prefix.parse().map_err(|err| { - PluginError::Rejected(format!("`{prefix}` is not a prefix length: {err}")) - })?; - Self::new(base, prefix_len) - } -} - -/// RFC 6598 shared address space, offered only as a reference point. -/// -/// **Not a default, and usually a bad choice.** Tailscale uses exactly this -/// range, and so does carrier-grade NAT, so a machine running either will -/// collide with it. There is no IPv4 range that is free on every host, which -/// is why the IPv4 overlay has no default at all and must be configured. -pub const RFC6598_SHARED_RANGE: Ipv4Range = Ipv4Range { - base: Ipv4Addr::new(100, 64, 0, 0), - prefix_len: 10, -}; - fn push_lp(out: &mut Vec, bytes: &[u8]) { let len = u32::try_from(bytes.len()).unwrap_or(u32::MAX); out.extend_from_slice(&len.to_be_bytes()); @@ -252,7 +180,7 @@ mod tests { #[test] fn ipv4_addresses_land_inside_the_range_and_avoid_its_edges() { let id = network("v4"); - let range = RFC6598_SHARED_RANGE; + let range: Ipv4Range = "100.64.0.0/10".parse().unwrap(); for byte in 0..64u8 { let key = WgPublicKey::from_bytes([byte; 32]); let addr = overlay_address_v4(id, &key, range).unwrap(); @@ -286,7 +214,7 @@ mod tests { let key = WgPublicKey::from_bytes([9u8; 32]); let first = network("one"); let second = network("two"); - let range = RFC6598_SHARED_RANGE; + let range: Ipv4Range = "100.64.0.0/10".parse().unwrap(); assert_eq!( overlay_address_v4(first, &key, range), diff --git a/src/dataplane/wireguard/plugin.rs b/src/dataplane/wireguard/plugin.rs index f52a063..238517f 100644 --- a/src/dataplane/wireguard/plugin.rs +++ b/src/dataplane/wireguard/plugin.rs @@ -43,11 +43,10 @@ use super::announcement::{ValidatedAnnouncement, WgAnnouncement}; use super::config::{DEFAULT_INTERFACE_PREFIX, interface_name}; use super::device::{PeerSummary, WireguardDevice}; use super::keys::{WgPublicKey, WgSecretKey}; -use super::overlay::{ - Ipv4Range, OVERLAY_PREFIX_LEN, overlay_address, overlay_address_v4, overlay_prefix, -}; +use super::overlay::{OVERLAY_PREFIX_LEN, overlay_address, overlay_prefix}; use super::store::WgKeyStore; use super::tun::{TunFactory, TunRequest}; +use crate::state::Ipv4Range; /// The protocol identifier this plugin announces. pub const WIREGUARD_PROTOCOL: &str = "wireguard"; @@ -88,21 +87,6 @@ pub struct WireguardConfig { pub keepalive: Option, /// Interface MTU. See [`DEFAULT_MTU`]. pub mtu: u32, - /// IPv4 overlay range, or `None` for an IPv6-only overlay. - /// - /// **Every member of a network must configure the same range.** Addresses - /// are derived from it, so two members configured differently would - /// derive different addresses for each other. The range travels in the - /// announcement purely so that such a mismatch is detected and reported - /// instead of silently misrouting. - /// - /// There is no default, because no IPv4 range is free on every host: - /// `100.64.0.0/10` belongs to Tailscale and to carrier-grade NAT, - /// `10.0.0.0/8` and `192.168.0.0/16` are everywhere, `172.17.0.0/16` is - /// Docker. Pick one you know is unused on every machine that will join. - /// IPv6 needs none of this: its addresses are derived from the network - /// id and never collide. - pub ipv4_range: Option, /// How long to coalesce changes before reconciling. pub reconcile_debounce: Duration, /// How often to reconcile anyway, which is also when a packet interface @@ -118,9 +102,6 @@ impl WireguardConfig { interface_prefix: DEFAULT_INTERFACE_PREFIX.to_string(), keepalive: Some(25), mtu: DEFAULT_MTU, - // Off by default: no IPv4 range is free on every host. See - // `with_ipv4_range`. - ipv4_range: None, reconcile_debounce: Duration::from_millis(200), reconcile_interval: Duration::from_secs(15), } @@ -140,14 +121,6 @@ impl WireguardConfig { self } - /// Sets the IPv4 overlay range, or disables IPv4 with `None`. - /// - /// Must match on every member; see the field documentation. - pub fn with_ipv4_range(mut self, range: Option) -> Self { - self.ipv4_range = range; - self - } - /// Sets the reconciliation timings. pub fn with_reconcile(mut self, debounce: Duration, interval: Duration) -> Self { self.reconcile_debounce = debounce; @@ -230,6 +203,10 @@ struct NetworkState { device: Option>, announcements: HashMap, links: HashMap, + /// What the network agreed, pushed in by the agent. Authoritative. + allocations: HashMap, + /// The range those allocations came from. + ipv4_range: Option, } #[derive(Debug, Default)] @@ -252,6 +229,8 @@ enum Command { struct Worker { config: WireguardConfig, + /// This agent's endpoint id, learned when the plugin is attached. + local_id: OnceLock, tun_factory: Arc, store: WgKeyStore, shared: Mutex, @@ -303,6 +282,7 @@ impl WireguardPlugin { let worker = Arc::new(Worker { config, + local_id: OnceLock::new(), tun_factory, store, shared: Mutex::new(Shared::default()), @@ -343,9 +323,7 @@ impl WireguardPlugin { endpoint_id: *endpoint_id, public_key: announcement.public_key, overlay_address: IpAddr::V6(announcement.overlay_address), - overlay_address_v4: tunnels - .get(&announcement.public_key) - .and_then(|tunnel| tunnel.overlay_address_v4), + overlay_address_v4: state.allocations.get(endpoint_id).copied(), has_link: state.links.contains_key(endpoint_id), tunnel: tunnels.get(&announcement.public_key).cloned(), }) @@ -360,12 +338,8 @@ impl WireguardPlugin { overlay_address: IpAddr::V6(overlay_address(network, &state.key.public())), overlay_prefix: IpAddr::V6(overlay_prefix(network)), overlay_prefix_len: OVERLAY_PREFIX_LEN, - overlay_address_v4: self - .worker - .config - .ipv4_range - .and_then(|range| overlay_address_v4(network, &state.key.public(), range)), - ipv4_range: self.worker.config.ipv4_range, + overlay_address_v4: state.allocations.get(&self.worker.local_id()).copied(), + ipv4_range: state.ipv4_range, peers, unroutable_packets: state .device @@ -395,6 +369,13 @@ impl WireguardPlugin { } impl Worker { + /// This agent's endpoint id, or a placeholder before it is attached. + fn local_id(&self) -> EndpointId { + self.local_id.get().copied().unwrap_or_else(|| { + EndpointId::from_bytes(&[1u8; 32]).unwrap_or_else(|_| unreachable!("a fixed valid key")) + }) + } + fn lock_shared(&self) -> std::sync::MutexGuard<'_, Shared> { match self.shared.lock() { Ok(guard) => guard, @@ -452,6 +433,8 @@ impl Worker { device: None, announcements: HashMap::new(), links: HashMap::new(), + allocations: HashMap::new(), + ipv4_range: None, }); } @@ -466,12 +449,15 @@ impl Worker { /// Creates the packet interface and starts the WireGuard device. async fn ensure_device(&self, network: NetworkId) -> Result<(), PluginError> { - let (name, key) = { + let (name, key, own_v4, own_range) = { let shared = self.lock_shared(); match shared.networks.get(&network) { - Some(state) if state.device.is_none() => { - (state.interface.clone(), state.key.clone()) - } + Some(state) if state.device.is_none() => ( + state.interface.clone(), + state.key.clone(), + state.allocations.get(&self.local_id()).copied(), + state.ipv4_range, + ), _ => return Ok(()), } }; @@ -480,20 +466,12 @@ impl Worker { name: name.clone(), address: overlay_address(network, &key.public()), prefix_len: OVERLAY_PREFIX_LEN, - address_v4: self - .config - .ipv4_range - .and_then(|range| overlay_address_v4(network, &key.public(), range)), - prefix_len_v4: self.config.ipv4_range.map_or(0, |range| range.prefix_len), + address_v4: own_v4, + prefix_len_v4: own_range.map_or(0, |range| range.prefix_len), mtu: self.config.mtu, }; let tun = self.tun_factory.create(request).await?; - let device = Arc::new(WireguardDevice::start( - network, - key, - tun, - self.config.ipv4_range, - )); + let device = Arc::new(WireguardDevice::start(network, key, tun, own_range)); let mut shared = self.lock_shared(); if let Some(state) = shared.networks.get_mut(&network) { @@ -516,9 +494,9 @@ impl Worker { return; }; + let allocations = state.allocations.clone(); let mut wanted: Vec = Vec::new(); let mut too_small: Vec<(usize, usize)> = Vec::new(); - let mut mismatched: Vec<(WgPublicKey, Ipv4Range, Ipv4Range)> = Vec::new(); for (endpoint_id, announcement) in &state.announcements { let Some(link) = state.links.get(endpoint_id) else { continue; @@ -540,19 +518,10 @@ impl Worker { too_small.push((available, needed)); } - // A peer only gets an IPv4 address if both sides were configured - // with the same range. Otherwise the two would derive different - // addresses for each other and IPv4 would silently misroute. - let peer_v4 = match (self.config.ipv4_range, announcement.ipv4_range) { - (Some(ours), Some(theirs)) if ours == theirs => { - overlay_address_v4(network, &announcement.public_key, ours) - } - (Some(ours), Some(theirs)) => { - mismatched.push((announcement.public_key, ours, theirs)); - None - } - (Some(_), None) | (None, Some(_)) | (None, None) => None, - }; + // The address comes from the agreed signed state, not from + // anything this peer said and not from a derivation: that is what + // makes it survive the peer being away. + let peer_v4 = allocations.get(endpoint_id).copied(); if let Err(err) = device.add_peer( *endpoint_id, @@ -567,18 +536,6 @@ impl Worker { device.retain_peers(&wanted); drop(shared); - for (key, ours, theirs) in mismatched { - self.report( - network, - format!( - "peer {} is configured with the IPv4 overlay range {theirs} but this agent \ - uses {ours}; every member must use the same one. That peer has no IPv4 \ - address here and is reachable over IPv6 only.", - key.fmt_short() - ), - ); - } - for (available, needed) in too_small { self.report( network, @@ -686,6 +643,9 @@ impl IpPlugin for WireguardPlugin { } fn attach(&self, context: PluginContext) { + if let Some(local) = context.local_endpoint_id() { + let _ = self.worker.local_id.set(local); + } let _ = self.worker.context.set(context); } @@ -707,8 +667,7 @@ impl IpPlugin for WireguardPlugin { }; // Identity only. Where to send packets is the transport's business. - let announcement = - WgAnnouncement::new(network, &state.key.public(), self.worker.config.ipv4_range); + let announcement = WgAnnouncement::new(network, &state.key.public()); Ok(Some(PluginCapability { protocol: WIREGUARD_PROTOCOL.to_string(), version: super::announcement::ANNOUNCEMENT_VERSION, @@ -754,6 +713,31 @@ impl IpPlugin for WireguardPlugin { Ok(()) } + fn on_address_allocation( + &self, + network: NetworkId, + range: Ipv4Range, + allocations: &[(EndpointId, Ipv4Addr)], + ) { + let changed = { + let mut shared = self.worker.lock_shared(); + match shared.networks.get_mut(&network) { + Some(state) => { + let fresh: HashMap = + allocations.iter().copied().collect(); + let changed = state.allocations != fresh || state.ipv4_range != Some(range); + state.allocations = fresh; + state.ipv4_range = Some(range); + changed + } + None => false, + } + }; + if changed { + self.nudge(Command::Sync(network)); + } + } + fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) { self.nudge(Command::Link { network, diff --git a/src/identity/mod.rs b/src/identity/mod.rs index c1ad0b9..662c5c1 100644 --- a/src/identity/mod.rs +++ b/src/identity/mod.rs @@ -58,6 +58,14 @@ impl DeviceIdentity { pub(crate) fn secret_key(&self) -> SecretKey { self.secret.clone() } + + /// A clone of the key used to sign this device's own state records. + /// + /// The same persistent identity the control plane authenticates, so a + /// record signed today is still attributable after any absence. + pub(crate) fn signing_key(&self) -> SecretKey { + self.secret.clone() + } } impl std::fmt::Debug for DeviceIdentity { diff --git a/src/lib.rs b/src/lib.rs index eb92b1b..c786f62 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,8 @@ //! * [`agent`] — the runtime: agent lifecycle, per-network runtimes, reconnect. //! * [`dataplane`] — the contract IP plugins satisfy, the packet transport, //! and the WireGuard data plane. +//! * [`state`] — signed records that outlive a session, and the rules for +//! merging them between replicas. //! * [`ipc`] — the local control interface a command line tool talks to. An //! adapter over the public API; the core does not know it exists. //! @@ -42,6 +44,7 @@ pub mod identity; pub mod ipc; pub mod net; pub mod proto; +pub mod state; pub mod storage; pub use agent::{Agent, AgentStatus, Event, NetworkStatus, PeerStatus}; diff --git a/src/proto/message.rs b/src/proto/message.rs index 9439499..15384fb 100644 --- a/src/proto/message.rs +++ b/src/proto/message.rs @@ -33,6 +33,9 @@ pub const DATA_ALPN: &[u8] = b"tsunagi/data/1"; /// Largest plugin protocol identifier accepted when opening a data channel. pub const MAX_DATA_PROTOCOL_LEN: usize = 32; +/// Largest accepted signature on a signed record, in bytes. +pub const MAX_SIGNATURE_LEN: usize = 64; + /// Control protocol version carried inside the handshake. pub const PROTOCOL_VERSION: u16 = 1; @@ -112,6 +115,15 @@ pub enum ControlMessage { /// Echoed payload. payload: Vec, }, + /// A snapshot of signed records this agent holds for the network. + /// + /// A snapshot is merged into what the receiver already has, never + /// substituted for it: an author missing from the batch is left alone, + /// because absence is not deletion. + State { + /// The records. Bounded by [`crate::config::Limits::max_state_records`]. + records: Vec, + }, /// Graceful goodbye. /// /// A peer going away is not a revocation of anything. @@ -197,6 +209,12 @@ pub fn validate(message: &ControlMessage, limits: &Limits) -> Result<(), Protoco ControlMessage::Ping { payload, .. } | ControlMessage::Pong { payload, .. } => { check_len("echo.payload", payload.len(), limits.max_echo_payload_len)?; } + ControlMessage::State { records } => { + check_len("state.records", records.len(), limits.max_state_records)?; + for record in records { + check_len("state.signature", record.signature.len(), MAX_SIGNATURE_LEN)?; + } + } ControlMessage::Bye { reason } => { check_len("bye.reason", reason.len(), limits.max_reason_len)?; } @@ -210,6 +228,7 @@ pub fn kind(message: &ControlMessage) -> &'static str { ControlMessage::Announce(_) => "announce", ControlMessage::Ping { .. } => "ping", ControlMessage::Pong { .. } => "pong", + ControlMessage::State { .. } => "state", ControlMessage::Bye { .. } => "bye", } } diff --git a/src/state/allocator.rs b/src/state/allocator.rs new file mode 100644 index 0000000..aa5e4da --- /dev/null +++ b/src/state/allocator.rs @@ -0,0 +1,279 @@ +//! Choosing a free overlay address. +//! +//! Allocation, not derivation. Derivation needs no coordination but cannot +//! avoid collisions in a space as small as IPv4; allocation avoids them but +//! has to look at what everybody else already holds. The signed records in +//! [`super`] are what makes that possible without a coordinator. +//! +//! The rules: +//! +//! * an address a participant already holds is kept, because stability across +//! an absence is the whole point; +//! * otherwise the search starts at a position derived from the participant's +//! own identity, so two participants joining at once rarely start in the +//! same place; +//! * the search then walks the range, so a free address is found whenever one +//! exists. + +use std::collections::HashSet; +use std::net::Ipv4Addr; + +use iroh::EndpointId; +use sha2::{Digest, Sha256}; + +use super::Ipv4Range; +use crate::identity::NetworkId; + +/// Why no address could be allocated. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +pub enum AllocationError { + /// Every address in the range is taken. + #[error("the overlay range {range} is full: {holders} of {usable} addresses are taken")] + RangeFull { + /// The range that is full. + range: Ipv4Range, + /// How many are held. + holders: usize, + /// How many the range has. + usable: u64, + }, + /// The range has no usable host addresses. + #[error("the overlay range {range} has no room for hosts")] + NoRoom { + /// The offending range. + range: Ipv4Range, + }, +} + +/// How many host addresses a range holds, excluding network and broadcast. +pub fn usable_addresses(range: Ipv4Range) -> u64 { + let host_bits = 32u32.saturating_sub(u32::from(range.prefix_len)); + if host_bits < 2 { + return 0; + } + (1u64 << host_bits) - 2 +} + +/// The nth host address of a range. +fn address_at(range: Ipv4Range, offset: u64) -> Ipv4Addr { + let host_bits = 32u32.saturating_sub(u32::from(range.prefix_len)); + let mask = if host_bits >= 32 { + 0 + } else { + u32::MAX << host_bits + }; + let network_part = u32::from(range.base) & mask; + // Offsets run 1..=usable, so the network address is never handed out. + Ipv4Addr::from(network_part | ((offset % (1u64 << host_bits)) as u32)) +} + +/// Picks an address for `author`, keeping `current` if it is still usable. +/// +/// `taken` is what every other participant is known to hold. +pub fn allocate( + network: NetworkId, + author: EndpointId, + range: Ipv4Range, + taken: &HashSet, + current: Option, +) -> Result { + let usable = usable_addresses(range); + if usable == 0 { + return Err(AllocationError::NoRoom { range }); + } + + // Keeping what we already hold is what lets a participant come back to + // the same address after any length of absence. + if let Some(current) = current + && range.contains(current) + && !taken.contains(¤t) + { + return Ok(current); + } + + // Start somewhere derived from who we are, so two newcomers do not both + // begin at the first address and collide every time. + let mut hash = Sha256::new(); + hash.update(b"tsunagi-ipv4-allocation-v1"); + hash.update(network.as_bytes()); + hash.update(author.as_bytes()); + let digest = hash.finalize(); + let seed = u64::from_be_bytes([ + digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7], + ]); + + for step in 0..usable { + let offset = ((seed.wrapping_add(step)) % usable) + 1; + let candidate = address_at(range, offset); + if !taken.contains(&candidate) { + return Ok(candidate); + } + } + + Err(AllocationError::RangeFull { + range, + holders: taken.len(), + usable, + }) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use crate::identity::{NetworkKeys, NetworkName, NetworkSecret}; + use iroh::SecretKey; + + fn network(name: &str) -> NetworkId { + NetworkKeys::derive( + &NetworkName::new(name).unwrap(), + &NetworkSecret::from_bytes(vec![7u8; 32]).unwrap(), + ) + .network_id() + } + + fn slash24() -> Ipv4Range { + "10.13.37.0/24".parse().unwrap() + } + + #[test] + fn a_range_reports_its_usable_size() { + assert_eq!(usable_addresses(slash24()), 254); + assert_eq!(usable_addresses("10.0.0.0/16".parse().unwrap()), 65534); + assert_eq!(usable_addresses("10.0.0.0/30".parse().unwrap()), 2); + // The type refuses a /31, but the function is defensive anyway. + assert!("10.0.0.0/31".parse::().is_err()); + assert_eq!( + usable_addresses(Ipv4Range { + base: "10.0.0.0".parse().unwrap(), + prefix_len: 31 + }), + 0 + ); + } + + #[test] + fn an_allocated_address_is_inside_the_range_and_not_its_edges() { + let id = network("inside"); + let taken = HashSet::new(); + for _ in 0..64 { + let author = SecretKey::generate().public(); + let address = allocate(id, author, slash24(), &taken, None).unwrap(); + assert!(slash24().contains(address)); + assert_ne!(address.octets()[3], 0, "never the network address"); + assert_ne!(address.octets()[3], 255, "never the broadcast address"); + } + } + + #[test] + fn an_address_already_held_is_kept() { + let id = network("sticky"); + let author = SecretKey::generate().public(); + let mine: Ipv4Addr = "10.13.37.42".parse().unwrap(); + + // This is what lets a participant return to the same address. + let taken = HashSet::new(); + assert_eq!( + allocate(id, author, slash24(), &taken, Some(mine)).unwrap(), + mine + ); + + // Unless somebody else took it while we were away. + let taken = HashSet::from([mine]); + assert_ne!( + allocate(id, author, slash24(), &taken, Some(mine)).unwrap(), + mine + ); + + // Or unless the range changed under us. + let elsewhere: Ipv4Range = "10.99.0.0/16".parse().unwrap(); + let moved = allocate(id, author, elsewhere, &HashSet::new(), Some(mine)).unwrap(); + assert!(elsewhere.contains(moved)); + } + + #[test] + fn allocation_is_deterministic_and_spread_out() { + let id = network("spread"); + let authors: Vec<_> = (0..40).map(|_| SecretKey::generate().public()).collect(); + + let first: Vec<_> = authors + .iter() + .map(|author| allocate(id, *author, slash24(), &HashSet::new(), None).unwrap()) + .collect(); + let again: Vec<_> = authors + .iter() + .map(|author| allocate(id, *author, slash24(), &HashSet::new(), None).unwrap()) + .collect(); + assert_eq!(first, again, "the same inputs give the same answer"); + + // Starting points are spread, so concurrent newcomers rarely clash. + let distinct: HashSet<_> = first.iter().collect(); + assert!( + distinct.len() >= 35, + "only {} distinct starting points out of 40", + distinct.len() + ); + } + + #[test] + fn the_search_walks_past_everything_taken() { + let id = network("crowded"); + let author = SecretKey::generate().public(); + + // Everything taken except one address. + let free: Ipv4Addr = "10.13.37.200".parse().unwrap(); + let taken: HashSet = (1..=254u8) + .map(|host| Ipv4Addr::new(10, 13, 37, host)) + .filter(|addr| *addr != free) + .collect(); + assert_eq!(allocate(id, author, slash24(), &taken, None).unwrap(), free); + } + + #[test] + fn a_full_range_is_an_error_rather_than_a_duplicate() { + let id = network("full"); + let author = SecretKey::generate().public(); + let taken: HashSet = (1..=254u8) + .map(|host| Ipv4Addr::new(10, 13, 37, host)) + .collect(); + + assert!(matches!( + allocate(id, author, slash24(), &taken, None), + Err(AllocationError::RangeFull { .. }) + )); + assert!(matches!( + allocate( + id, + author, + Ipv4Range { + base: "10.0.0.0".parse().unwrap(), + prefix_len: 31 + }, + &HashSet::new(), + None + ), + Err(AllocationError::NoRoom { .. }) + )); + } + + #[test] + fn every_address_in_a_small_range_can_be_handed_out() { + let id = network("exhaustive"); + let small: Ipv4Range = "10.13.37.0/29".parse().unwrap(); + let mut taken = HashSet::new(); + let mut handed = Vec::new(); + + for _ in 0..usable_addresses(small) { + let author = SecretKey::generate().public(); + let address = allocate(id, author, small, &taken, None).unwrap(); + assert!(taken.insert(address), "handed out {address} twice"); + handed.push(address); + } + assert_eq!(handed.len(), 6); + // And then it is genuinely full. + let author = SecretKey::generate().public(); + assert!(allocate(id, author, small, &taken, None).is_err()); + } +} diff --git a/src/state/mod.rs b/src/state/mod.rs new file mode 100644 index 0000000..8a4e997 --- /dev/null +++ b/src/state/mod.rs @@ -0,0 +1,753 @@ +//! Signed state that outlives a session. +//! +//! This is the first slice of the model in `docs/sync-model.md`: **each author +//! signs its own records, and replicas merge them**. It exists because some +//! facts have to survive a participant being away — an overlay address it +//! claimed months ago, for instance — and a fact that only lives in a live +//! session cannot do that. +//! +//! # Why there is no vote +//! +//! Anyone who knows the network secret can mint as many identities as they +//! like, so a majority proves nothing; the threat model says as much. A quorum +//! would also stall whenever a single participant is online and diverge across +//! a partition. Instead: +//! +//! * an author signs only its **own** records, so nobody needs anybody's +//! permission to state a fact about itself; +//! * merging is **deterministic**, so every replica that has seen the same +//! records reaches the same conclusion without exchanging opinions; +//! * a genuine clash — two authors claiming one address at the same moment — +//! is resolved by a rule both sides compute identically, and the loser +//! simply picks again with a higher version. +//! +//! # What a record is +//! +//! One record per author per network, holding that author's **complete +//! current** statement rather than a delta, exactly as the model requires: a +//! replica that has the record needs nothing else to interpret it, and +//! recovery never depends on replaying a chain from the beginning. +//! +//! # 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. + +pub mod allocator; + +use std::collections::HashMap; +use std::net::Ipv4Addr; + +use iroh::{EndpointId, SecretKey, Signature}; +use serde::{Deserialize, Serialize}; + +use crate::identity::NetworkId; + +/// Why an IPv4 range could not be used. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("{0}")] +pub struct RangeError(pub String); + +/// An IPv4 range the overlay allocates addresses from. +/// +/// One range per network. An agent proposes one through +/// [`crate::config::AgentConfig::overlay_ipv4_range`], but a network that has +/// already settled on another wins: see [`StateSet::agreed_range`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct Ipv4Range { + /// Base address of the range. + pub base: Ipv4Addr, + /// Prefix length, at most 30 so there is room for hosts. + pub prefix_len: u8, +} + +impl Ipv4Range { + /// Builds a range, rejecting one with no room for hosts. + pub fn new(base: Ipv4Addr, prefix_len: u8) -> Result { + if prefix_len > 30 { + return Err(RangeError(format!( + "a /{prefix_len} has no room for hosts; use /30 or larger" + ))); + } + Ok(Self { base, prefix_len }) + } + + /// Whether an address falls inside the range. + pub fn contains(&self, address: Ipv4Addr) -> bool { + let host_bits = 32 - u32::from(self.prefix_len); + let mask = if host_bits >= 32 { + 0 + } else { + u32::MAX << host_bits + }; + u32::from(address) & mask == u32::from(self.base) & mask + } +} + +impl std::fmt::Display for Ipv4Range { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.base, self.prefix_len) + } +} + +impl std::str::FromStr for Ipv4Range { + type Err = RangeError; + + fn from_str(text: &str) -> Result { + let (base, prefix) = text.split_once('/').ok_or_else(|| { + RangeError(format!( + "`{text}` is not an address with a prefix, for example 10.77.0.0/16" + )) + })?; + let base = base + .parse() + .map_err(|err| RangeError(format!("`{base}` is not an IPv4 address: {err}")))?; + let prefix_len = prefix + .parse() + .map_err(|err| RangeError(format!("`{prefix}` is not a prefix length: {err}")))?; + Self::new(base, prefix_len) + } +} + +/// The IPv4 overlay range used unless something else is configured or agreed. +/// +/// A small, specific `/24`: memorable, and far less likely to overlap a +/// network the machine is already on than taking a whole `/8` or `/10` would +/// be. Because addresses are allocated rather than derived, 254 of them is +/// plenty for the size of network this is for. +pub const DEFAULT_IPV4_RANGE: Ipv4Range = Ipv4Range { + base: Ipv4Addr::new(10, 13, 37, 0), + prefix_len: 24, +}; + +/// Frozen domain separator for the bytes a record signature covers. +pub const RECORD_DOMAIN: &str = "tsunagi-signed-record-v1"; + +/// Largest number of records accepted in one exchange. +pub const MAX_RECORDS_PER_MESSAGE: usize = 256; + +/// Why a record could not be used. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +pub enum StateError { + /// The signature does not match the author. + #[error("record signature does not verify")] + BadSignature, + /// The author field is not a valid public key. + #[error("record author is not a valid endpoint id")] + BadAuthor, + /// The signature field is not the right length. + #[error("record signature is not {expected} bytes")] + BadSignatureLength { + /// Expected length. + expected: usize, + }, + /// The record belongs to a different network. + #[error("record belongs to another network")] + WrongNetwork, + /// The record's contents are not acceptable. + #[error("record is malformed: {0}")] + Malformed(&'static str), +} + +/// What an author is saying about itself. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[non_exhaustive] +pub enum RecordBody { + /// This author holds an IPv4 overlay address, in this range. + /// + /// The range travels with the claim so that a participant joining later + /// learns which range the network actually settled on, rather than having + /// to be told. + Ipv4Claim { + /// The address this author holds. + address: Ipv4Addr, + /// The overlay range it was allocated from. + range: Ipv4Range, + }, + /// This author gave its address up. + /// + /// A tombstone, not an absence: it is a positive statement, so it + /// survives merging and cannot be undone by a replica that simply has not + /// heard of it. + Ipv4Release, +} + +impl RecordBody { + /// The address this body claims, if any. + pub fn claimed_address(&self) -> Option { + match self { + RecordBody::Ipv4Claim { address, .. } => Some(*address), + RecordBody::Ipv4Release => None, + } + } + + /// The range this body names, if any. + pub fn range(&self) -> Option { + match self { + RecordBody::Ipv4Claim { range, .. } => Some(*range), + RecordBody::Ipv4Release => None, + } + } + + fn canonical(&self) -> Vec { + let mut out = Vec::with_capacity(48); + match self { + RecordBody::Ipv4Claim { address, range } => { + push_lp(&mut out, b"ipv4-claim"); + push_lp(&mut out, &address.octets()); + push_lp(&mut out, &range.base.octets()); + push_lp(&mut out, &[range.prefix_len]); + } + RecordBody::Ipv4Release => { + push_lp(&mut out, b"ipv4-release"); + } + } + out + } +} + +fn push_lp(out: &mut Vec, bytes: &[u8]) { + let len = u32::try_from(bytes.len()).unwrap_or(u32::MAX); + out.extend_from_slice(&len.to_be_bytes()); + out.extend_from_slice(bytes); +} + +/// One author's current statement about itself, signed by that author. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SignedRecord { + /// The author's persistent endpoint id. + pub author: [u8; 32], + /// The network the statement belongs to. + pub network: [u8; 32], + /// The author's own counter. Only the author increments it. + pub version: u64, + /// The statement. + pub body: RecordBody, + /// Ed25519 signature over [`SignedRecord::canonical_bytes`]. + pub signature: Vec, +} + +impl SignedRecord { + /// The bytes a signature covers. + /// + /// Length-prefixed throughout, so no two different records can produce the + /// same bytes. + pub fn canonical_bytes( + network: NetworkId, + author: EndpointId, + version: u64, + body: &RecordBody, + ) -> Vec { + let mut out = Vec::with_capacity(160); + push_lp(&mut out, RECORD_DOMAIN.as_bytes()); + push_lp(&mut out, network.as_bytes()); + push_lp(&mut out, author.as_bytes()); + push_lp(&mut out, &version.to_be_bytes()); + push_lp(&mut out, &body.canonical()); + out + } + + /// Signs a new record with the author's persistent device key. + pub fn sign(secret: &SecretKey, network: NetworkId, version: u64, body: RecordBody) -> Self { + let author = secret.public(); + let signature = secret.sign(&Self::canonical_bytes(network, author, version, &body)); + Self { + author: *author.as_bytes(), + network: *network.as_bytes(), + version, + body, + signature: signature.to_bytes().to_vec(), + } + } + + /// The author, if the field is a valid key. + pub fn author_id(&self) -> Result { + EndpointId::from_bytes(&self.author).map_err(|_| StateError::BadAuthor) + } + + /// The network this record belongs to. + pub fn network_id(&self) -> NetworkId { + NetworkId::from_bytes(self.network) + } + + /// Checks the signature and that the record belongs to `network`. + /// + /// Everything that reaches this from the network goes through it first. + pub fn verify(&self, network: NetworkId) -> Result { + if self.network != *network.as_bytes() { + return Err(StateError::WrongNetwork); + } + if let RecordBody::Ipv4Claim { address, range } = &self.body { + if range.prefix_len > 30 { + return Err(StateError::Malformed("claimed range has no room for hosts")); + } + if !range.contains(*address) { + return Err(StateError::Malformed( + "claimed address is outside its range", + )); + } + } + + let author = self.author_id()?; + let raw: [u8; Signature::LENGTH] = + self.signature + .as_slice() + .try_into() + .map_err(|_| StateError::BadSignatureLength { + expected: Signature::LENGTH, + })?; + let signature = Signature::from_bytes(&raw); + author + .verify( + &Self::canonical_bytes(network, author, self.version, &self.body), + &signature, + ) + .map_err(|_| StateError::BadSignature)?; + Ok(author) + } +} + +/// What merging one record did. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Merged { + /// Nothing was known about this author; the record was taken. + Added, + /// It replaced an older version from the same author. + Updated, + /// Already known, or older than what is held. Nothing changed. + /// + /// An older version never rolls back a newer one. + Ignored, + /// Two different records from one author at the same version. + /// + /// Resolved deterministically so every replica picks the same one, and + /// reported because it means a key is being used from two places at once. + Conflicted, +} + +/// Everything known about one network, one record per author. +#[derive(Debug, Clone, Default)] +pub struct StateSet { + records: HashMap, +} + +impl StateSet { + /// An empty set. + pub fn new() -> Self { + Self::default() + } + + /// Builds a set from records already known to be verified. + pub fn from_verified(records: impl IntoIterator) -> Self { + Self { + records: records.into_iter().collect(), + } + } + + /// How many authors are known. + pub fn len(&self) -> usize { + self.records.len() + } + + /// Whether nothing is known. + pub fn is_empty(&self) -> bool { + self.records.is_empty() + } + + /// The record of one author. + pub fn get(&self, author: &EndpointId) -> Option<&SignedRecord> { + self.records.get(author) + } + + /// Every record, in a stable order. + pub fn records(&self) -> Vec { + let mut authors: Vec<&EndpointId> = self.records.keys().collect(); + authors.sort_by_key(|author| *author.as_bytes()); + authors + .into_iter() + .filter_map(|author| self.records.get(author).cloned()) + .collect() + } + + /// Merges one record, verifying it first. + /// + /// Merging is into the existing set, never a wholesale replacement, and an + /// author missing from an incoming batch is left untouched — absence is + /// not deletion. + pub fn merge( + &mut self, + network: NetworkId, + record: SignedRecord, + ) -> Result { + let author = record.verify(network)?; + + match self.records.get(&author) { + None => { + self.records.insert(author, record); + Ok(Merged::Added) + } + Some(existing) if existing.version < record.version => { + self.records.insert(author, record); + Ok(Merged::Updated) + } + Some(existing) if existing.version > record.version => Ok(Merged::Ignored), + Some(existing) if existing.body == record.body => Ok(Merged::Ignored), + Some(existing) => { + // Same author, same version, different content: the author's + // key is in use in two places. Neither is more true than the + // other, so pick by a rule every replica computes identically + // and report it rather than letting replicas diverge. + if record.signature < existing.signature { + self.records.insert(author, record); + } + Ok(Merged::Conflicted) + } + } + } + + /// Merges a batch, returning what happened and the first error seen. + /// + /// A bad record in a batch is skipped; the rest still merge. + pub fn merge_all( + &mut self, + network: NetworkId, + records: impl IntoIterator, + ) -> (Vec, Vec) { + let mut outcomes = Vec::new(); + let mut errors = Vec::new(); + for record in records { + match self.merge(network, record) { + Ok(outcome) => outcomes.push(outcome), + Err(err) => errors.push(err), + } + } + (outcomes, errors) + } + + /// Who currently holds each claimed address. + /// + /// When two authors claim one address, the one whose endpoint id sorts + /// lower holds it — again a rule every replica computes identically. The + /// other is expected to notice and claim a different one. + pub fn address_holders(&self) -> HashMap { + let mut holders: HashMap = HashMap::new(); + for (author, record) in &self.records { + let Some(address) = record.body.claimed_address() else { + continue; + }; + holders + .entry(address) + .and_modify(|held| { + if author.as_bytes() < held.as_bytes() { + *held = *author; + } + }) + .or_insert(*author); + } + holders + } + + /// The address an author holds, if it holds one uncontested. + pub fn address_of(&self, author: &EndpointId) -> Option { + let address = self.records.get(author)?.body.claimed_address()?; + (self.address_holders().get(&address) == Some(author)).then_some(address) + } + + /// The range the network settled on, if anybody has said. + /// + /// When claims disagree, the one from the lowest author id wins, so every + /// replica reaches the same answer. A participant joining later therefore + /// adopts the range already in use instead of imposing its own. + pub fn agreed_range(&self) -> Option { + let mut authors: Vec<&EndpointId> = self.records.keys().collect(); + authors.sort_by_key(|author| *author.as_bytes()); + authors + .into_iter() + .find_map(|author| self.records.get(author)?.body.range()) + } + + /// The highest version this author has published, as far as is known. + pub fn version_of(&self, author: &EndpointId) -> u64 { + self.records.get(author).map_or(0, |record| record.version) + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use crate::identity::{NetworkKeys, NetworkName, NetworkSecret}; + + fn network(name: &str) -> NetworkId { + NetworkKeys::derive( + &NetworkName::new(name).unwrap(), + &NetworkSecret::from_bytes(vec![6u8; 32]).unwrap(), + ) + .network_id() + } + + fn range() -> Ipv4Range { + "10.13.37.0/24".parse().unwrap() + } + + fn claim(address: &str) -> RecordBody { + RecordBody::Ipv4Claim { + address: address.parse().unwrap(), + range: range(), + } + } + + #[test] + fn a_record_verifies_only_against_its_own_author_and_network() { + let id = network("verify"); + let secret = SecretKey::generate(); + let record = SignedRecord::sign(&secret, id, 1, claim("10.13.37.5")); + + assert_eq!(record.verify(id).unwrap(), secret.public()); + // A record from another network does not apply here. + assert_eq!( + record.verify(network("other")).unwrap_err(), + StateError::WrongNetwork + ); + + // Changing anything invalidates the signature. + for tampered in [ + SignedRecord { + version: 2, + ..record.clone() + }, + SignedRecord { + body: claim("10.13.37.6"), + ..record.clone() + }, + SignedRecord { + author: *SecretKey::generate().public().as_bytes(), + ..record.clone() + }, + ] { + assert!(tampered.verify(id).is_err(), "tampering must be caught"); + } + } + + #[test] + fn malformed_records_are_rejected_without_panicking() { + let id = network("malformed"); + let secret = SecretKey::generate(); + let good = SignedRecord::sign(&secret, id, 1, claim("10.13.37.5")); + + let short_signature = SignedRecord { + signature: vec![0u8; 8], + ..good.clone() + }; + assert!(matches!( + short_signature.verify(id), + Err(StateError::BadSignatureLength { .. }) + )); + + // An address outside the range it names is nonsense. + let outside = SignedRecord::sign( + &secret, + id, + 1, + RecordBody::Ipv4Claim { + address: "10.99.0.1".parse().unwrap(), + range: range(), + }, + ); + assert!(matches!(outside.verify(id), Err(StateError::Malformed(_)))); + + let no_hosts = SignedRecord::sign( + &secret, + id, + 1, + RecordBody::Ipv4Claim { + address: "10.13.37.1".parse().unwrap(), + range: Ipv4Range { + base: "10.13.37.0".parse().unwrap(), + prefix_len: 31, + }, + }, + ); + assert!(matches!(no_hosts.verify(id), Err(StateError::Malformed(_)))); + } + + #[test] + fn a_newer_version_wins_and_an_older_one_never_rolls_back() { + let id = network("versions"); + let secret = SecretKey::generate(); + let mut set = StateSet::new(); + + let first = SignedRecord::sign(&secret, id, 1, claim("10.13.37.5")); + let second = SignedRecord::sign(&secret, id, 2, claim("10.13.37.6")); + + assert_eq!(set.merge(id, first.clone()).unwrap(), Merged::Added); + assert_eq!(set.merge(id, second.clone()).unwrap(), Merged::Updated); + // The old one coming back later must not undo the new one. + assert_eq!(set.merge(id, first).unwrap(), Merged::Ignored); + assert_eq!( + set.address_of(&secret.public()), + Some("10.13.37.6".parse().unwrap()) + ); + // Merging the same record twice changes nothing. + assert_eq!(set.merge(id, second).unwrap(), Merged::Ignored); + assert_eq!(set.len(), 1); + } + + #[test] + fn two_authors_claiming_one_address_resolve_the_same_way_everywhere() { + let id = network("clash"); + let (low, high) = { + let a = SecretKey::generate(); + let b = SecretKey::generate(); + if a.public().as_bytes() < b.public().as_bytes() { + (a, b) + } else { + (b, a) + } + }; + + let record_low = SignedRecord::sign(&low, id, 1, claim("10.13.37.5")); + let record_high = SignedRecord::sign(&high, id, 1, claim("10.13.37.5")); + + // Merge order must not change the outcome. + let mut forwards = StateSet::new(); + forwards.merge(id, record_low.clone()).unwrap(); + forwards.merge(id, record_high.clone()).unwrap(); + + let mut backwards = StateSet::new(); + backwards.merge(id, record_high).unwrap(); + backwards.merge(id, record_low).unwrap(); + + let expected = Some(low.public()); + assert_eq!( + forwards + .address_holders() + .get(&"10.13.37.5".parse().unwrap()), + expected.as_ref() + ); + assert_eq!( + backwards + .address_holders() + .get(&"10.13.37.5".parse().unwrap()), + expected.as_ref() + ); + // The loser holds nothing, and is expected to pick again. + assert_eq!(forwards.address_of(&high.public()), None); + assert_eq!( + forwards.address_of(&low.public()), + Some("10.13.37.5".parse().unwrap()) + ); + } + + #[test] + fn one_key_used_in_two_places_is_reported_not_silently_merged() { + let id = network("split-brain"); + let secret = SecretKey::generate(); + let mut set = StateSet::new(); + + let here = SignedRecord::sign(&secret, id, 3, claim("10.13.37.5")); + let there = SignedRecord::sign(&secret, id, 3, claim("10.13.37.9")); + + set.merge(id, here.clone()).unwrap(); + assert_eq!(set.merge(id, there.clone()).unwrap(), Merged::Conflicted); + + // Whatever it picked, it must pick the same thing from the other side. + let mut other = StateSet::new(); + other.merge(id, there).unwrap(); + assert_eq!(other.merge(id, here).unwrap(), Merged::Conflicted); + assert_eq!( + set.get(&secret.public()).unwrap(), + other.get(&secret.public()).unwrap() + ); + } + + #[test] + fn a_release_is_a_statement_that_survives_merging() { + let id = network("release"); + let secret = SecretKey::generate(); + let mut set = StateSet::new(); + + set.merge(id, SignedRecord::sign(&secret, id, 1, claim("10.13.37.5"))) + .unwrap(); + assert!(set.address_of(&secret.public()).is_some()); + + set.merge( + id, + SignedRecord::sign(&secret, id, 2, RecordBody::Ipv4Release), + ) + .unwrap(); + assert_eq!(set.address_of(&secret.public()), None); + assert!(set.address_holders().is_empty()); + + // The old claim arriving late does not resurrect the address. + assert_eq!( + set.merge(id, SignedRecord::sign(&secret, id, 1, claim("10.13.37.5"))) + .unwrap(), + Merged::Ignored + ); + assert_eq!(set.address_of(&secret.public()), None); + } + + #[test] + fn a_batch_with_one_bad_record_still_merges_the_rest() { + let id = network("batch"); + let good = SecretKey::generate(); + let mut set = StateSet::new(); + + let valid = SignedRecord::sign(&good, id, 1, claim("10.13.37.5")); + let forged = SignedRecord { + signature: vec![0u8; Signature::LENGTH], + ..SignedRecord::sign(&SecretKey::generate(), id, 1, claim("10.13.37.6")) + }; + + let (outcomes, errors) = set.merge_all(id, [forged, valid]); + assert_eq!(outcomes, vec![Merged::Added]); + assert_eq!(errors, vec![StateError::BadSignature]); + assert_eq!(set.len(), 1); + } + + #[test] + fn a_later_joiner_adopts_the_range_already_in_use() { + let id = network("ranges"); + let mut set = StateSet::new(); + assert_eq!(set.agreed_range(), None, "nothing known yet"); + + let custom: Ipv4Range = "10.99.0.0/16".parse().unwrap(); + let author = SecretKey::generate(); + set.merge( + id, + SignedRecord::sign( + &author, + id, + 1, + RecordBody::Ipv4Claim { + address: "10.99.0.7".parse().unwrap(), + range: custom, + }, + ), + ) + .unwrap(); + assert_eq!(set.agreed_range(), Some(custom)); + } + + #[test] + fn the_signed_bytes_are_unambiguous() { + let id = network("encoding"); + let author = SecretKey::generate().public(); + // Two bodies whose parts would concatenate identically must not + // produce the same signed bytes. + let a = SignedRecord::canonical_bytes(id, author, 1, &claim("10.13.37.5")); + let b = SignedRecord::canonical_bytes(id, author, 1, &claim("10.13.37.6")); + assert_ne!(a, b); + assert_ne!( + a, + SignedRecord::canonical_bytes(id, author, 2, &claim("10.13.37.5")) + ); + assert_ne!( + a, + SignedRecord::canonical_bytes(network("other"), author, 1, &claim("10.13.37.5")) + ); + } +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 086d461..dcd54f3 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -27,6 +27,7 @@ use std::sync::{Arc, Mutex, MutexGuard}; use crate::config::StoragePaths; use crate::error::{Error, Result}; use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret}; +use crate::state::SignedRecord; /// Applies the pragmas both stores share. fn apply_common_pragmas(conn: &rusqlite::Connection) -> rusqlite::Result<()> { @@ -246,8 +247,11 @@ impl Storage { /// Removes a network configuration and its cached hints. pub async fn remove_network(&self, network_id: NetworkId) -> Result<()> { - self.with_state(move |state| state.remove_network(network_id)) - .await?; + self.with_state(move |state| { + state.remove_network(network_id)?; + state.forget_signed_records(network_id) + }) + .await?; self.with_cache((), move |cache| cache.forget_network(network_id)) .await; Ok(()) @@ -278,6 +282,31 @@ impl Storage { .await; } + /// Loads every signed record known for a network. + pub async fn signed_records(&self, network_id: NetworkId) -> Result> { + self.with_state(move |state| state.signed_records(network_id)) + .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)) + .await + } + + /// Stores one of this agent's own records and bumps its counter in one + /// transaction, which must happen before the record is announced. + pub async fn publish_own_record(&self, record: SignedRecord) -> Result<()> { + self.with_state(move |state| state.publish_own_record(&record)) + .await + } + + /// The highest version this agent has ever published for a network. + pub async fn own_record_version(&self, network_id: NetworkId) -> Result { + self.with_state(move |state| state.own_record_version(network_id)) + .await + } + /// Reads cached address hints. Returns an empty list if the cache is gone. pub async fn hints_for_network(&self, network_id: NetworkId) -> Vec { self.with_cache(Vec::new(), move |cache| cache.hints_for_network(network_id)) diff --git a/src/storage/state.rs b/src/storage/state.rs index d4c1aa0..5a519fe 100644 --- a/src/storage/state.rs +++ b/src/storage/state.rs @@ -18,9 +18,10 @@ use rusqlite::{Connection, OptionalExtension, params}; use crate::error::{Error, Result}; use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret}; +use crate::state::SignedRecord; /// Schema version written by this build. -pub const SCHEMA_VERSION: i64 = 1; +pub const SCHEMA_VERSION: i64 = 2; /// Key of the stored hostname setting. const SETTING_HOSTNAME: &str = "hostname"; @@ -115,6 +116,13 @@ impl StateStore { return self.verify_shape(); } + // Migration 1 -> 2: signed records that outlive a session. + if (1..2).contains(&found) { + self.conn + .execute_batch(SIGNED_RECORDS_SCHEMA) + .map_err(|err| self.corrupt(format!("cannot migrate schema to 2: {err}")))?; + } + // Migration 0 -> 1: initial schema. if found < 1 { self.conn @@ -140,6 +148,9 @@ impl StateStore { COMMIT;", ) .map_err(|err| self.corrupt(format!("cannot create schema: {err}")))?; + self.conn + .execute_batch(SIGNED_RECORDS_SCHEMA) + .map_err(|err| self.corrupt(format!("cannot create schema: {err}")))?; } Ok(()) } @@ -147,7 +158,7 @@ impl StateStore { /// Confirms the expected tables exist, so that a truncated or foreign /// database is reported rather than used. fn verify_shape(&self) -> Result<()> { - for table in ["device_identity", "networks", "settings"] { + for table in ["device_identity", "networks", "settings", "signed_records"] { let present: Option = self .conn .query_row( @@ -297,6 +308,147 @@ impl StateStore { self.set_setting(SETTING_HOSTNAME, hostname) } + /// Loads every signed record known for a network. + /// + /// Records are returned as stored; the caller verifies them, because the + /// database is not a trust boundary — a restored backup or a copied file + /// could contain anything. + pub fn signed_records(&self, network_id: NetworkId) -> Result> { + let mut stmt = self + .conn + .prepare( + "SELECT author, version, body, signature FROM signed_records + WHERE network_id = ?1", + ) + .map_err(|err| Error::Storage(format!("cannot read signed records: {err}")))?; + let rows = stmt + .query_map(params![network_id.as_bytes().as_slice()], |row| { + let author: Vec = row.get(0)?; + let version: i64 = row.get(1)?; + let body: Vec = row.get(2)?; + let signature: Vec = row.get(3)?; + Ok((author, version, body, signature)) + }) + .map_err(|err| Error::Storage(format!("cannot read signed records: {err}")))?; + + let mut out = Vec::new(); + for row in rows { + let (author, version, body, signature) = + row.map_err(|err| Error::Storage(format!("cannot read a record row: {err}")))?; + let Ok(author) = <[u8; 32]>::try_from(author.as_slice()) else { + continue; + }; + let Ok(body) = postcard::from_bytes(&body) else { + continue; + }; + out.push(SignedRecord { + author, + network: *network_id.as_bytes(), + version: version as u64, + body, + signature, + }); + } + Ok(out) + } + + /// Stores a record received from somebody else. + pub fn put_signed_record(&self, record: &SignedRecord) -> Result<()> { + let body = postcard::to_stdvec(&record.body) + .map_err(|err| Error::Storage(format!("cannot encode a record body: {err}")))?; + self.conn + .execute( + "INSERT INTO signed_records (network_id, author, version, body, signature) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(network_id, author) DO UPDATE SET + version = excluded.version, + body = excluded.body, + signature = excluded.signature", + params![ + record.network.as_slice(), + record.author.as_slice(), + record.version as i64, + body, + record.signature + ], + ) + .map_err(|err| Error::Storage(format!("cannot store a signed record: {err}")))?; + Ok(()) + } + + /// Stores one of **our own** records and bumps our counter, atomically. + /// + /// The model requires that a record and the author's own version counter + /// are committed together, and **before** the record is published, so a + /// crash can never leave us able to reuse a version number we already put + /// on the wire. + pub fn publish_own_record(&self, record: &SignedRecord) -> Result<()> { + let body = postcard::to_stdvec(&record.body) + .map_err(|err| Error::Storage(format!("cannot encode a record body: {err}")))?; + let transaction = self + .conn + .unchecked_transaction() + .map_err(|err| Error::Storage(format!("cannot begin a transaction: {err}")))?; + + transaction + .execute( + "INSERT INTO signed_records (network_id, author, version, body, signature) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(network_id, author) DO UPDATE SET + version = excluded.version, + body = excluded.body, + signature = excluded.signature", + params![ + record.network.as_slice(), + record.author.as_slice(), + record.version as i64, + body, + record.signature + ], + ) + .map_err(|err| Error::Storage(format!("cannot store our record: {err}")))?; + transaction + .execute( + "INSERT INTO own_record_version (network_id, version) VALUES (?1, ?2) + ON CONFLICT(network_id) DO UPDATE SET + version = max(version, excluded.version)", + params![record.network.as_slice(), record.version as i64], + ) + .map_err(|err| Error::Storage(format!("cannot store our version: {err}")))?; + + transaction + .commit() + .map_err(|err| Error::Storage(format!("cannot commit our record: {err}"))) + } + + /// The highest version we have ever published for a network. + /// + /// Monotonic even if our record is later replaced by a conflicting one, + /// so we never reuse a number. + pub fn own_record_version(&self, network_id: NetworkId) -> Result { + let version: Option = self + .conn + .query_row( + "SELECT version FROM own_record_version WHERE network_id = ?1", + params![network_id.as_bytes().as_slice()], + |row| row.get(0), + ) + .optional() + .map_err(|err| Error::Storage(format!("cannot read our version: {err}")))?; + Ok(version.unwrap_or(0).max(0) as u64) + } + + /// Forgets every record of a network. + pub fn forget_signed_records(&self, network_id: NetworkId) -> Result<()> { + self.conn + .execute( + "DELETE FROM signed_records WHERE network_id = ?1", + params![network_id.as_bytes().as_slice()], + ) + .map_err(|err| Error::Storage(format!("cannot clear signed records: {err}")))?; + Ok(()) + } + /// Reads an arbitrary setting. pub fn get_setting(&self, key: &str) -> Result> { self.conn @@ -322,6 +474,23 @@ impl StateStore { } } +/// Schema for the signed records described in [`crate::state`]. +const SIGNED_RECORDS_SCHEMA: &str = "BEGIN; + CREATE TABLE IF NOT EXISTS signed_records ( + network_id BLOB NOT NULL, + author BLOB NOT NULL, + version INTEGER NOT NULL, + body BLOB NOT NULL, + signature BLOB NOT NULL, + PRIMARY KEY (network_id, author) + ); + CREATE TABLE IF NOT EXISTS own_record_version ( + network_id BLOB PRIMARY KEY, + version INTEGER NOT NULL + ); + PRAGMA user_version = 2; + COMMIT;"; + /// Seconds since the Unix epoch, saturating at 0 before it. pub(crate) fn now_unix() -> i64 { std::time::SystemTime::now() diff --git a/tests/wireguard.rs b/tests/wireguard.rs index eb0bc6f..0dda8e3 100644 --- a/tests/wireguard.rs +++ b/tests/wireguard.rs @@ -45,9 +45,33 @@ impl WgAgent { discovery: &SharedMemoryDiscovery, tag: &str, tune: impl FnOnce(WireguardConfig) -> WireguardConfig, + ) -> Self { + Self::spawn_full( + discovery, + tag, + tune, + Some(tsunagi::state::DEFAULT_IPV4_RANGE), + ) + .await + } + + /// Starts an agent with an explicit overlay range, or none at all. + async fn spawn_range( + discovery: &SharedMemoryDiscovery, + tag: &str, + range: Option, + ) -> Self { + Self::spawn_full(discovery, tag, |config| config, range).await + } + + async fn spawn_full( + discovery: &SharedMemoryDiscovery, + tag: &str, + tune: impl FnOnce(WireguardConfig) -> WireguardConfig, + range: Option, ) -> Self { let dir = TempDir::new().unwrap(); - let (agent, plugin, tuns) = Self::open(dir.path(), discovery, tag, tune).await; + let (agent, plugin, tuns) = Self::open_with(dir.path(), discovery, tag, tune, range).await; Self { dir, agent, @@ -61,6 +85,23 @@ impl WgAgent { discovery: &SharedMemoryDiscovery, tag: &str, tune: impl FnOnce(WireguardConfig) -> WireguardConfig, + ) -> (Agent, Arc, MemoryTunFactory) { + Self::open_with( + root, + discovery, + tag, + tune, + Some(tsunagi::state::DEFAULT_IPV4_RANGE), + ) + .await + } + + async fn open_with( + root: &std::path::Path, + discovery: &SharedMemoryDiscovery, + tag: &str, + tune: impl FnOnce(WireguardConfig) -> WireguardConfig, + range: Option, ) -> (Agent, Arc, MemoryTunFactory) { let tuns = MemoryTunFactory::new(); let wg = tune( @@ -72,7 +113,9 @@ impl WgAgent { .await .unwrap(); let agent = Agent::spawn( - config_with(root, discovery).with_plugin(plugin.clone() as Arc), + config_with(root, discovery) + .with_overlay_ipv4_range(range) + .with_plugin(plugin.clone() as Arc), ) .await .unwrap(); @@ -268,8 +311,8 @@ async fn the_overlay_carries_ipv4_alongside_ipv6() { let (name, secret) = network("wg-dual-stack"); let range = Some("10.77.0.0/16".parse::().unwrap()); - let a = WgAgent::spawn_with(&discovery, "ta", |c| c.with_ipv4_range(range)).await; - let b = WgAgent::spawn_with(&discovery, "tb", |c| c.with_ipv4_range(range)).await; + let a = WgAgent::spawn_range(&discovery, "ta", range).await; + let b = WgAgent::spawn_range(&discovery, "tb", range).await; let network_id = a.agent.join_network(&name, &secret).await.unwrap(); b.agent.join_network(&name, &secret).await.unwrap(); @@ -329,8 +372,8 @@ async fn an_ipv4_source_a_peer_does_not_own_is_dropped() { let (name, secret) = network("wg-v4-spoof"); let range = Some("10.78.0.0/16".parse::().unwrap()); - let a = WgAgent::spawn_with(&discovery, "ta", |c| c.with_ipv4_range(range)).await; - let b = WgAgent::spawn_with(&discovery, "tb", |c| c.with_ipv4_range(range)).await; + let a = WgAgent::spawn_range(&discovery, "ta", range).await; + let b = WgAgent::spawn_range(&discovery, "tb", range).await; let network_id = a.agent.join_network(&name, &secret).await.unwrap(); b.agent.join_network(&name, &secret).await.unwrap(); a.wait_for_tunnels(network_id, 1).await; @@ -375,13 +418,12 @@ async fn an_ipv4_source_a_peer_does_not_own_is_dropped() { } #[tokio::test] -async fn an_ipv6_only_overlay_is_the_default() { +async fn an_ipv6_only_overlay_can_be_asked_for() { let discovery = SharedMemoryDiscovery::new(); let (name, secret) = network("wg-v6-only"); - // No IPv4 range is configured, which is the default. - let a = WgAgent::spawn(&discovery, "ta").await; - let b = WgAgent::spawn(&discovery, "tb").await; + let a = WgAgent::spawn_range(&discovery, "ta", None).await; + let b = WgAgent::spawn_range(&discovery, "tb", None).await; let network_id = a.agent.join_network(&name, &secret).await.unwrap(); b.agent.join_network(&name, &secret).await.unwrap(); @@ -411,57 +453,163 @@ async fn an_ipv6_only_overlay_is_the_default() { } #[tokio::test] -async fn members_configured_with_different_ipv4_ranges_are_told_so() { +async fn a_joining_member_adopts_the_range_the_network_already_uses() { let discovery = SharedMemoryDiscovery::new(); - let (name, secret) = network("wg-range-mismatch"); + let (name, secret) = network("wg-range-adopted"); - // Two members configured differently. Deriving addresses from the range - // means they would disagree about each other, so IPv4 must be withheld - // and the mismatch reported rather than silently misrouting. - let a = WgAgent::spawn_with(&discovery, "ta", |c| { - c.with_ipv4_range(Some("10.80.0.0/16".parse().unwrap())) - }) - .await; - let b = WgAgent::spawn_with(&discovery, "tb", |c| { - c.with_ipv4_range(Some("10.81.0.0/16".parse().unwrap())) - }) - .await; + // The two were started with different ranges. Rather than misroute, they + // converge on one, and every replica computes the same answer. + let a = WgAgent::spawn_range(&discovery, "ta", Some("10.80.0.0/16".parse().unwrap())).await; + let b = WgAgent::spawn_range(&discovery, "tb", Some("10.81.0.0/16".parse().unwrap())).await; - let mut events = a.agent.subscribe(); let network_id = a.agent.join_network(&name, &secret).await.unwrap(); b.agent.join_network(&name, &secret).await.unwrap(); a.wait_for_tunnels(network_id, 1).await; + b.wait_for_tunnels(network_id, 1).await; - let reason = wait_event(&mut events, |event| match event { - Event::PluginError { reason, .. } if reason.contains("IPv4 overlay range") => { - Some(reason.clone()) - } - _ => None, + let agreed = wait_until("both settle on one range", || async { + let one = a.plugin.overview(network_id)?.ipv4_range?; + let two = b.plugin.overview(network_id)?.ipv4_range?; + (one == two).then_some(one) }) .await; - assert!(reason.contains("10.81.0.0/16"), "unexpected: {reason}"); - assert!(reason.contains("10.80.0.0/16"), "unexpected: {reason}"); - // The peer has no IPv4 here, but IPv6 is unaffected. - let view = a.plugin.overview(network_id).unwrap(); - assert_eq!(view.peers[0].overlay_address_v4, None); - assert!(view.peers[0].is_up(), "the tunnel itself still works"); + // Whichever won, both hold an address inside it, and they differ. + let view_a = wait_until("a has an address in the agreed range", || async { + let view = a.plugin.overview(network_id)?; + let address = view.overlay_address_v4?; + agreed.contains(address).then_some(view) + }) + .await; + let view_b = wait_until("b has an address in the agreed range", || async { + let view = b.plugin.overview(network_id)?; + let address = view.overlay_address_v4?; + agreed.contains(address).then_some(view) + }) + .await; + assert_ne!(view_a.overlay_address_v4, view_b.overlay_address_v4); - let addr_a = a.overlay(network_id).await; - let addr_b = b.overlay(network_id).await; - a.tun(network_id) - .await - .push_from_os(ipv6_packet(addr_a, addr_b, b"v6 still fine")); - let received = tokio::time::timeout(common::DEADLINE, b.tun(network_id).await.pop_to_os()) - .await - .expect("IPv6 should be unaffected") - .unwrap(); - assert_eq!(&received[40..], b"v6 still fine"); + // And each sees the other at the same address it sees for itself. + let seen_b = wait_until("a sees b's address", || async { + a.plugin + .overview(network_id)? + .peers + .first()? + .overlay_address_v4 + }) + .await; + assert_eq!(Some(seen_b), view_b.overlay_address_v4); a.shutdown().await; b.shutdown().await; } +#[tokio::test] +async fn an_address_is_kept_across_a_restart() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-address-persists"); + + let peer = WgAgent::spawn(&discovery, "tp").await; + let subject = WgAgent::spawn(&discovery, "ts").await; + + let network_id = peer.agent.join_network(&name, &secret).await.unwrap(); + subject.agent.join_network(&name, &secret).await.unwrap(); + peer.wait_for_tunnels(network_id, 1).await; + + let before = wait_until("the subject has an address", || async { + subject.plugin.overview(network_id)?.overlay_address_v4 + }) + .await; + // The peer agrees about it. + let seen_before = wait_until("the peer sees it", || async { + peer.plugin + .overview(network_id)? + .peers + .first()? + .overlay_address_v4 + }) + .await; + assert_eq!(seen_before, before); + + // Go away, come back. The address is a signed record, not a derivation + // and not a session fact, so it survives. + let dir = subject.shutdown().await; + let (agent, plugin, _tuns) = WgAgent::open(dir.path(), &discovery, "ts", |config| config).await; + + let after = wait_until("the restarted agent has an address", || async { + plugin.overview(network_id)?.overlay_address_v4 + }) + .await; + assert_eq!( + after, before, + "a returning participant must reclaim the address it signed for" + ); + + // And the peer still agrees, without having had to do anything. + let seen_after = wait_until("the peer still agrees", || async { + let seen = peer + .plugin + .overview(network_id)? + .peers + .first()? + .overlay_address_v4?; + (seen == after).then_some(seen) + }) + .await; + assert_eq!(seen_after, after); + + agent.shutdown().await; + peer.agent.shutdown().await; + drop(agent); + drop(dir); +} + +#[tokio::test] +async fn three_members_get_three_different_addresses() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("wg-allocation"); + + let a = WgAgent::spawn(&discovery, "ta").await; + let b = WgAgent::spawn(&discovery, "tb").await; + let c = WgAgent::spawn(&discovery, "tc").await; + + let network_id = a.agent.join_network(&name, &secret).await.unwrap(); + b.agent.join_network(&name, &secret).await.unwrap(); + c.agent.join_network(&name, &secret).await.unwrap(); + + for agent in [&a, &b, &c] { + agent.wait_for_tunnels(network_id, 2).await; + } + + // Everybody ends up knowing all three addresses, and they are distinct. + let addresses = wait_until("everyone agrees on three addresses", || async { + let mut all = std::collections::BTreeSet::new(); + for agent in [&a, &b, &c] { + let view = agent.plugin.overview(network_id)?; + all.insert(view.overlay_address_v4?); + for peer in &view.peers { + all.insert(peer.overlay_address_v4?); + } + } + (all.len() == 3).then_some(all) + }) + .await; + + let default_range = tsunagi::state::DEFAULT_IPV4_RANGE; + for address in &addresses { + assert!( + default_range.contains(*address), + "{address} outside the range" + ); + assert_ne!(address.octets()[3], 0); + assert_ne!(address.octets()[3], 255); + } + + a.shutdown().await; + b.shutdown().await; + c.shutdown().await; +} + #[tokio::test] async fn packets_for_an_unknown_address_are_counted_not_broadcast() { let discovery = SharedMemoryDiscovery::new(); @@ -727,7 +875,7 @@ async fn the_core_carries_the_payload_without_interpreting_it() { assert_eq!(capability.protocol, WIREGUARD_PROTOCOL); let view_b = b.plugin.overview(network_id).unwrap(); - let expected = WgAnnouncement::new(network_id, &view_b.public_key, None) + let expected = WgAnnouncement::new(network_id, &view_b.public_key) .encode() .unwrap(); assert_eq!(capability.data, expected); @@ -750,7 +898,7 @@ async fn a_forged_overlay_claim_is_rejected_and_never_reaches_a_tunnel() { // A legitimate member — it knows the secret — claims the victim's overlay // address with its own WireGuard key. let attacker_key = WgSecretKey::generate().public(); - let mut forged = WgAnnouncement::new(network_id, &attacker_key, None); + let mut forged = WgAnnouncement::new(network_id, &attacker_key); forged.overlay_address = victim_address; let forger = Arc::new(ForgingPlugin {