nordisk-pairing 0.5.2

//! **The VM's side of the installer: the bootstrap listener** (0.3.0).
//!
//! The owner, 2026-09-24: the installer runs FIRST and lives on the
//! operator's machine — which is behind NAT, so the direction is reversed.
//! The VMs do not dial a join service on the front; every VM of the medium
//! opens THIS listener from its first moment, and the installer connects OUT
//! to it, retrying until each VM answers.
//!
//! Mutual TLS both ways under the installer CA:
//!
//! * the VM shows the medium-wide bootstrap certificate (or, after a restart,
//!   the certificate the installer signed for it), named
//!   [`crate::mtls::VM_SERVER_NAME`];
//! * the installer shows its own client certificate, CN
//!   [`crate::mtls::INSTALLER_CN`], and the VM serves THAT ROLE ONLY — the
//!   bootstrap certificate on the medium, a box certificate, a stranger's:
//!   all refused by name (`installer-only`).
//!
//! Over it the installer, one request per connection (minimal HTTP/1.1 in
//! TLS 1.3, TOML bodies like the rest of the crate):
//!
//! | request | what the VM does |
//! |---|---|
//! | `GET /install/hello` | presents itself: the [`JoinRequest`] (kind, tunnel key, capacity, a signing request for a key it minted) |
//! | `GET /install/log?from=N` | its boot log from line `N` ([`crate::bootlog`]) |
//! | `POST /install/feed` | takes the [`JoinReply`]: its name, tunnel address, own certificate, role, peers, the estate values |
//! | `GET /install/status` | its pairing [`Signal`] |
//! | `GET /install/offer` | what the VM OFFERS the installer (0.3.1): public values only it can mint — the primary's ticket key, the twin's replication key and signing request, the certificate the primary signed for its twin |
//! | `GET /install/secret-offer` | what the VM HANDS OVER as a secret (0.5.0, [`SecretOffer`]): the front's certificate and key, for its successor — asked once, before the close; never fed on, never logged |
//! | `POST /install/close` | closes the listener for good: the bootstrap certificate is dead from then on |
//!
//! **The offer is fed onward** (0.3.1): the installer feeds every VM's offer
//! to every VM of the run (the offering one too), each value under [`fed_offer_key`]
//! (`offer.<vm>.<key>`), and hands it to its [`crate::installer::Sink::offer`].
//! That is how two VMs that do not yet reach each other trade what only they
//! can mint, with the installer — already trusted both ways — as the carrier.
//!
//! What the VM does with a feed, how it keeps its log and how it remembers
//! that it was closed is the VM's ([`Vm`]); the wire and the one-role rule
//! are here, shared by gunnar and holger.

use crate::bootlog::{BootLog, MAX_READ};
use crate::door::{parse_request, Request, Response};

/// **How large one request to a VM's bootstrap listener may be: 16 MiB**
/// (0.5.1). A FEED carries vm1's own values since 0.5.0 — the front's
/// monetize plugins among them, ≈0.8 MB base64 each — and the door's 64 KiB
/// ([`crate::door::MAX_REQUEST_BYTES`], a JOIN) refused every feed to a paid
/// front: `request too large` (MEASURED 2026-09-25, lane B's bench). The
/// listener answers only over the installer's WireGuard link, under mutual TLS
/// on the installer CA, so the bound is against a runaway, not a stranger.
pub const MAX_FEED_BYTES: usize = 16 * 1024 * 1024;
use crate::join::{JoinReply, JoinRequest};
use crate::mtls;
use crate::signal::Signal;
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::time::Duration;

/// **The installer's mTLS endpoint on every VM.** The port the front's join
/// service used until 0.2 — the VM now listens on it, and the firewall opens
/// it to the operator's public address only.
pub const PORT: u16 = 8474;

pub const HELLO_PATH: &str = "/install/hello";
pub const LOG_PATH: &str = "/install/log";
pub const FEED_PATH: &str = "/install/feed";
pub const STATUS_PATH: &str = "/install/status";
pub const CLOSE_PATH: &str = "/install/close";
/// 0.3.1: what the VM offers the installer ([`Vm::offer`]).
pub const OFFER_PATH: &str = "/install/offer";

/// 0.5.0: what the VM hands over as a SECRET ([`Vm::secret_offer`]).
pub const SECRET_OFFER_PATH: &str = "/install/secret-offer";

/// **A secret offer** (0.5.0, owner 2026-09-25): values a VM hands the
/// installer that must survive it — the front's TLS certificate AND ITS
/// PRIVATE KEY (`tls.<host>.cert`, `tls.<host>.key`), so the NEXT front of
/// the estate is fed them and does not spend one of Let's Encrypt's five
/// certificates a week. Marked secret by its own path and its own type:
///
/// * the installer asks for it at the END of a VM's install, right before the
///   close ([`crate::installer`]), and hands it to
///   [`crate::installer::Sink::secret_offer`] — ONLY there;
/// * it is never fed to another VM (no `offer.<vm>.<key>`), never in the
///   door's feed, the ledger, `offers.toml`, the narration or a log — the
///   narration names its keys, never a value;
/// * [`crate::installer::FolderSink`] keeps it in `<run>/secret-offers.toml`,
///   0600; the operator's next run feeds it to the successor VM with
///   [`crate::installer::successor_feed`] (`PlanVm::feed`, that VM only).
///
/// `Debug` prints the keys only.
#[derive(Clone, Default, PartialEq, Eq)]
pub struct SecretOffer(BTreeMap<String, String>);

impl SecretOffer {
    pub fn new(values: BTreeMap<String, String>) -> SecretOffer {
        SecretOffer(values)
    }
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
    pub fn len(&self) -> usize {
        self.0.len()
    }
    /// The keys — the only part that may be said.
    pub fn keys(&self) -> Vec<String> {
        self.0.keys().cloned().collect()
    }
    /// The values, for the one place that stores them.
    pub fn expose(&self) -> &BTreeMap<String, String> {
        &self.0
    }
    pub fn into_inner(self) -> BTreeMap<String, String> {
        self.0
    }
    /// The wire body (a TOML table of strings, like the public offer).
    pub fn render(&self) -> String {
        toml::to_string(&self.0).unwrap_or_default()
    }
    /// The wire body back. The error never quotes the body.
    pub fn parse(text: &str) -> Result<SecretOffer, String> {
        toml::from_str(text).map(SecretOffer).map_err(|_| "not a secret offer (a TOML table of strings); the body is not quoted".to_string())
    }
}

impl std::fmt::Debug for SecretOffer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "SecretOffer({:?} — values withheld)", self.keys())
    }
}

/// The feed key an offered value travels under to the other VMs:
/// `offer.<vm>.<key>` — `vm` the offering VM's name as the installer knows it.
pub fn fed_offer_key(vm: &str, key: &str) -> String {
    format!("offer.{vm}.{key}")
}

/// Every value `vm` offered, out of a feed ([`fed_offer_key`] undone).
pub fn offers_in(feed: &BTreeMap<String, String>, vm: &str) -> BTreeMap<String, String> {
    let prefix = format!("offer.{vm}.");
    feed.iter().filter_map(|(k, v)| k.strip_prefix(&prefix).map(|k| (k.to_string(), v.clone()))).collect()
}

/// An offer as its wire body (a TOML table of strings).
pub fn render_offer(offer: &BTreeMap<String, String>) -> String {
    toml::to_string(offer).unwrap_or_default()
}

/// The wire body back. An empty body is an empty offer.
pub fn parse_offer(text: &str) -> Result<BTreeMap<String, String>, String> {
    toml::from_str(text).map_err(|e| format!("not an offer: {e}"))
}

/// **What a VM does when the installer asks.** Implemented by gunnar-server
/// (PID 1) and by the fleet test's probe.
pub trait Vm: Send + Sync {
    /// Present this VM: estate, kind, tunnel key, capacity, and — while it
    /// has no certificate of its own — a signing request. `Err` = not ready
    /// yet (a transient: the installer asks again).
    fn hello(&self) -> Result<JoinRequest, String>;
    /// Take what the installer fed. Called again whenever the installer's
    /// answer changes (the role is decided, a peer's key arrives).
    fn fed(&self, reply: &JoinReply) -> Result<(), String>;
    /// Where this VM's pairing is.
    fn status(&self) -> Signal;
    /// Remember, durably, that the listener was closed — it never opens again
    /// on this installation.
    fn close(&self) -> Result<(), String>;
    /// The certificate this VM shows the installer NOW: the medium's
    /// bootstrap certificate until the installer has signed the VM's own.
    fn identity(&self) -> mtls::Issued;
    /// **What this VM offers the installer** (0.3.1): PUBLIC values only it
    /// can mint, as they come into being (empty until then). The installer
    /// hands them to its sink and feeds them to the other VMs under
    /// [`fed_offer_key`]. Never a private key: the offer is logged.
    fn offer(&self) -> BTreeMap<String, String> {
        BTreeMap::new()
    }
    /// **What this VM hands over as a SECRET** (0.5.0, [`SecretOffer`]):
    /// values its SUCCESSOR must be fed — the front's current certificate and
    /// key per host. Asked once, at the end of the install, right before the
    /// close; kept by the operator only. Empty by default (the boxes).
    fn secret_offer(&self) -> SecretOffer {
        SecretOffer::default()
    }
}

/// The one-role rule, as the refusal a VM answers with.
pub fn installer_only(cn: &str) -> String {
    format!(
        "REFUSED [installer-only] this is a VM's bootstrap listener; it serves the installer's own certificate (CN `{}`) and nothing else — `{cn}` was presented",
        mtls::INSTALLER_CN
    )
}

fn query_from(path: &str) -> u64 {
    path.split_once('?')
        .map(|(_, q)| q)
        .unwrap_or("")
        .split('&')
        .find_map(|kv| kv.strip_prefix("from="))
        .and_then(|v| v.parse().ok())
        .unwrap_or(0)
}

/// **One request, answered** — pure over the VM, its log and the request.
/// The second value says whether the listener was closed by it.
pub fn respond(vm: &dyn Vm, log: &BootLog, req: &Request) -> (Response, bool) {
    let path = req.path.split('?').next().unwrap_or("");
    let ok = |body: String| Response { status: 200, body };
    match (req.method.as_str(), path) {
        ("GET", HELLO_PATH) => match vm.hello().and_then(|h| h.render().map_err(|e| e.to_string())) {
            Ok(t) => (ok(t), false),
            Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
        },
        ("GET", LOG_PATH) => (ok(log.since(query_from(&req.path), MAX_READ).render()), false),
        ("POST", FEED_PATH) => match JoinReply::parse(&String::from_utf8_lossy(&req.body)) {
            Err(e) => (Response { status: 400, body: format!("{e}\n") }, false),
            Ok(reply) => match vm.fed(&reply) {
                Ok(()) => (ok(format!("fed `{}`\n", reply.me.name)), false),
                Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
            },
        },
        ("GET", STATUS_PATH) => (ok(vm.status().render().unwrap_or_default()), false),
        ("GET", OFFER_PATH) => (ok(render_offer(&vm.offer())), false),
        ("GET", SECRET_OFFER_PATH) => (ok(vm.secret_offer().render()), false),
        ("POST", CLOSE_PATH) => match vm.close() {
            Ok(()) => (ok("closed: the bootstrap certificate is dead on this VM\n".into()), true),
            Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
        },
        (_, HELLO_PATH | LOG_PATH | FEED_PATH | STATUS_PATH | OFFER_PATH | SECRET_OFFER_PATH | CLOSE_PATH) => (Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() }, false),
        _ => (Response { status: 404, body: format!("REFUSED [install-path] this listener has {HELLO_PATH}, {LOG_PATH}, {FEED_PATH}, {STATUS_PATH}, {OFFER_PATH}, {SECRET_OFFER_PATH} and {CLOSE_PATH}\n") }, false),
    }
}

/// **Serve one connection**: the handshake (a client certificate from the
/// installer CA REQUIRED), the one-role check, one request, one answer.
/// Returns whether the listener was closed by it.
pub fn serve_one(vm: &dyn Vm, log: &BootLog, cfg: Arc<rustls::ServerConfig>, stream: TcpStream) -> Result<bool, String> {
    let from = stream.peer_addr().map(|a| a.to_string()).unwrap_or_else(|_| "?".into());
    let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
    let _ = stream.set_write_timeout(Some(Duration::from_secs(20)));
    let conn = rustls::ServerConnection::new(cfg).map_err(|e| e.to_string())?;
    let mut tls = rustls::StreamOwned::new(conn, stream);
    let mut bytes = Vec::new();
    let mut chunk = [0u8; 4096];
    let req = loop {
        let n = tls.read(&mut chunk).map_err(|e| format!("{from}: {e}"))?;
        if n == 0 {
            return Err(format!("{from}: closed before a request"));
        }
        bytes.extend_from_slice(&chunk[..n]);
        if bytes.len() > MAX_FEED_BYTES {
            return Err(format!("{from}: request too large"));
        }
        if let Some(r) = parse_request(&bytes) {
            break r?;
        }
    };
    let cn = mtls::peer_cn(&tls.conn).unwrap_or_default();
    let (resp, closed) = if cn == mtls::INSTALLER_CN {
        respond(vm, log, &req)
    } else {
        (Response { status: 403, body: format!("{}\n", installer_only(&cn)) }, false)
    };
    tls.write_all(&resp.bytes()).map_err(|e| e.to_string())?;
    tls.conn.send_close_notify();
    let _ = tls.flush();
    if cn != mtls::INSTALLER_CN {
        return Err(format!("{from}: {}", installer_only(&cn)));
    }
    Ok(closed)
}

/// **The listener's loop**: one connection at a time until the installer
/// closes it. Returns once closed (the socket is dropped with it). `say`
/// hears every refused connection, in one line.
pub fn serve(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
    // The certificate the VM shows follows `Vm::identity`: the bootstrap one
    // until the installer signs the VM's own, then its own — so the ledger
    // hears "this is `main` again", not "the bootstrap certificate again".
    let mut current: Option<(String, Arc<rustls::ServerConfig>)> = None;
    for stream in listener.incoming() {
        let Ok(s) = stream else { continue };
        let id = vm.identity();
        let cfg = match &current {
            Some((pem, cfg)) if *pem == id.cert_pem => cfg.clone(),
            _ => match mtls::server_config(ca_pem, &id) {
                Ok(cfg) => {
                    current = Some((id.cert_pem.clone(), cfg.clone()));
                    cfg
                }
                Err(e) => {
                    say(&format!("the bootstrap listener cannot show its certificate: {e}"));
                    continue;
                }
            },
        };
        match serve_one(vm, log, cfg, s) {
            Ok(true) => {
                say("the installer closed the bootstrap listener: the bootstrap certificate is dead on this VM");
                return;
            }
            Ok(false) => {}
            Err(e) => say(&format!("a connection to the bootstrap listener was refused: {e}")),
        }
    }
}

/// The address the listener binds: every interface, [`PORT`] — the firewall
/// decides who reaches it (the operator's public address only).
pub fn bind(port: u16) -> std::io::Result<TcpListener> {
    TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::elect::Capacity;
    use crate::seed::{Kind, TunnelKind};
    use std::sync::Mutex;

    struct FakeVm {
        fed: Mutex<Vec<JoinReply>>,
        closed: Mutex<bool>,
    }

    impl Vm for FakeVm {
        fn hello(&self) -> Result<JoinRequest, String> {
            Ok(JoinRequest { estate: "e".into(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: Some(Capacity { cores: 2, ram_bytes: 1, data_bytes: 1, disk_mib_s: 1 }), csr_pem: String::new() })
        }
        fn fed(&self, reply: &JoinReply) -> Result<(), String> {
            self.fed.lock().unwrap().push(reply.clone());
            Ok(())
        }
        fn status(&self) -> Signal {
            Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
        }
        fn close(&self) -> Result<(), String> {
            *self.closed.lock().unwrap() = true;
            Ok(())
        }
        fn identity(&self) -> crate::mtls::Issued {
            crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
        }
        fn offer(&self) -> BTreeMap<String, String> {
            BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())])
        }
    }

    fn get(path: &str) -> Request {
        Request { method: "GET".into(), path: path.into(), body: Vec::new() }
    }

    #[test]
    fn the_listener_answers_hello_log_feed_status_and_close_and_nothing_else() {
        let vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
        let log = BootLog::new(10);
        log.push("korp-installer: pass began\ngunnar: listening");
        let (r, c) = respond(&vm, &log, &get(HELLO_PATH));
        assert_eq!((r.status, c), (200, false));
        assert_eq!(JoinRequest::parse(&r.body).unwrap().tunnel_pubkey, "K");
        let (r, _) = respond(&vm, &log, &get(&format!("{LOG_PATH}?from=1")));
        let chunk = crate::bootlog::Chunk::parse(&r.body).unwrap();
        assert_eq!((chunk.lines, chunk.next), (vec!["gunnar: listening".to_string()], 2));
        let reply = JoinReply { estate: "e".into(), me: crate::join::Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_A }, cert_pem: None, role: None, primary_overlay: None, primary_utility: None, peers: vec![], feed: Default::default() };
        let (r, _) = respond(&vm, &log, &Request { method: "POST".into(), path: FEED_PATH.into(), body: reply.render().unwrap().into_bytes() });
        assert_eq!(r.status, 200, "{}", r.body);
        assert_eq!(vm.fed.lock().unwrap().len(), 1);
        let (r, _) = respond(&vm, &log, &get(STATUS_PATH));
        assert!(Signal::parse(&r.body).is_ok(), "{}", r.body);
        let (r, _) = respond(&vm, &log, &get(OFFER_PATH));
        assert_eq!(parse_offer(&r.body).unwrap()["ticket_pubkey"], "ab12");
        assert_eq!(parse_offer("").unwrap(), BTreeMap::new(), "an empty body is an empty offer");
        let feed = BTreeMap::from([(fed_offer_key("main", "ticket_pubkey"), "ab12".to_string()), (fed_offer_key("twin", "twin_pubkey"), "ssh-ed25519 X".to_string()), ("monetize_pubkey".to_string(), "m".to_string())]);
        assert_eq!(offers_in(&feed, "main"), BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())]));
        assert_eq!(offers_in(&feed, "twin")["twin_pubkey"], "ssh-ed25519 X");
        assert_eq!(respond(&vm, &log, &get(CLOSE_PATH)).0.status, 405);
        assert_eq!(respond(&vm, &log, &get("/pair/join")).0.status, 404);
        let (r, closed) = respond(&vm, &log, &Request { method: "POST".into(), path: CLOSE_PATH.into(), body: Vec::new() });
        assert!(r.status == 200 && closed && *vm.closed.lock().unwrap());
    }

    /// 0.5.0: the secret offer has its own path, a VM that has none answers
    /// an empty one, and its `Debug` never shows a value.
    #[test]
    fn the_secret_offer_is_its_own_path_and_never_shows_a_value() {
        struct Front;
        impl Vm for Front {
            fn hello(&self) -> Result<JoinRequest, String> {
                Err("n/a".into())
            }
            fn fed(&self, _r: &JoinReply) -> Result<(), String> {
                Ok(())
            }
            fn status(&self) -> Signal {
                Signal::of("front", TunnelKind::Wireguard, &crate::pair::Machine::default())
            }
            fn close(&self) -> Result<(), String> {
                Ok(())
            }
            fn identity(&self) -> crate::mtls::Issued {
                crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
            }
            fn secret_offer(&self) -> SecretOffer {
                SecretOffer::new(BTreeMap::from([("tls.gunnar.rs.key".to_string(), "-----BEGIN PRIVATE KEY-----\nK\n".to_string())]))
            }
        }
        let log = BootLog::new(4);
        let (r, _) = respond(&Front, &log, &get(SECRET_OFFER_PATH));
        assert_eq!(r.status, 200);
        let s = SecretOffer::parse(&r.body).unwrap();
        assert_eq!(s.expose()["tls.gunnar.rs.key"], "-----BEGIN PRIVATE KEY-----\nK\n");
        assert_eq!(format!("{s:?}"), "SecretOffer([\"tls.gunnar.rs.key\"] — values withheld)");
        assert!(parse_offer(&respond(&Front, &log, &get(OFFER_PATH)).0.body).unwrap().is_empty(), "the public offer does not carry it");
        let box_vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
        assert!(SecretOffer::parse(&respond(&box_vm, &log, &get(SECRET_OFFER_PATH)).0.body).unwrap().is_empty(), "a box hands over nothing");
        assert!(!SecretOffer::parse("= nope").unwrap_err().contains("nope"), "the error never quotes the body");
    }

    #[test]
    fn the_from_query_is_read_and_a_missing_one_is_the_start() {
        assert_eq!(query_from("/install/log?from=42"), 42);
        assert_eq!(query_from("/install/log?x=1&from=7"), 7);
        assert_eq!(query_from("/install/log"), 0);
        assert!(installer_only("bootstrap").starts_with("REFUSED [installer-only]"));
    }
}