crates/installer/nordisk-pairing/src/boxloop.rs
//! **The pairing loop a product box runs itself, after the feed** (0.5.5,
//! 2026-09-27).
//!
//! [`crate::boxside`] keeps what the installer FEEDS a box; this is what the
//! box then DOES with it, until it can report PAIRED — the installer closes on
//! `CloseWhen::Paired`, so a box that never writes a paired signal never
//! finishes an install (measured on holger 2026-09-27: it wrote no signal,
//! brought no network, raised no wg0). The loop was gunnar-server's own
//! (`pairing.rs`: `turn()`, `BoxDoors`, the reporting thread); it lives here
//! once, and what only the product knows — how its primary is ready, how its
//! twin becomes the twin, is certified and follows — is the [`Steps`] half.
//!
//! ```text
//! the installer feeds (boxside) ──▶ Join: read the feed ──▶ BringTunnelUp: wg0 on the fed
//! overlay address, the front's relay owning 10.44.0.0/24 ──▶ primary: Steps::primary_ready
//! └─▶ twin: become → certify → follow
//! every change ──▶ <root>/pairing/signal.toml ──▶ Vm::status ──▶ the installer: PAIRED → close
//! ```
//!
//! The join is a READ of what the installer fed ([`crate::boxside::Installed::latest`]):
//! nothing is dialled. Until the installer has reached this box's listener
//! there is nothing to read, which is a transient — the loop asks again.
use std::collections::BTreeMap;
use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::boxside::{write_atomic, Installed, Product, SIGNAL_REL};
use crate::join::{JoinReply, Peer};
use crate::pair::{Event, Machine, Step};
use crate::seed::{Seed, TunnelKind};
use crate::signal::Signal;
use crate::tunnel::{Identity, Tunnel, TunnelError};
use crate::Role;
/// **What only the product knows** — the steps after the tunnel is up.
/// gunnar's are its control plane (`bootstrap_admin`, `BecomeTwin`, its
/// follower's heartbeat); holger's are its role record and, once
/// holger-twin-mtls lands, its certified twin.
pub trait Steps: Send + Sync {
/// PRIMARY: install what the feed hands the primary (gunnar: the fed admin
/// key and principals), then say whether this box is ready as the primary.
/// `Ok(false)` = not yet (a transient, asked again).
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String>;
/// TWIN: become the twin of the primary at `primary` (its utility address
/// when the installer fed one, else its overlay address). Idempotent.
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
/// TWIN: `(applied, head)` of this box's follower, or why not.
fn follow_state(&self) -> Result<(u64, u64), String>;
/// TWIN: is this twin certified by the primary's issuer? gunnar's answer
/// (the default): the certificate is the FRONT's move, and the twin sees it
/// took when its follower has applied an entry.
fn certified(&self, _primary: Ipv4Addr) -> Result<bool, String> {
self.follow_state().map(|(applied, _)| applied > 0)
}
}
/// **The doors the loop opens**, as a trait so [`turn`] is testable without a
/// network, a data root or a control plane. [`BoxDoors`] is the real one.
pub trait Doors {
/// What the installer fed, or why there is nothing yet.
fn join(&self) -> Result<JoinReply, String>;
/// Raise this box's tunnel on `overlay` with these peers. `Ok(true)` = up.
fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String>;
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String>;
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
fn certified(&self, primary: Ipv4Addr) -> Result<bool, String>;
fn follow_state(&self) -> Result<(u64, u64), String>;
/// **After the loop rests** (0.5.17): what the box still does for the life of the
/// process, on the loop's thread (the tunnel driver is the loop's alone). [`BoxDoors`]
/// follows the front every re-feed names ([`BoxDoors::follow_the_front`]); the default
/// holds the doors, and the tunnel with them, and does nothing else.
fn rest(&self, _say: &dyn Fn(&str)) {
loop {
std::thread::park();
}
}
}
/// The loop's memory between turns: what the feed said.
#[derive(Debug, Default, Clone)]
pub struct Learned {
pub overlay: Option<Ipv4Addr>,
/// The peers the feed named (handed to the tunnel driver).
pub peers: Vec<Peer>,
pub name: Option<String>,
pub feed: BTreeMap<String, String>,
}
/// The refusals no retry changes: a spent or forged join, a medium composed
/// without the tunnel its seed names.
const FOR_GOOD: [&str; 7] = ["[join-spent]", "[join-forged]", "[join-unknown]", "[join-estate]", "[join-withdrawn]", "[tunnel-not-built]", "[installer-link-none]"];
/// **One turn of the loop**: do what the machine asks, tell it what happened.
/// Pure over the doors — gunnar-server's `pairing::turn`, product-neutral.
pub fn turn(m: &mut Machine, learned: &mut Learned, seed: &Seed, doors: &dyn Doors) -> Step {
let step = m.next();
let event = match &step {
Step::Join | Step::FrontAwaitPeers => match doors.join() {
Ok(reply) if reply.estate != seed.estate => Event::JoinRefusedForGood(format!("the installer fed `{}`, this medium is `{}`", reply.estate, seed.estate)),
Ok(reply) => {
learned.overlay = Some(reply.me.overlay);
learned.peers = reply.peers.clone();
learned.name = Some(reply.me.name.clone());
learned.feed = reply.feed.clone();
match reply.role {
None => Event::Transient(format!("fed as `{}`; the election has not spoken yet", reply.me.name)),
Some(r) => Event::Joined { role: r, primary: reply.primary_utility.or(reply.primary_overlay), peers_admitted: reply.peers.len() as u8, peers_expected: seed.peers_expected },
}
}
Err(e) if FOR_GOOD.iter().any(|w| e.contains(w)) => Event::JoinRefusedForGood(e),
Err(e) => Event::Transient(e),
},
Step::BringTunnelUp => match learned.overlay {
// A WireGuard box dials the front it was FED; a feed that carried
// none yet (a front laid after the box) is read again here, or the
// step would retry the same stale answer forever.
Some(_) if seed.tunnel == TunnelKind::Wireguard && !learned.peers.iter().any(|p| p.role == Role::Front) => match doors.join() {
Ok(reply) if reply.estate == seed.estate && reply.front().is_some() => {
learned.peers = reply.peers.clone();
Event::Transient("the installer now feeds the front's tunnel; raising it next".into())
}
Ok(_) => Event::Transient("the installer fed no front tunnel yet".into()),
Err(e) => Event::Transient(e),
},
None => Event::Transient("no tunnel address yet".into()),
Some(o) => match doors.tunnel_up(o, &learned.peers) {
Ok(true) => Event::TunnelUp,
Ok(false) => Event::Transient(format!("the {} tunnel is not up yet", seed.tunnel.as_str())),
Err(e) if FOR_GOOD.iter().any(|w| e.contains(w)) => Event::JoinRefusedForGood(e),
Err(e) => Event::Transient(e),
},
},
Step::PrimaryCheckAdmin => match doors.join().map(|r| r.feed).map(|f| {
// The LATEST feed, not the copy the join kept: gunnar MEASURED
// 2026-09-25 that the front's offers reach the primary after its
// join and before its admin.
learned.feed = f;
doors.primary_ready(&learned.feed)
}) {
Ok(Ok(true)) => Event::AdminPresent,
Ok(Ok(false)) => Event::Transient("the primary is not ready yet (the product's own step)".into()),
Ok(Err(e)) | Err(e) => Event::Transient(e),
},
Step::TwinBecome { primary } => match doors.become_twin(*primary) {
Ok(()) => Event::TwinConferred,
Err(e) => Event::Transient(e),
},
Step::TwinCertify { primary } => match doors.certified(*primary) {
Ok(true) => Event::TwinCertified,
Ok(false) => Event::Transient("waiting for this twin's certificate (the front certifies it against the primary)".into()),
Err(e) => Event::Transient(e),
},
Step::TwinFollow { .. } => match doors.follow_state() {
Ok((applied, head)) => Event::Following { applied, head },
Err(e) => Event::Transient(e),
},
Step::Nothing => return step,
};
m.apply(event);
step
}
/// **The signal, written where [`crate::boxside::Installed`]'s `status` reads
/// it** — what the installer sees, and closes on when it says PAIRED.
pub fn report(root: &Path, seed: &Seed, name: &str, m: &Machine) -> Signal {
let s = Signal::of(name, seed.tunnel, m);
if let Some(text) = s.render() {
let _ = write_atomic(root, SIGNAL_REL, text.as_bytes(), false);
}
s
}
/// **Run the loop until PAIRED or refused for good**, reporting every change
/// (`say` gets the signal line). `pause` is the backoff's sleep (the tests
/// pass a short one). Returns the machine where it rested.
pub fn run(root: &Path, seed: &Seed, doors: &dyn Doors, say: &dyn Fn(&str), pause: &dyn Fn(Duration)) -> Machine {
let mut m = Machine::default();
let mut learned = Learned::default();
report(root, seed, "?", &m);
loop {
let before = m.clone();
let step = turn(&mut m, &mut learned, seed, doors);
let name = learned.name.clone().unwrap_or_else(|| "?".into());
if m != before {
say(report(root, seed, &name, &m).line().trim());
}
if step == Step::Nothing {
say(&if m.paired() { format!("pairing: {name} PAIRED — the loop rests") } else { format!("pairing: {name} REFUSED for good — re-seed this box") });
return m;
}
pause(Duration::from_secs(m.backoff_s()));
}
}
/// **Start the loop on its own thread.** `make` builds the doors ON that
/// thread (a tunnel driver is the loop's alone, and need not be `Send`), and
/// the thread keeps them — the tunnel with them — for the life of the process
/// once the loop rests: dropping the driver would take wg0 down.
pub fn start<D, F>(root: PathBuf, seed: Seed, make: F, say: Arc<dyn Fn(&str) + Send + Sync>) -> std::io::Result<()>
where
D: Doors + 'static,
F: FnOnce() -> D + Send + 'static,
{
start_again_after(root, seed, make, say, AGAIN_AFTER_PANIC)
}
/// The duty of the pairing loop (A28 M18).
pub const DUTY: &str = "box-pairing";
/// How long a loop that panicked waits before it runs again.
pub const AGAIN_AFTER_PANIC: Duration = Duration::from_secs(30);
/// [`start`], a loop that panicked run again after `again` (a test's seam).
pub fn start_again_after<D, F>(root: PathBuf, seed: Seed, make: F, say: Arc<dyn Fn(&str) + Send + Sync>, again: Duration) -> std::io::Result<()>
where
D: Doors + 'static,
F: FnOnce() -> D + Send + 'static,
{
// A duty (A28 M18): the doors are built on its thread and kept there, the tunnel with them;
// a loop or a rest that panics is said and runs again after `again`.
crate::duty::start_with(DUTY, again, Arc::clone(&say), move || {
let doors = make();
let (mut ran, mut rested) = (false, false);
move || {
if !ran {
let _ = run(&root, &seed, &doors, &|l| say(l), &std::thread::sleep);
ran = true;
}
if !rested {
doors.rest(&|l| say(l));
rested = true;
}
crate::duty::Round::Again
}
})?;
Ok(())
}
/// Builds the seed's tunnel driver from the seed and the latest feed (the
/// credentials a mullvad driver needs ride the feed).
pub type TunnelFactory = Box<dyn Fn(&Seed, &BTreeMap<String, String>) -> Result<Box<dyn Tunnel>, TunnelError> + Send>;
/// **The real doors of a product box**: the join is what the installer fed
/// ([`Installed::latest`]), the tunnel is the seed's driver (built at the
/// tunnel step), the rest is the product's [`Steps`].
pub struct BoxDoors<P: Product, S: Steps> {
installed: Arc<Installed<P>>,
steps: S,
make_tunnel: TunnelFactory,
tunnel: Mutex<Option<Result<Box<dyn Tunnel>, TunnelError>>>,
/// What the box's feed bus says the front is (0.5.17): every kept feed and re-feed whose
/// front, address or PSK differ, for [`Doors::rest`] to follow.
fronts: Mutex<std::sync::mpsc::Receiver<JoinReply>>,
/// What the raised tunnel dials now (0.5.17): a re-feed naming the same is no change.
dialled: Mutex<Option<Dial>>,
}
/// This box's address, the front it dials, the link's PSK.
type Dial = (Ipv4Addr, Option<Peer>, Option<String>);
/// The PSK of this box's link, as the installer fed it ([`crate::installer::FEED_WG_PSK`]),
/// trimmed. A fed but BLANK key stays `Some("")` (lane D14, decision c1) and is refused.
fn fed_psk(reply: &JoinReply) -> Option<String> {
reply.feed.get(crate::installer::FEED_WG_PSK).map(|p| p.trim().to_string())
}
/// A fed but blank PSK refuses the tunnel by name, whatever the driver (lane D14, decision c1).
fn refuse_blank(psk: Option<&str>) -> Result<(), String> {
match psk {
Some("") => Err("REFUSED [wg-psk-blank] the installer fed this box a BLANK WireGuard pre-shared key; its tunnel is never raised without its key".into()),
_ => Ok(()),
}
}
/// The front a dial names, for a log line: `<key, 8 characters>@<endpoint>` (Gunnar's
/// `peers_said`), `-` for none. The PSK is never said.
fn dial_said(d: Option<&Dial>) -> String {
match d.and_then(|d| d.1.as_ref()) {
Some(f) => format!("{}@{}", f.tunnel_pubkey.trim().chars().take(8).collect::<String>(), f.endpoint.as_deref().unwrap_or("-")),
None => "-".into(),
}
}
/// What the tunnel dials, as a reply names it: this box's address, the front, the PSK.
fn dial_of(reply: &JoinReply) -> Dial {
(reply.me.overlay, reply.front().cloned(), fed_psk(reply))
}
/// How often the rested loop looks for the first handshake after a re-point (0.5.18).
#[cfg(not(test))]
pub const WATCH_EVERY: Duration = Duration::from_secs(3);
#[cfg(test)]
pub const WATCH_EVERY: Duration = Duration::from_millis(20);
/// How many looks before the rested loop says that nobody answered (0.5.18).
pub const WATCH_ATTEMPTS: u32 = 20;
impl<P: Product, S: Steps> BoxDoors<P, S> {
pub fn new(installed: Arc<Installed<P>>, steps: S, make_tunnel: TunnelFactory) -> BoxDoors<P, S> {
let (tx, rx) = std::sync::mpsc::channel();
let tx = Mutex::new(tx);
let last: Mutex<Option<Dial>> = Mutex::new(None);
installed.feed_bus().on_published("boxloop-follow-the-front", move |r: &JoinReply| {
let now = dial_of(r);
let mut seen = last.lock().unwrap_or_else(|p| p.into_inner());
if seen.as_ref() != Some(&now) {
*seen = Some(now);
let _ = tx.lock().unwrap_or_else(|p| p.into_inner()).send(r.clone());
}
});
BoxDoors { installed, steps, make_tunnel, tunnel: Mutex::new(None), fronts: Mutex::new(rx), dialled: Mutex::new(None) }
}
/// **Follow the front a re-feed names** (0.5.17, A7 row 7; Gunnar 907f4f99
/// `Installed::follow_the_front`): a FRONT recharged onto new addresses, with a new relay
/// key, re-feeds main and twin; the loop's own driver is re-pointed at the relay as the
/// re-feed names it ([`Tunnel::repoint`], with the re-feed's PSK), at once — before, a
/// closed box kept the dead relay until its next start, and the recharged front could not
/// reach it. `Ok(Some(words))` when the dial changed; `Ok(None)` when it did not, or when
/// no tunnel is raised yet (the tunnel step takes the latest feed) or the estate runs none.
pub fn follow_the_front(&self, reply: &JoinReply) -> Result<Option<String>, String> {
if self.installed.seed().tunnel == TunnelKind::None || reply.estate != self.installed.seed().estate {
return Ok(None);
}
let Some(front) = reply.front() else { return Ok(None) };
let mut slot = self.tunnel.lock().map_err(|e| e.to_string())?;
let Some(Ok(driver)) = slot.as_mut() else { return Ok(None) };
let mut dialled = self.dialled.lock().map_err(|e| e.to_string())?;
let now = dial_of(reply);
if dialled.is_none() || dialled.as_ref() == Some(&now) {
// No tunnel raised yet (the tunnel step takes the latest feed), or the same dial.
return Ok(None);
}
refuse_blank(now.2.as_deref())?;
let psk = driver.set_preshared_key(now.2.as_deref()).map_err(|e| e.to_string())?;
let me = Identity { overlay: reply.me.overlay, pubkey: driver.public_key() };
let moved = driver.repoint(&me, &reply.peers).map_err(|e| e.to_string())?;
let from = dial_said(dialled.as_ref());
let to = dial_said(Some(&now));
*dialled = Some(now);
Ok((moved || psk).then(|| {
format!(
"pairing: re-fed — {} RE-POINTED from [{from}] to [{to}]: it dials the front `{}` at {} as the re-feed names it, without a restart",
self.installed.seed().tunnel.as_str(),
front.name,
front.endpoint.as_deref().unwrap_or("-")
)
}))
}
/// The tunnel's status, once a driver is built.
pub fn tunnel_status(&self) -> Option<crate::tunnel::TunnelStatus> {
match self.tunnel.lock().ok()?.as_ref()? {
Ok(t) => Some(t.status()),
Err(_) => None,
}
}
}
impl<P: Product, S: Steps> Doors for BoxDoors<P, S> {
fn join(&self) -> Result<JoinReply, String> {
self.installed
.latest()
.ok_or_else(|| "waiting for the installer: it connects to this VM's bootstrap listener and feeds it (name, tunnel address, role, peers)".to_string())
}
fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
let mut slot = self.tunnel.lock().map_err(|e| e.to_string())?;
if slot.is_none() {
let feed = self.installed.latest().map(|r| r.feed).unwrap_or_default();
*slot = Some((self.make_tunnel)(self.installed.seed(), &feed));
}
let driver = slot.as_mut().expect("built above").as_mut().map_err(|e| e.to_string())?;
// The link's PSK, as fed (0.5.17, A7 row 13): set before the raise.
let psk = self.installed.latest().as_ref().and_then(fed_psk);
refuse_blank(psk.as_deref())?;
driver.set_preshared_key(psk.as_deref()).map_err(|e| e.to_string())?;
let me = Identity { overlay, pubkey: driver.public_key() };
driver.up(&me, peers).map_err(|e| e.to_string())?;
if let Ok(mut d) = self.dialled.lock() {
*d = Some((overlay, peers.iter().find(|p| p.role == Role::Front).cloned(), psk));
}
Ok(driver.status().up)
}
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
self.steps.primary_ready(feed)
}
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
self.steps.become_twin(primary)
}
fn certified(&self, primary: Ipv4Addr) -> Result<bool, String> {
self.steps.certified(primary)
}
fn follow_state(&self) -> Result<(u64, u64), String> {
self.steps.follow_state()
}
/// The rested loop follows the front every re-feed names, for the life of the process,
/// and watches a re-pointed tunnel for its first handshake (0.5.18, Gunnar's
/// `handshake_watch`): said ONCE, that the front answered or that nobody did.
fn rest(&self, say: &dyn Fn(&str)) {
use std::sync::mpsc::RecvTimeoutError;
// After a re-point: the front's name and the looks left.
let mut watch: Option<(String, u32)> = None;
loop {
let next = {
let rx = self.fronts.lock().unwrap_or_else(|p| p.into_inner());
if watch.is_some() { rx.recv_timeout(WATCH_EVERY) } else { rx.recv().map_err(|_| RecvTimeoutError::Disconnected) }
};
match next {
Ok(reply) => match self.follow_the_front(&reply) {
Ok(Some(words)) => {
say(&words);
watch = reply.front().map(|f| (f.name.clone(), WATCH_ATTEMPTS));
}
Ok(None) => {}
Err(e) => say(&format!("pairing: re-fed — the tunnel could not follow the front now ({e}); the next start raises it from the feed")),
},
Err(RecvTimeoutError::Timeout) => {
let Some((front, left)) = watch.take() else { continue };
let reached = self.tunnel_status().is_some_and(|s| s.reachable.contains(&front));
if reached {
say(&format!("pairing: re-pointed — WireGuard HANDSHAKE COMPLETED with the front `{front}`: this box is reachable over the tunnel again"));
} else if left <= 1 {
say(&format!(
"pairing: re-pointed — NO handshake with the front `{front}` within {:?}: the tunnel dials it and nothing answers; check the front's relay endpoint, its key, and that UDP reaches it",
WATCH_EVERY * WATCH_ATTEMPTS
));
} else {
watch = Some((front, left - 1));
}
}
Err(RecvTimeoutError::Disconnected) => break,
}
}
loop {
std::thread::park();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::join::Assigned;
use crate::pair::State;
use std::cell::RefCell;
fn seed() -> Seed {
Seed {
estate: "holger.test".into(),
tunnel: TunnelKind::Wireguard,
front: Default::default(),
kind: crate::seed::Kind::Holger,
installer_ca_pem: String::new(),
bootstrap_cert_pem: String::new(),
bootstrap_key_pem: String::new(),
peers_expected: 3,
mullvad: None,
tailscale: None,
installer_link: None,
}
}
fn front() -> Peer {
Peer { name: "front".into(), role: Role::Front, tunnel_pubkey: "F".into(), overlay: crate::OVERLAY_FRONT, endpoint: Some("10.13.8.1:51820".into()), utility: None }
}
fn reply(role: Option<Role>, with_front: bool) -> JoinReply {
JoinReply {
estate: "holger.test".into(),
me: Assigned { name: if role == Some(Role::Twin) { "vm3".into() } else { "vm2".into() }, overlay: crate::OVERLAY_BOX_A },
cert_pem: None,
role,
primary_overlay: Some(crate::OVERLAY_BOX_A),
primary_utility: Some(Ipv4Addr::new(10, 13, 8, 2)),
peers: if with_front { vec![front()] } else { Vec::new() },
feed: BTreeMap::from([("k".to_string(), "v".to_string())]),
}
}
/// Fake doors: the feeds in order (the last one repeats), and a record of
/// every door asked.
struct Fake {
feeds: RefCell<Vec<Result<JoinReply, String>>>,
asked: RefCell<Vec<String>>,
applied: RefCell<u64>,
}
impl Fake {
fn new(feeds: Vec<Result<JoinReply, String>>) -> Fake {
Fake { feeds: RefCell::new(feeds), asked: RefCell::new(Vec::new()), applied: RefCell::new(0) }
}
}
impl Doors for Fake {
fn join(&self) -> Result<JoinReply, String> {
self.asked.borrow_mut().push("join".into());
let mut f = self.feeds.borrow_mut();
if f.len() > 1 {
f.remove(0)
} else {
f[0].clone()
}
}
fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
self.asked.borrow_mut().push(format!("tunnel {overlay} {}", peers.len()));
Ok(true)
}
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
self.asked.borrow_mut().push(format!("primary {}", feed.len()));
Ok(true)
}
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
self.asked.borrow_mut().push(format!("become {primary}"));
Ok(())
}
fn certified(&self, _: Ipv4Addr) -> Result<bool, String> {
self.asked.borrow_mut().push("certified".into());
*self.applied.borrow_mut() += 1;
Ok(*self.applied.borrow() > 1)
}
fn follow_state(&self) -> Result<(u64, u64), String> {
Ok((*self.applied.borrow(), 2))
}
}
/// Doors whose first join PANICS, then pair as a primary; `rest` says it rested.
struct Breaks {
fake: Fake,
joins: std::sync::atomic::AtomicU32,
rested: Arc<std::sync::atomic::AtomicBool>,
}
impl Doors for Breaks {
fn join(&self) -> Result<JoinReply, String> {
if self.joins.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
panic!("the first join breaks");
}
self.fake.join()
}
fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
self.fake.tunnel_up(overlay, peers)
}
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
self.fake.primary_ready(feed)
}
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
self.fake.become_twin(primary)
}
fn certified(&self, primary: Ipv4Addr) -> Result<bool, String> {
self.fake.certified(primary)
}
fn follow_state(&self) -> Result<(u64, u64), String> {
self.fake.follow_state()
}
fn rest(&self, _say: &dyn Fn(&str)) {
self.rested.store(true, std::sync::atomic::Ordering::SeqCst);
loop {
std::thread::park();
}
}
}
/// ★ **A pairing loop that panics is said and runs again** (A28 M18): the raw thread ended
/// at the panic, its tunnel with it, and the box never paired, without a word.
#[test]
fn a_pairing_loop_that_panics_is_said_and_runs_again() {
let _one = crate::duty::tests::ONE.lock().unwrap_or_else(|p| p.into_inner());
let root = std::env::temp_dir().join(format!("np-boxloop-{}", std::process::id()));
std::fs::create_dir_all(&root).unwrap();
let rested = Arc::new(std::sync::atomic::AtomicBool::new(false));
let lines = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let (r, l) = (Arc::clone(&rested), Arc::clone(&lines));
let say: Arc<dyn Fn(&str) + Send + Sync> = Arc::new(move |s: &str| l.lock().unwrap().push(s.to_string()));
start_again_after(root.clone(), seed(), move || Breaks { fake: Fake::new(vec![Ok(reply(Some(Role::Primary), true))]), joins: Default::default(), rested: r }, say, Duration::from_millis(10)).unwrap();
let t = std::time::Instant::now();
while !rested.load(std::sync::atomic::Ordering::SeqCst) && t.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(5));
}
assert!(rested.load(std::sync::atomic::Ordering::SeqCst), "the loop ran again and rests: {:?}", lines.lock().unwrap());
assert!(lines.lock().unwrap().iter().any(|l| l.contains("ALERT box-pairing") && l.contains("the first join breaks")), "the panic is said: {:?}", lines.lock().unwrap());
assert!(crate::duty::find(DUTY).is_some(), "the loop is a listed duty");
let _ = std::fs::remove_dir_all(&root);
}
fn walk(doors: &Fake, max: usize) -> (Machine, Vec<&'static str>) {
let (mut m, mut l, s) = (Machine::default(), Learned::default(), seed());
let mut steps = Vec::new();
for _ in 0..max {
let step = turn(&mut m, &mut l, &s, doors);
steps.push(step.name());
if step == Step::Nothing {
break;
}
}
(m, steps)
}
#[test]
fn a_primary_waits_for_the_feed_raises_its_tunnel_on_the_fed_address_and_is_paired() {
let doors = Fake::new(vec![Err("waiting for the installer".into()), Ok(reply(None, true)), Ok(reply(Some(Role::Primary), true))]);
let (m, steps) = walk(&doors, 20);
assert_eq!(m.state, State::Paired { role: Role::Primary });
assert_eq!(steps, ["join", "join", "join", "tunnel-up", "check-admin", "nothing"], "no feed and no role are both transients");
assert!(doors.asked.borrow().contains(&format!("tunnel {} 1", crate::OVERLAY_BOX_A)), "{:?}", doors.asked.borrow());
}
#[test]
fn a_twin_becomes_the_twin_of_the_primarys_utility_address_is_certified_and_follows() {
let doors = Fake::new(vec![Ok(reply(Some(Role::Twin), true))]);
let (m, steps) = walk(&doors, 20);
assert_eq!(m.state, State::Paired { role: Role::Twin });
assert_eq!(steps, ["join", "tunnel-up", "become-twin", "certify", "certify", "follow", "nothing"]);
assert!(doors.asked.borrow().contains(&"become 10.13.8.2".to_string()), "the utility address wins: {:?}", doors.asked.borrow());
}
#[test]
fn a_wireguard_feed_without_the_front_is_read_again_before_the_tunnel_and_another_estate_is_refused() {
let doors = Fake::new(vec![Ok(reply(Some(Role::Primary), false)), Ok(reply(Some(Role::Primary), false)), Ok(reply(Some(Role::Primary), true))]);
let (m, steps) = walk(&doors, 20);
assert!(m.paired());
assert_eq!(&steps[..4], ["join", "tunnel-up", "tunnel-up", "tunnel-up"], "no front: the feed is read again, never a tunnel to nobody");
let other = Fake::new(vec![Ok(JoinReply { estate: "other".into(), ..reply(Some(Role::Primary), true) })]);
let (m, _) = walk(&other, 5);
assert!(matches!(m.state, State::Refused { .. }));
let unbuilt = Fake::new(vec![Err("REFUSED [tunnel-not-built] …".into())]);
assert!(matches!(walk(&unbuilt, 5).0.state, State::Refused { .. }), "a medium without its driver is refused for good");
}
#[test]
fn run_writes_the_signal_the_installer_reads_and_rests_paired() {
let root = std::env::temp_dir().join(format!("boxloop-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
let doors = Fake::new(vec![Ok(reply(Some(Role::Primary), true))]);
let said = RefCell::new(Vec::new());
let m = run(&root, &seed(), &doors, &|l| said.borrow_mut().push(l.to_string()), &|_| {});
assert!(m.paired());
let s = Signal::parse(&std::fs::read_to_string(root.join(SIGNAL_REL)).unwrap()).unwrap();
assert!(s.paired && s.name == "vm2" && s.role == Some(Role::Primary), "{s:?}");
assert!(said.borrow().iter().any(|l| l.ends_with("PAIRED")), "{:?}", said.borrow());
let _ = std::fs::remove_dir_all(&root);
}
/// A tunnel driver that only says what it was asked (the real one needs CAP_NET_ADMIN).
struct Said(Arc<Mutex<Vec<String>>>);
impl Tunnel for Said {
fn kind(&self) -> TunnelKind {
TunnelKind::Wireguard
}
fn public_key(&self) -> String {
"KB".into()
}
fn up(&mut self, me: &Identity, peers: &[Peer]) -> Result<(), TunnelError> {
let front = peers.iter().find(|p| p.role == Role::Front).and_then(|p| p.endpoint.clone()).unwrap_or_default();
self.0.lock().unwrap().push(format!("up {} {front}", me.overlay));
Ok(())
}
fn repoint(&mut self, me: &Identity, peers: &[Peer]) -> Result<bool, TunnelError> {
let front = peers.iter().find(|p| p.role == Role::Front).and_then(|p| p.endpoint.clone()).unwrap_or_default();
self.0.lock().unwrap().push(format!("repoint {} {front}", me.overlay));
Ok(true)
}
fn set_preshared_key(&mut self, psk: Option<&str>) -> Result<bool, TunnelError> {
self.0.lock().unwrap().push(format!("psk {}", psk.unwrap_or("-")));
Ok(psk.is_some())
}
fn add_peer(&mut self, _: &Peer) -> Result<(), TunnelError> {
Ok(())
}
/// The front is reachable once the test logs `handshake` (a real pump's first handshake).
fn status(&self) -> crate::tunnel::TunnelStatus {
let reachable = if self.0.lock().unwrap().iter().any(|l| l == "handshake") { vec!["front".to_string()] } else { Vec::new() };
crate::tunnel::TunnelStatus { kind: TunnelKind::Wireguard, up: true, reachable, detail: String::new() }
}
}
struct Box2;
impl Product for Box2 {
fn kind(&self) -> crate::seed::Kind {
crate::seed::Kind::Holger
}
fn tunnel_pubkey(&self) -> String {
"KB".into()
}
fn capacity(&self) -> Result<Option<crate::elect::Capacity>, String> {
Ok(None)
}
fn offer(&self, _: Option<&JoinReply>) -> BTreeMap<String, String> {
BTreeMap::new()
}
fn fed(&self, _: &JoinReply) {}
}
struct NoSteps;
impl Steps for NoSteps {
fn primary_ready(&self, _: &BTreeMap<String, String>) -> Result<bool, String> {
Ok(true)
}
fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
Ok(())
}
fn follow_state(&self) -> Result<(u64, u64), String> {
Ok((1, 1))
}
}
const PSK: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
fn fed_box(tag: &str, front_at: &str) -> (Arc<Installed<Box2>>, JoinReply, PathBuf) {
let root = std::env::temp_dir().join(format!("boxloop-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
let installed = Arc::new(Installed::open(&root, seed(), Box2).unwrap());
let mut r = reply(Some(Role::Primary), true);
r.peers[0].endpoint = Some(front_at.into());
r.feed.insert(crate::installer::FEED_WG_PSK.into(), PSK.into());
use crate::bootstrap::Vm as _;
installed.fed(&r).unwrap();
installed.close().unwrap();
(installed, r, root)
}
fn said_doors(installed: &Arc<Installed<Box2>>, log: &Arc<Mutex<Vec<String>>>) -> BoxDoors<Box2, NoSteps> {
let l = Arc::clone(log);
BoxDoors::new(Arc::clone(installed), NoSteps, Box::new(move |_, _| Ok(Box::new(Said(Arc::clone(&l))) as Box<dyn Tunnel>)))
}
/// ★ 0.5.17 (A7 row 13, Gunnar 356e8dd2): the box's link to the front carries the PSK the
/// installer fed it (`wg_psk`), set on the driver before the tunnel is raised.
#[test]
fn the_box_raises_its_link_with_the_psk_the_installer_fed() {
let (installed, r, root) = fed_box("psk", "10.13.8.1:51820");
let log = Arc::new(Mutex::new(Vec::new()));
let doors = said_doors(&installed, &log);
assert!(doors.tunnel_up(r.me.overlay, &r.peers).unwrap());
assert_eq!(*log.lock().unwrap(), vec![format!("psk {PSK}"), format!("up {} 10.13.8.1:51820", r.me.overlay)]);
let _ = std::fs::remove_dir_all(&root);
}
/// ★ Lane D14, decision c1: a `wg_psk` that was fed BLANK refuses the box's tunnel by name;
/// it is never raised without a key (a silent downgrade on a security gateway).
#[test]
fn a_blank_fed_psk_refuses_the_boxs_tunnel_instead_of_raising_it_without_one() {
let (installed, mut r, root) = fed_box("psk-blank", "10.13.8.1:51820");
r.feed.insert(crate::installer::FEED_WG_PSK.into(), " ".into());
use crate::bootstrap::Vm as _;
let _ = installed.fed(&r);
let log = Arc::new(Mutex::new(Vec::new()));
let doors = said_doors(&installed, &log);
let e = doors.tunnel_up(r.me.overlay, &r.peers).unwrap_err();
assert!(e.contains("[wg-psk-blank]"), "{e}");
assert!(!log.lock().unwrap().iter().any(|l| l.starts_with("up ")), "never raised: {:?}", log.lock().unwrap());
let _ = std::fs::remove_dir_all(&root);
}
/// ★ 0.5.17 (A7 row 7, Gunnar 907f4f99): a closed box RE-FED after a FRONT recharge (the
/// front on a new address with a new relay key) re-points its tunnel at the new relay at
/// once, without a restart: through the loop's own driver, from the rested loop.
#[test]
fn a_closed_box_refed_after_a_front_recharge_dials_the_new_relay() {
let (installed, r, root) = fed_box("refeed", "127.86.33.51:51821");
let log = Arc::new(Mutex::new(Vec::new()));
let (i, l) = (Arc::clone(&installed), Arc::clone(&log));
let overlay = r.me.overlay;
std::thread::spawn(move || {
let doors = said_doors(&i, &l);
doors.tunnel_up(overlay, &i.latest().unwrap().peers).unwrap();
doors.rest(&|_| {});
});
let wait_for = |want: &str| {
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while std::time::Instant::now() < deadline && !log.lock().unwrap().iter().any(|l| l == want) {
std::thread::sleep(Duration::from_millis(20));
}
log.lock().unwrap().clone()
};
let up = wait_for(&format!("up {overlay} 127.86.33.51:51821"));
assert!(up.contains(&format!("up {overlay} 127.86.33.51:51821")), "{up:?}");
// The front is recharged onto a new address with a new relay key.
let mut moved = r.clone();
moved.peers[0].endpoint = Some("127.86.33.10:51821".into());
moved.peers[0].tunnel_pubkey = "F2".into();
use crate::bootstrap::Vm as _;
installed.refed(&moved).expect("the re-feed is taken");
let after = wait_for(&format!("repoint {overlay} 127.86.33.10:51821"));
assert!(after.contains(&format!("repoint {overlay} 127.86.33.10:51821")), "the box still dials the dead relay: {after:?}");
let _ = std::fs::remove_dir_all(&root);
}
/// 0.5.17: a re-feed that names the SAME front changes nothing; one before any tunnel is
/// raised is the tunnel step's to take.
#[test]
fn a_refeed_naming_the_same_front_or_before_the_tunnel_repoints_nothing() {
let (installed, r, root) = fed_box("same", "10.13.8.1:51820");
let log = Arc::new(Mutex::new(Vec::new()));
let doors = said_doors(&installed, &log);
assert_eq!(doors.follow_the_front(&r).unwrap(), None, "no tunnel yet: nothing to re-point");
doors.tunnel_up(r.me.overlay, &r.peers).unwrap();
assert_eq!(doors.follow_the_front(&r).unwrap(), None, "the same front: nothing to do");
let mut moved = r.clone();
moved.peers[0].endpoint = Some("10.13.8.9:51820".into());
assert!(doors.follow_the_front(&moved).unwrap().is_some_and(|w| w.contains("10.13.8.9:51820")));
assert!(log.lock().unwrap().iter().any(|l| l.starts_with("repoint ")));
let _ = std::fs::remove_dir_all(&root);
}
/// ★ 0.5.18 (lane G7, Gunnar 7e33c0f7 `wg::raise_from_config`): the re-point SAYS the old
/// and the new peer, `<key>@<endpoint>` each, as Gunnar's `WireGuard RE-POINTED` line does.
#[test]
fn a_repoint_says_the_old_and_the_new_peer() {
let (installed, r, root) = fed_box("said-peers", "10.13.8.1:51820");
let log = Arc::new(Mutex::new(Vec::new()));
let doors = said_doors(&installed, &log);
doors.tunnel_up(r.me.overlay, &r.peers).unwrap();
let mut moved = r.clone();
moved.peers[0].endpoint = Some("10.13.8.9:51820".into());
moved.peers[0].tunnel_pubkey = "F2NEWRELAYKEY".into();
let words = doors.follow_the_front(&moved).unwrap().expect("re-pointed");
assert!(words.contains("RE-POINTED") && words.contains("from [F@10.13.8.1:51820] to [F2NEWREL@10.13.8.9:51820]"), "{words}");
let _ = std::fs::remove_dir_all(&root);
}
type Lines = Arc<Mutex<Vec<String>>>;
fn rested(tag: &str) -> (Arc<Installed<Box2>>, JoinReply, PathBuf, Lines, Lines) {
let (installed, r, root) = fed_box(tag, "127.86.33.51:51821");
let log = Arc::new(Mutex::new(Vec::new()));
let said = Arc::new(Mutex::new(Vec::new()));
let (i, l, s) = (Arc::clone(&installed), Arc::clone(&log), Arc::clone(&said));
let overlay = r.me.overlay;
std::thread::spawn(move || {
let doors = said_doors(&i, &l);
doors.tunnel_up(overlay, &i.latest().unwrap().peers).unwrap();
doors.rest(&|line| s.lock().unwrap().push(line.to_string()));
});
(installed, r, root, log, said)
}
fn wait_said(said: &Arc<Mutex<Vec<String>>>, want: &str) -> Vec<String> {
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while std::time::Instant::now() < deadline && !said.lock().unwrap().iter().any(|l| l.contains(want)) {
std::thread::sleep(Duration::from_millis(10));
}
said.lock().unwrap().clone()
}
fn recharge(installed: &Arc<Installed<Box2>>, r: &JoinReply) {
let mut moved = r.clone();
moved.peers[0].endpoint = Some("127.86.33.10:51821".into());
moved.peers[0].tunnel_pubkey = "F2".into();
use crate::bootstrap::Vm as _;
installed.refed(&moved).expect("the re-feed is taken");
}
/// ★ 0.5.18 (lane G7, Gunnar 7e33c0f7 `wg::handshake_watch` after a re-point): the rested
/// loop watches the re-pointed tunnel and says ONCE when the front answered.
#[test]
fn the_rested_loop_says_the_handshake_after_a_repoint() {
let (installed, r, root, log, said) = rested("watch-ok");
while !log.lock().unwrap().iter().any(|l| l.starts_with("up ")) {
std::thread::sleep(Duration::from_millis(5));
}
recharge(&installed, &r);
assert!(wait_said(&said, "RE-POINTED").iter().any(|l| l.contains("RE-POINTED")));
log.lock().unwrap().push("handshake".into());
let heard = wait_said(&said, "HANDSHAKE COMPLETED");
assert!(heard.iter().any(|l| l.contains("HANDSHAKE COMPLETED") && l.contains("`front`")), "{heard:?}");
std::thread::sleep(WATCH_EVERY * 3);
assert_eq!(said.lock().unwrap().iter().filter(|l| l.contains("HANDSHAKE")).count(), 1, "said once");
let _ = std::fs::remove_dir_all(&root);
}
/// ★ 0.5.18 (lane G7): a re-pointed tunnel nobody answers is said ONCE, with a verdict,
/// after the watch's budget — never a line per poll, never silence.
#[test]
fn the_rested_loop_says_no_handshake_once_after_its_budget() {
let (installed, r, root, log, said) = rested("watch-none");
while !log.lock().unwrap().iter().any(|l| l.starts_with("up ")) {
std::thread::sleep(Duration::from_millis(5));
}
recharge(&installed, &r);
let heard = wait_said(&said, "NO handshake");
assert!(heard.iter().any(|l| l.contains("NO handshake") && l.contains("`front`")), "{heard:?}");
std::thread::sleep(WATCH_EVERY * 3);
assert_eq!(said.lock().unwrap().iter().filter(|l| l.contains("handshake")).count(), 1, "said once: {:?}", said.lock().unwrap());
let _ = std::fs::remove_dir_all(&root);
}
}