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;
pub trait Steps: Send + Sync {
fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String>;
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
fn follow_state(&self) -> Result<(u64, u64), String>;
fn certified(&self, _primary: Ipv4Addr) -> Result<bool, String> {
self.follow_state().map(|(applied, _)| applied > 0)
}
}
pub trait Doors {
fn join(&self) -> Result<JoinReply, String>;
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>;
fn rest(&self, _say: &dyn Fn(&str)) {
loop {
std::thread::park();
}
}
}
#[derive(Debug, Default, Clone)]
pub struct Learned {
pub overlay: Option<Ipv4Addr>,
pub peers: Vec<Peer>,
pub name: Option<String>,
pub feed: BTreeMap<String, String>,
}
const FOR_GOOD: [&str; 7] = ["[join-spent]", "[join-forged]", "[join-unknown]", "[join-estate]", "[join-withdrawn]", "[tunnel-not-built]", "[installer-link-none]"];
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 {
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| {
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
}
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
}
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()));
}
}
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,
{
std::thread::Builder::new().name("box-pairing".into()).spawn(move || {
let doors = make();
let _ = run(&root, &seed, &doors, &|l| say(l), &std::thread::sleep);
doors.rest(&|l| say(l));
loop {
std::thread::park();
}
})?;
Ok(())
}
pub type TunnelFactory = Box<dyn Fn(&Seed, &BTreeMap<String, String>) -> Result<Box<dyn Tunnel>, TunnelError> + Send>;
pub struct BoxDoors<P: Product, S: Steps> {
installed: Arc<Installed<P>>,
steps: S,
make_tunnel: TunnelFactory,
tunnel: Mutex<Option<Result<Box<dyn Tunnel>, TunnelError>>>,
fronts: Mutex<std::sync::mpsc::Receiver<JoinReply>>,
dialled: Mutex<Option<Dial>>,
}
type Dial = (Ipv4Addr, Option<Peer>, Option<String>);
fn fed_psk(reply: &JoinReply) -> Option<String> {
reply.feed.get(crate::installer::FEED_WG_PSK).map(|p| p.trim().to_string())
}
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(()),
}
}
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(),
}
}
fn dial_of(reply: &JoinReply) -> Dial {
(reply.me.overlay, reply.front().cloned(), fed_psk(reply))
}
#[cfg(not(test))]
pub const WATCH_EVERY: Duration = Duration::from_secs(3);
#[cfg(test)]
pub const WATCH_EVERY: Duration = Duration::from_millis(20);
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) }
}
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) {
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("-")
)
}))
}
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())?;
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()
}
fn rest(&self, say: &dyn Fn(&str)) {
use std::sync::mpsc::RecvTimeoutError;
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())]),
}
}
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))
}
}
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);
}
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(())
}
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>)))
}
#[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);
}
#[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);
}
#[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:?}");
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);
}
#[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);
}
#[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");
}
#[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);
}
#[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);
}
}