nordisk-pairing 0.5.29

//! **`pairing-probe`** — the pairing protocol with real sockets, real mutual
//! TLS and real WireGuard, for the draupnir fleet test
//! (`tests/fleet_draupnir.rs`). Since 0.3.0 the INSTALLER connects out to
//! the VMs (owner 2026-09-24), so four containers run it:
//!
//! ```text
//! installer: <shared>/plan.toml + <shared>/medium/ (the CA, the installer's
//!        own certificate, its WireGuard link keys) → nordisk_pairing::
//!        installer::run against every VM — through its user-space WireGuard
//!        link (tunnel = wireguard: no TUN, no NET_ADMIN in this container) or
//!        straight to the VM (tunnel = none). Logs into <shared>/run/, the
//!        ledger into <shared>/run/ledger.toml. Writes <shared>/installer.result.
//! vm:    <shared>/seed.toml → the installer's link (wireguard: wg-installer
//!        at 10.45.0.2, the installer its one peer) → the bootstrap listener
//!        (on the link address only, or every interface with tunnel = none)
//!        → fed → the mesh (wg0 via the front's relay, or the cloud's own
//!        addresses) → reaches the OTHER VM on :7000 → PAIRED. On close: the
//!        listener goes and the link is dropped. Writes <shared>/<name>.result.
//! front: the relay (userspace WireGuard hub, :51820), listing every VM the
//!        installer admitted — read from <shared>/run/ledger.toml, the way the
//!        xtask pushes the installer's ledger to the real front.
//! ```

use nordisk_pairing::bootlog::BootLog;
use nordisk_pairing::bootstrap::{self, Vm};
use nordisk_pairing::door::Door;
use nordisk_pairing::elect::Capacity;
use nordisk_pairing::installer::{self, FolderSink, Material, Plan};
use nordisk_pairing::installer_link::{self, Reach};
use nordisk_pairing::join::{JoinReply, JoinRequest};
use nordisk_pairing::mtls::{self, Issued};
use nordisk_pairing::pair::{Event, Machine};
use nordisk_pairing::seed::{Kind, Seed, TunnelKind};
use nordisk_pairing::signal::Signal;
use nordisk_pairing::{Role, OVERLAY_FRONT};
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpListener, TcpStream, UdpSocket};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use wg_appliance_core::{addr, dataplane, decode_key, encode_key, generate_keypair, PeerSpec, TunDevice, TunnelSpec};

const WG_PORT: u16 = 51820;
const PROBE_PORT: u16 = 7000;

fn say(line: &str) {
    println!("{line}");
    let _ = std::io::stdout().flush();
}

fn die(why: &str) -> ! {
    say(&format!("REFUSED {why}"));
    std::process::exit(1)
}

/// This container's own address on the shared network.
fn own_ipv4() -> Ipv4Addr {
    let s = UdpSocket::bind("0.0.0.0:0").unwrap_or_else(|e| die(&format!("[probe-udp] {e}")));
    s.connect("192.0.2.1:9").unwrap_or_else(|e| die(&format!("[probe-route] no route out: {e}")));
    match s.local_addr() {
        Ok(SocketAddr::V4(a)) => *a.ip(),
        other => die(&format!("[probe-addr] {other:?}")),
    }
}

fn wait_for(path: &Path, budget: Duration) -> String {
    let t0 = Instant::now();
    loop {
        if let Ok(t) = std::fs::read_to_string(path) {
            if !t.trim().is_empty() {
                return t;
            }
        }
        if t0.elapsed() > budget {
            die(&format!("[probe-wait] {} never appeared within {} s", path.display(), budget.as_secs()));
        }
        std::thread::sleep(Duration::from_millis(250));
    }
}

fn hex32(s: &str) -> [u8; 32] {
    let s = s.trim();
    if s.len() != 64 {
        die("[probe-key] the front key is not 64 hex digits");
    }
    let mut out = [0u8; 32];
    for (i, b) in out.iter_mut().enumerate() {
        *b = u8::from_str_radix(&s[2 * i..2 * i + 2], 16).unwrap_or_else(|_| die("[probe-key] not hex"));
    }
    out
}

// ── the front: the relay only ────────────────────────────────────────────────

fn front(shared: &Path) {
    let me = own_ipv4();
    std::fs::write(shared.join("front.addr"), format!("{me}\n")).unwrap_or_else(|e| die(&format!("[probe-shared] {e}")));
    say(&format!("front: utility address {me}"));
    let key = hex32(&wait_for(&shared.join("front.key"), Duration::from_secs(60)));
    let relay = wg_appliance_core::hubrelay::HubRelay::spawn(WG_PORT, key, OVERLAY_FRONT).unwrap_or_else(|e| die(&format!("[probe-relay] {e}")));
    say(&format!("front: relay on :{WG_PORT} as {OVERLAY_FRONT}; listing what the installer admits"));
    let ledger = shared.join("run").join("ledger.toml");
    let mut listed = std::collections::BTreeSet::new();
    loop {
        if let Ok(d) = Door::load(&ledger) {
            for (name, a) in &d.ledger.admitted {
                if listed.contains(name) {
                    continue;
                }
                if let (Some((_, slot)), Ok(k)) = (d.ledger.slot_of(name), decode_key(&a.tunnel_pubkey)) {
                    if slot.kind == Kind::Gunnar && relay.add_peer(k, slot.overlay).is_ok() {
                        say(&format!("front: `{name}` listed on the relay as {}", slot.overlay));
                        listed.insert(name.clone());
                    }
                }
            }
        }
        std::thread::sleep(Duration::from_millis(500));
    }
}

// ── a VM ─────────────────────────────────────────────────────────────────────

struct ProbeVm {
    seed: Seed,
    cores: u32,
    mesh_private: [u8; 32],
    mesh_public: String,
    own_key: String,
    own_cert: Mutex<Option<String>>,
    fed: Mutex<Option<JoinReply>>,
    machine: Mutex<Machine>,
    name: Mutex<String>,
    closed: AtomicBool,
    log: Arc<BootLog>,
}

impl ProbeVm {
    fn line(&self, l: &str) {
        say(l);
        self.log.push(l);
    }
}

impl Vm for ProbeVm {
    fn hello(&self) -> Result<JoinRequest, String> {
        let csr = if self.own_cert.lock().map_err(|e| e.to_string())?.is_some() { String::new() } else { mtls::box_request_for(&self.own_key).map_err(|e| e.to_string())?.0 };
        Ok(JoinRequest {
            estate: self.seed.estate.clone(),
            kind: Kind::Gunnar,
            tunnel_pubkey: if self.seed.tunnel == TunnelKind::None { String::new() } else { self.mesh_public.clone() },
            capacity: Some(Capacity { cores: self.cores, ram_bytes: 4 << 30, data_bytes: 10 << 30, disk_mib_s: 500 }),
            csr_pem: csr,
        })
    }

    fn fed(&self, reply: &JoinReply) -> Result<(), String> {
        if let Some(c) = &reply.cert_pem {
            *self.own_cert.lock().map_err(|e| e.to_string())? = Some(c.clone());
            self.line(&format!("vm: the installer signed my own certificate as `{}` — I show it from now on", reply.me.name));
        }
        *self.name.lock().map_err(|e| e.to_string())? = reply.me.name.clone();
        self.line(&format!("vm: fed as `{}` at {}, role {}, {} peer(s)", reply.me.name, reply.me.overlay, reply.role.map(|r| r.as_str()).unwrap_or("pending"), reply.peers.len()));
        *self.fed.lock().map_err(|e| e.to_string())? = Some(reply.clone());
        Ok(())
    }

    fn status(&self) -> Signal {
        let name = self.name.lock().map(|n| n.clone()).unwrap_or_default();
        let m = self.machine.lock().map(|m| m.clone()).unwrap_or_default();
        Signal::of(&name, self.seed.tunnel, &m)
    }

    fn close(&self) -> Result<(), String> {
        self.closed.store(true, Ordering::SeqCst);
        Ok(())
    }

    fn identity(&self) -> Issued {
        match self.own_cert.lock().ok().and_then(|c| c.clone()) {
            Some(c) => Issued { cert_pem: c, key_pem: self.own_key.clone() },
            None => Issued { cert_pem: self.seed.bootstrap_cert_pem.clone(), key_pem: self.seed.bootstrap_key_pem.clone() },
        }
    }
}

fn vm(shared: &Path) {
    let me = own_ipv4();
    let host = std::env::var("PROBE_NAME").unwrap_or_else(|_| me.to_string());
    std::fs::write(shared.join(format!("{host}.addr")), format!("{me}\n")).unwrap_or_else(|e| die(&format!("[probe-shared] {e}")));
    let log = Arc::new(BootLog::new(2000));
    log.push("korp-installer: install pass began (the medium's installer phase, served after first boot)\nkorp-installer: INSTALL-OK");
    say(&format!("vm {host}: utility address {me}"));
    let seed = Seed::parse(&wait_for(&shared.join("seed.toml"), Duration::from_secs(120))).unwrap_or_else(|e| die(&e.to_string()));
    let (private, public) = generate_keypair();
    let cores: u32 = std::env::var("PROBE_CORES").ok().and_then(|v| v.parse().ok()).unwrap_or(2);
    let v = Arc::new(ProbeVm {
        seed: seed.clone(),
        cores,
        mesh_private: private,
        mesh_public: encode_key(&public),
        own_key: mtls::box_request().unwrap_or_else(|e| die(&e.to_string())).1,
        own_cert: Mutex::new(None),
        fed: Mutex::new(None),
        machine: Mutex::new(Machine::default()),
        name: Mutex::new(String::new()),
        closed: AtomicBool::new(false),
        log: log.clone(),
    });
    // The installer's link FIRST, then the listener on it (or on every
    // interface with tunnel = none).
    let mut link = installer_link::vm_link(&seed).unwrap_or_else(|e| die(&e.to_string()));
    let reach = installer_link::reach(seed.tunnel).unwrap_or_else(|e| die(&e.to_string()));
    let listener = TcpListener::bind(SocketAddrV4::new(link.bind, bootstrap::PORT)).unwrap_or_else(|e| die(&format!("[probe-listen] {}:{}: {e}", link.bind, bootstrap::PORT)));
    v.line(&format!("vm {host}: bootstrap listener on {}:{} ({})", link.bind, bootstrap::PORT, if reach == Reach::Tunnel { "the installer's link only — nothing on the public interface" } else { "direct" }));
    // The pairing worker: once fed a role, the mesh and the other VM.
    let worker = {
        let v = v.clone();
        let host = host.clone();
        let shared = shared.to_path_buf();
        std::thread::spawn(move || pair_worker(&v, &host, &shared))
    };
    let ca = seed.installer_ca_pem.clone();
    let vv = v.clone();
    bootstrap::serve(vv.as_ref(), &log, &ca, listener, &|l| v.line(&format!("vm {host}: {l}")));
    // Closed: drop the installer as a peer.
    let had_link = link.up();
    link.drop_installer();
    v.line(&format!("vm {host}: closed{}", if had_link { " — the installer is dropped as a peer (the link is down)" } else { "" }));
    let _ = std::fs::write(shared.join(format!("{host}.closed")), if had_link { "listener=closed link=dropped\n" } else { "listener=closed\n" });
    let _ = worker.join();
    std::thread::sleep(Duration::from_secs(60));
}

fn pair_worker(v: &ProbeVm, host: &str, shared: &Path) {
    let t0 = Instant::now();
    let reply = loop {
        if let Some(r) = v.fed.lock().ok().and_then(|f| f.clone()) {
            let others = r.peers.iter().filter(|p| p.role != Role::Front && p.name != r.me.name).count();
            if r.role.is_some() && (v.seed.tunnel == TunnelKind::None || (r.front().is_some() && others > 0 && r.peers.iter().all(|p| p.role == Role::Front || !p.tunnel_pubkey.is_empty()))) {
                break r;
            }
        }
        if t0.elapsed() > Duration::from_secs(240) {
            die("[probe-feed] no role and peers were fed within 240 s");
        }
        std::thread::sleep(Duration::from_millis(500));
    };
    let name = reply.me.name.clone();
    let role = reply.role.expect("the loop breaks with a role");
    if let Ok(mut m) = v.machine.lock() {
        m.apply(Event::Joined { role, primary: reply.primary_utility, peers_admitted: 3, peers_expected: 3 });
    }
    // The mesh: wg0 via the front's relay, or the cloud's own addresses.
    let (listen_on, other_at) = if v.seed.tunnel == TunnelKind::None {
        // With no tunnel the "overlay" addresses ARE the cloud's own (the
        // plan lays them so): the other VM is dialled where it is.
        let other = reply.peers.iter().find(|p| p.role != Role::Front && p.name != name).unwrap_or_else(|| die("[probe-peer] no other VM in the peer list"));
        v.line(&format!("vm {host}: no tunnel — the other VM at its own address {}", other.overlay));
        (Ipv4Addr::UNSPECIFIED, other.overlay)
    } else {
        let tun = TunDevice::create("wg0", 1420).unwrap_or_else(|e| die(&format!("[probe-tun] wg0: {e}")));
        addr::assign_ipv4("wg0", reply.me.overlay, 24).unwrap_or_else(|e| die(&format!("[probe-addr] {e}")));
        addr::link_up("wg0").unwrap_or_else(|e| die(&format!("[probe-link] {e}")));
        let front = reply.front().unwrap_or_else(|| die("[probe-front-unfed]"));
        let front_key = decode_key(&front.tunnel_pubkey).unwrap_or_else(|e| die(&format!("[probe-front-key] {}", e.0)));
        let endpoint: SocketAddrV4 = front.endpoint.as_deref().unwrap_or("").parse().unwrap_or_else(|e| die(&format!("[probe-endpoint] {e}")));
        let spec = TunnelSpec {
            private_key: v.mesh_private,
            listen_port: None,
            peers: vec![PeerSpec { public_key: front_key, preshared_key: None, endpoint: Some(endpoint), allowed_ips: vec![nordisk_pairing::OVERLAY_NET], persistent_keepalive: Some(5) }],
        };
        let handle = dataplane::spawn(tun, &spec).unwrap_or_else(|e| die(&format!("[probe-pump] {e}")));
        std::mem::forget(handle);
        let other = reply.peers.iter().find(|p| p.role != Role::Front && p.name != name).unwrap_or_else(|| die("[probe-peer]"));
        v.line(&format!("vm {host}: wg0 up as {}/24 via the front's relay {endpoint}", reply.me.overlay));
        (reply.me.overlay, other.overlay)
    };
    if let Ok(mut m) = v.machine.lock() {
        m.apply(Event::TunnelUp);
    }
    let listen = TcpListener::bind(SocketAddrV4::new(listen_on, PROBE_PORT)).unwrap_or_else(|e| die(&format!("[probe-listen] {e}")));
    let me_name = name.clone();
    std::thread::spawn(move || {
        for mut s in listen.incoming().flatten() {
            let _ = s.write_all(format!("{me_name}\n").as_bytes());
        }
    });
    let target = SocketAddrV4::new(other_at, PROBE_PORT);
    let heard = loop {
        if let Ok(mut s) = TcpStream::connect_timeout(&target.into(), Duration::from_secs(3)) {
            let mut b = String::new();
            let _ = s.set_read_timeout(Some(Duration::from_secs(3)));
            let _ = s.read_to_string(&mut b);
            if !b.trim().is_empty() {
                break b.trim().to_string();
            }
        }
        if t0.elapsed() > Duration::from_secs(300) {
            die(&format!("[probe-unreached] {target} never answered"));
        }
        std::thread::sleep(Duration::from_secs(1));
    };
    if let Ok(mut m) = v.machine.lock() {
        match role {
            Role::Primary => m.apply(Event::AdminPresent),
            _ => {
                m.apply(Event::TwinConferred);
                m.apply(Event::TwinCertified);
                m.apply(Event::Following { applied: 1, head: 1 });
            }
        }
    }
    let result = format!(
        "PAIRED name={name} role={} overlay={} reached={target} heard={heard} fed={}\n",
        role.as_str(),
        reply.me.overlay,
        reply.feed.keys().cloned().collect::<Vec<_>>().join(",")
    );
    v.line(&format!("vm {host}: {}", result.trim()));
    let _ = std::fs::write(shared.join(format!("{name}.result")), &result);
}

// ── the installer ────────────────────────────────────────────────────────────

fn installer(shared: &Path) {
    let plan = Plan::parse(&wait_for(&shared.join("plan.toml"), Duration::from_secs(120))).unwrap_or_else(|e| die(&e));
    let m = Material::load(&shared.join("medium")).unwrap_or_else(|e| die(&e));
    say(&format!("installer: {} VM(s), tunnel {}, my address {}", plan.vms.len(), plan.tunnel.as_str(), own_ipv4()));
    // The medium's bootstrap certificate is NOT the installer: every VM
    // refuses it as a client. Asked once, on the first VM, before the run.
    let boot = Issued { cert_pem: std::fs::read_to_string(shared.join("medium/bootstrap.cert.pem")).unwrap_or_default(), key_pem: std::fs::read_to_string(shared.join("medium/bootstrap.key.pem")).unwrap_or_default() };
    let first = plan.vms[0].clone();
    let t0 = Instant::now();
    let as_bootstrap = loop {
        let path = installer::path_to(&plan, &first, m.link.as_ref()).unwrap_or_else(|e| die(&e));
        let mut c = installer::Client::new(path, &m.ca.cert_pem, &boot, Duration::from_secs(10)).unwrap_or_else(|e| die(&e));
        match c.hello() {
            Err(e) if e.contains("[installer-only]") => break "REFUSED [installer-only]".to_string(),
            Ok(_) => break "ACCEPTED".to_string(),
            Err(e) if t0.elapsed() > Duration::from_secs(120) => break format!("unreachable: {e}"),
            Err(_) => std::thread::sleep(Duration::from_secs(1)),
        }
    };
    say(&format!("installer: the medium's bootstrap certificate as a client: {as_bootstrap}"));
    // Nothing of the listener on the public interface in tunnel mode.
    let public = match TcpStream::connect_timeout(&SocketAddr::V4(SocketAddrV4::new(first.public, plan.port)), Duration::from_secs(3)) {
        Ok(_) => "open",
        Err(_) => "closed",
    };
    say(&format!("installer: the bootstrap port on {}'s public address: {public}", first.name));
    let sink = FolderSink::new(&shared.join("run"), true).unwrap_or_else(|e| die(&e));
    let paths = plan.vms.iter().map(|v| (v.clone(), installer::path_to(&plan, v, m.link.as_ref()).unwrap_or_else(|e| die(&e)))).collect();
    let result = match installer::run(&plan, &m, paths, &sink) {
        Ok(out) => {
            // After the close: the installer's own certificate reaches nothing.
            let path = installer::path_to(&plan, &first, m.link.as_ref()).unwrap_or_else(|e| die(&e));
            let after = match installer::Client::new(path, &m.ca.cert_pem, &m.me, Duration::from_secs(5)).unwrap_or_else(|e| die(&e)).hello() {
                Ok(_) => "ANSWERED".to_string(),
                Err(_) => "unreachable".to_string(),
            };
            format!(
                "CLOSED bootstrap_as_client={as_bootstrap} public_port={public} after_close={after} {}\n",
                out.vms.iter().map(|v| format!("{}:role={}:lines={}:gone={}", v.name, v.role.map(|r| r.as_str()).unwrap_or("-"), v.log_lines, v.gone)).collect::<Vec<_>>().join(" ")
            )
        }
        Err(e) => format!("{e}\n"),
    };
    say(&format!("installer: {}", result.trim()));
    let _ = std::fs::write(shared.join("installer.result"), &result);
    std::thread::sleep(Duration::from_secs(30));
}

fn main() {
    let args: Vec<String> = std::env::args().collect();
    let role = args.get(1).map(String::as_str).unwrap_or("");
    let shared = args.iter().position(|a| a == "--shared").and_then(|i| args.get(i + 1)).map(PathBuf::from).unwrap_or_else(|| PathBuf::from("/shared"));
    match role {
        "front" => front(&shared),
        "vm" | "box" => vm(&shared),
        "installer" => installer(&shared),
        _ => die("[probe-usage] pairing-probe front|vm|installer [--shared <dir>]"),
    }
}