nordisk-pairing 0.5.2

//! **The join door on the front — the installer** — shared by gunnar and holger.
//!
//! It holds the [`Ledger`] of the boxes the estate laid (one slot per utility
//! address), speaks mutual TLS with a certificate from the installer CA, admits
//! each box ONCE on the bootstrap certificate the medium carries, signs the
//! box's own certificate from its request, elects the weakest gunnar box the
//! twin when both have joined — decided once, kept — and answers every box
//! with its name, overlay address, role, the peers and the estate values it
//! FEEDS (so the medium carries none of them). See [`crate::mtls`].
//!
//! The wire is minimal HTTP/1.1 inside TLS 1.3 over `std` sockets:
//! `POST /pair/join` with the [`JoinRequest`] as TOML, `GET /pair/signal` for
//! the door's own view. WHO is speaking is the certificate the handshake
//! verified; WHICH SLOT is the TCP peer address — the join goes over the
//! utility network because that address is what terraform knew before the
//! box existed.

use crate::elect::{weakest, Verdict};
use crate::join::{Admission, Assigned, JoinError, JoinReply, JoinRequest, Ledger, Peer, Presented};
use crate::mtls::{self, InstallerCa, Issued};
use crate::seed::Kind;
use crate::Role;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpStream};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

pub const JOIN_PATH: &str = "/pair/join";
pub const SIGNAL_PATH: &str = "/pair/signal";
pub const MAX_REQUEST_BYTES: usize = 64 * 1024;
/// How long a box's own installer certificate is valid **by default**: 24 h.
///
/// It was a fixed 6 h (audit 2026-09-24): a box re-joining after a late
/// reboot — or a slot re-joined after [`Door::forget`] — found its own
/// certificate dead. The caller sets it per estate with
/// [`Door::with_box_cert_valid`]. NOTE: a chain is only as valid as its
/// shortest link — the installer CA (minted per MEDIUM by the caller, e.g.
/// private-gunnar-ops `installer.rs`, and long-lived since 2026-09-24
/// evening) must outlive the box certificates it signs.
pub const BOX_CERT_VALID: Duration = Duration::from_secs(24 * 3600);

/// **The door's whole state**, one TOML file: the ledger, the verdict and the
/// estate values it feeds. The installer CA's key is NOT in it — it is its own
/// file on the front, handed to [`Door::with_ca`].
#[derive(Clone, Default, Serialize, Deserialize)]
pub struct Door {
    pub ledger: Ledger,
    #[serde(default)]
    pub verdict: Option<Verdict>,
    /// The estate values every admitted box is fed (`monetize_pubkey`, …).
    #[serde(default)]
    pub feed: BTreeMap<String, String>,
    /// **The values ONE box is fed, by its name** (0.5.0, owner 2026-09-25):
    /// what only that box may hold — the front's TLS key, the console's admin
    /// token, plugin modules. The reply to `name` carries [`Door::feed`] and
    /// then these (its own value wins on a clash); no other box's reply ever
    /// carries them. **Never serialized**: the ledger file ([`Door::save`],
    /// the run folder's `ledger.toml`) holds no per-box value — like the CA,
    /// it is a process parameter the installer's plan hands the door.
    #[serde(skip)]
    pub own_feed: BTreeMap<String, BTreeMap<String, String>>,
    /// **The front's tunnel, fed to every box** as the `front` peer of the
    /// reply ([`JoinReply::front`]) — the relay key and endpoint of THIS lay,
    /// which a constant medium cannot carry. `None` for a door with no relay
    /// (`tunnel = none`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub front: Option<crate::join::FrontTunnel>,
    #[serde(skip)]
    ca: Option<InstallerCa>,
    /// How long the box certificates this door signs are valid; `None` =
    /// [`BOX_CERT_VALID`]. A process parameter, not ledger state.
    #[serde(skip)]
    box_cert_valid: Option<Duration>,
}

/// `Debug` names each box's own feed by its KEYS only (0.5.0): those values
/// are the ones only one box may hold (a TLS private key, a bearer).
impl std::fmt::Debug for Door {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let own: BTreeMap<&String, Vec<&String>> = self.own_feed.iter().map(|(n, m)| (n, m.keys().collect())).collect();
        f.debug_struct("Door").field("ledger", &self.ledger).field("verdict", &self.verdict).field("feed", &self.feed).field("front", &self.front).field("own_feed (keys)", &own).finish_non_exhaustive()
    }
}

impl PartialEq for Door {
    fn eq(&self, o: &Door) -> bool {
        self.ledger == o.ledger && self.verdict == o.verdict && self.feed == o.feed && self.front == o.front
    }
}

impl Door {
    pub fn new(ledger: Ledger) -> Door {
        Door { ledger, ..Default::default() }
    }

    /// The CA that signs the boxes' own certificates.
    pub fn with_ca(mut self, ca: InstallerCa) -> Door {
        self.ca = Some(ca);
        self
    }

    /// How long the box certificates this door signs are valid (default
    /// [`BOX_CERT_VALID`], 24 h).
    pub fn with_box_cert_valid(mut self, valid_for: Duration) -> Door {
        self.box_cert_valid = Some(valid_for);
        self
    }

    pub fn box_cert_valid(&self) -> Duration {
        self.box_cert_valid.unwrap_or(BOX_CERT_VALID)
    }

    /// **Clear a slot's admission** — for a box that was RE-INSTALLED (a new
    /// disk, a new key, the old one gone). The slot stays laid; the box that
    /// comes back joins it again on the bootstrap certificate, gets a new
    /// certificate of its own and records its new tunnel key.
    ///
    /// * A name the estate never laid is refused by name ([`JoinError::NotLaid`]).
    /// * A slot that had not joined is `Ok(None)` — nothing to forget.
    /// * The VERDICT is kept: roles are decided once, and a re-installed box
    ///   takes its slot's role back when it rejoins.
    /// * Until the box rejoins, its OLD certificate is refused for the slot
    ///   (`join-forged`: a certificate for a slot that has not joined), and
    ///   the ledger is no longer complete, so the bootstrap certificate is
    ///   accepted again — for this slot only, and only while it is itself
    ///   still valid (the installer CA's lifetime is the caller's).
    pub fn forget(&mut self, name: &str) -> Result<Option<crate::join::Admitted>, JoinError> {
        if self.ledger.slot_of(name).is_none() {
            return Err(JoinError::NotLaid(name.to_string()));
        }
        Ok(self.ledger.admitted.remove(name))
    }

    pub fn load(path: &Path) -> Result<Door, String> {
        let text = std::fs::read_to_string(path).map_err(|e| format!("{}: {e}", path.display()))?;
        toml::from_str(&text).map_err(|e| format!("{}: not a door ledger: {e}", path.display()))
    }

    /// Whole, then renamed: a reader never sees half a ledger.
    pub fn save(&self, path: &Path) -> Result<(), String> {
        let text = toml::to_string(self).map_err(|e| e.to_string())?;
        let tmp = path.with_extension("toml.part");
        std::fs::write(&tmp, text).map_err(|e| format!("{}: {e}", tmp.display()))?;
        std::fs::rename(&tmp, path).map_err(|e| format!("{} → {}: {e}", tmp.display(), path.display()))
    }

    /// **A join from `from`, by `who`.** Admits under the ledger's rule, signs
    /// the box's own certificate on its first join, elects when both gunnar
    /// boxes have presented themselves, answers.
    pub fn join(&mut self, req: &JoinRequest, who: &Presented, from: Ipv4Addr) -> Result<JoinReply, JoinError> {
        // Sign BEFORE the slot is spent, so a request the CA cannot sign
        // leaves the slot open for the box's next try.
        let signed = if *who == Presented::Bootstrap && !self.ledger.admitted.values().any(|a| a.from == from) {
            let name = self.ledger.slots.get(&from).map(|s| s.name.clone()).ok_or(JoinError::Unknown(from))?;
            let ca = self.ca.as_ref().ok_or_else(|| JoinError::Sign("the door holds no installer CA".into()))?;
            Some(ca.sign_box(&req.csr_pem, &name, self.box_cert_valid()).map_err(|e| JoinError::Sign(e.to_string()))?)
        } else {
            None
        };
        let admission = self.ledger.admit(req, who, from)?;
        self.elect_if_ready();
        let mut reply = self.reply_for(admission.name());
        if let Admission::First(_) = admission {
            reply.cert_pem = signed;
        }
        Ok(reply)
    }

    fn elect_if_ready(&mut self) {
        if self.verdict.is_some() {
            return;
        }
        let gunnars: Vec<(String, Ipv4Addr, crate::elect::Capacity)> = self
            .ledger
            .admitted
            .iter()
            .filter_map(|(name, a)| {
                let (_, slot) = self.ledger.slot_of(name)?;
                (slot.kind == Kind::Gunnar).then_some((name.clone(), slot.overlay, a.capacity?))
            })
            .collect();
        if let [a, b] = &gunnars[..] {
            self.verdict = Some(weakest((&a.0, a.1, &a.2), (&b.0, b.1, &b.2)));
        }
    }

    pub fn role_of(&self, name: &str) -> Option<Role> {
        let (_, slot) = self.ledger.slot_of(name)?;
        if !self.ledger.admitted.contains_key(name) {
            return None;
        }
        match (slot.kind, &self.verdict) {
            (Kind::Front, _) => Some(Role::Front),
            (Kind::Gunnar, Some(v)) if v.twin == name => Some(Role::Twin),
            (Kind::Gunnar, Some(v)) if v.primary == name => Some(Role::Primary),
            _ => None,
        }
    }

    /// **The front's tunnel, as every box is fed it** (0.4.0): the one the
    /// door was given ([`Door::front`]) or else DERIVED from the admitted
    /// front VM — vm1 boots the same medium, presents its own tunnel key like
    /// every box, and that key is the relay's; it is dialled at the address
    /// the boxes reach it on (its slot's `endpoint`, or its UTILITY address),
    /// port [`crate::RELAY_PORT`]. `None` until the front has presented a key.
    pub fn front_tunnel(&self) -> Option<crate::join::FrontTunnel> {
        if let Some(f) = &self.front {
            return Some(f.clone());
        }
        self.ledger.slots.iter().filter(|(_, s)| s.kind == Kind::Front).find_map(|(addr, s)| {
            let a = self.ledger.admitted.get(&s.name)?;
            if a.tunnel_pubkey.trim().is_empty() {
                return None;
            }
            // 0.4.2: on the UTILITY address — the path that never leaves the
            // provider, and the one a mock guest can route (the public one is
            // what the installer dials, not what the boxes reach).
            let endpoint = s.endpoint.clone().unwrap_or_else(|| format!("{}:{}", s.utility_or(*addr), crate::RELAY_PORT));
            Some(crate::join::FrontTunnel { tunnel_pubkey: a.tunnel_pubkey.clone(), endpoint })
        })
    }

    /// The front (its tunnel, first), then every admitted gunnar box with a
    /// known role, as peers. The front VM itself is listed once, as the
    /// front peer.
    pub fn peers(&self) -> Vec<Peer> {
        let front = self.front_tunnel().map(|f| f.as_peer());
        front
            .into_iter()
            .chain(self.ledger.admitted.iter().filter_map(|(name, a)| {
                let (key, slot) = self.ledger.slot_of(name)?;
                let role = self.role_of(name)?;
                (role != Role::Front).then(|| Peer { name: name.clone(), role, tunnel_pubkey: a.tunnel_pubkey.clone(), overlay: slot.overlay, endpoint: slot.endpoint.clone(), utility: Some(slot.utility_or(*key)) })
            }))
            .collect()
    }

    pub fn reply_for(&self, name: &str) -> JoinReply {
        let (_, slot) = self.ledger.slot_of(name).expect("reply_for is called for a laid name");
        let primary_slot = self.verdict.as_ref().and_then(|v| self.ledger.slot_of(&v.primary));
        let primary_overlay = primary_slot.map(|(_, s)| s.overlay);
        let primary_utility = primary_slot.map(|(u, s)| s.utility_or(*u));
        JoinReply {
            estate: self.ledger.estate.clone(),
            me: Assigned { name: name.to_string(), overlay: slot.overlay },
            cert_pem: None,
            role: self.role_of(name),
            primary_overlay,
            primary_utility,
            peers: self.peers(),
            feed: self.feed_for(name),
        }
    }

    /// **What `name` is fed**: the shared [`Door::feed`], then its own
    /// ([`Door::own_feed`]) over it. Nothing another box was given alone.
    pub fn feed_for(&self, name: &str) -> BTreeMap<String, String> {
        let mut f = self.feed.clone();
        if let Some(own) = self.own_feed.get(name) {
            f.extend(own.iter().map(|(k, v)| (k.clone(), v.clone())));
        }
        f
    }

    /// The door's own signal: who joined, the verdict.
    pub fn signal(&self) -> String {
        let q = |it: Vec<&String>| it.iter().map(|s| format!("\"{s}\"")).collect::<Vec<_>>().join(", ");
        let mut out = format!(
            "estate = \"{}\"\njoined = [{}]\nexpected = [{}]\n",
            self.ledger.estate,
            q(self.ledger.admitted.keys().collect()),
            q(self.ledger.slots.values().map(|s| &s.name).collect())
        );
        match &self.verdict {
            Some(v) => out.push_str(&format!("twin = \"{}\"\nprimary = \"{}\"\nwhy = \"{}\"\n", v.twin, v.primary, v.why.replace('"', "'"))),
            None => out.push_str("verdict = \"pending\"\n"),
        }
        out.push_str(&format!("complete = {}\nbootstrap = \"{}\"\n", self.ledger.complete(), if self.ledger.complete() { "withdrawn" } else { "open" }));
        out
    }
}

// ── the wire ───────────────────────────────────────────────────────────────

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Request {
    pub method: String,
    pub path: String,
    pub body: Vec<u8>,
}

/// One request from the bytes read so far. `None` = incomplete (read more);
/// `Some(Err)` = malformed.
pub fn parse_request(bytes: &[u8]) -> Option<Result<Request, String>> {
    let head_end = bytes.windows(4).position(|w| w == b"\r\n\r\n")?;
    let head = String::from_utf8_lossy(&bytes[..head_end]).to_string();
    let mut lines = head.split("\r\n");
    let start = lines.next().unwrap_or_default();
    let mut parts = start.split_whitespace();
    let (Some(method), Some(path)) = (parts.next(), parts.next()) else {
        return Some(Err(format!("not an HTTP request line: `{start}`")));
    };
    let len: usize = lines
        .filter_map(|l| l.split_once(':'))
        .find(|(k, _)| k.trim().eq_ignore_ascii_case("content-length"))
        .and_then(|(_, v)| v.trim().parse().ok())
        .unwrap_or(0);
    let body_start = head_end + 4;
    if bytes.len() < body_start + len {
        return None;
    }
    Some(Ok(Request { method: method.to_string(), path: path.to_string(), body: bytes[body_start..body_start + len].to_vec() }))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Response {
    pub status: u16,
    pub body: String,
}

impl Response {
    pub fn bytes(&self) -> Vec<u8> {
        let reason = match self.status {
            200 => "OK",
            400 => "Bad Request",
            403 => "Forbidden",
            404 => "Not Found",
            405 => "Method Not Allowed",
            _ => "Error",
        };
        format!(
            "HTTP/1.1 {} {reason}\r\ncontent-type: application/toml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
            self.status,
            self.body.len(),
            self.body
        )
        .into_bytes()
    }

    pub fn parse(raw: &[u8]) -> Result<Response, String> {
        let text = String::from_utf8_lossy(raw);
        let (head, body) = text.split_once("\r\n\r\n").ok_or("no end of headers in the reply")?;
        let status = head.split_whitespace().nth(1).and_then(|s| s.parse().ok()).ok_or("no status in the reply")?;
        Ok(Response { status, body: body.to_string() })
    }
}

/// **One request, answered.** Pure over the door, the speaker and the address.
pub fn respond(door: &mut Door, req: &Request, who: &Presented, from: Ipv4Addr) -> Response {
    match (req.method.as_str(), req.path.as_str()) {
        ("POST", JOIN_PATH) => {
            let jr = match JoinRequest::parse(&String::from_utf8_lossy(&req.body)) {
                Ok(j) => j,
                Err(e) => return Response { status: 400, body: format!("{e}\n") },
            };
            match door.join(&jr, who, from).and_then(|r| r.render()) {
                Ok(text) => Response { status: 200, body: text },
                Err(e) => Response { status: 403, body: format!("{e}\n") },
            }
        }
        ("GET", SIGNAL_PATH) => Response { status: 200, body: door.signal() },
        (_, JOIN_PATH) | (_, SIGNAL_PATH) => Response { status: 405, body: "REFUSED [door-method] POST /pair/join or GET /pair/signal\n".into() },
        _ => Response { status: 404, body: "REFUSED [door-path] this door has /pair/join and /pair/signal\n".into() },
    }
}

/// **Serve one connection** (the door's side): the TLS handshake with a
/// client certificate REQUIRED, one request, one answer. Returns whether the
/// door changed, so the caller saves it. A handshake that fails — a stranger's
/// certificate, none at all — is refused by rustls before a byte is read.
pub fn serve_one(door: &mut Door, cfg: Arc<rustls::ServerConfig>, stream: TcpStream) -> Result<bool, String> {
    let from = match stream.peer_addr() {
        Ok(SocketAddr::V4(a)) => *a.ip(),
        other => return Err(format!("not an IPv4 peer: {other:?}")),
    };
    let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
    let _ = stream.set_write_timeout(Some(Duration::from_secs(10)));
    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_REQUEST_BYTES {
            return Err(format!("{from}: request too large"));
        }
        if let Some(r) = parse_request(&bytes) {
            break r?;
        }
    };
    let who = mtls::peer_cn(&tls.conn).map(|cn| Presented::from_cn(&cn)).ok_or_else(|| format!("{from}: no client certificate"))?;
    let before = door.clone();
    let r = respond(door, &req, &who, from);
    tls.write_all(&r.bytes()).map_err(|e| e.to_string())?;
    tls.conn.send_close_notify();
    let _ = tls.flush();
    Ok(*door != before)
}

/// **The box's side**: POST the join over mutual TLS — the installer CA pinned,
/// `identity` the certificate the box speaks with.
pub fn post_join(door: SocketAddrV4, ca_pem: &str, identity: &Issued, req: &JoinRequest, timeout: Duration) -> Result<JoinReply, String> {
    let cfg = mtls::client_config(ca_pem, identity).map_err(|e| e.to_string())?;
    let body = req.render().map_err(|e| e.to_string())?;
    let s = TcpStream::connect_timeout(&SocketAddr::V4(door), timeout).map_err(|e| format!("could not be reached: {door}: {e}"))?;
    let _ = s.set_read_timeout(Some(timeout));
    let _ = s.set_write_timeout(Some(timeout));
    let conn = rustls::ClientConnection::new(cfg, mtls::door_name(*door.ip())).map_err(|e| e.to_string())?;
    let mut tls = rustls::StreamOwned::new(conn, s);
    let raw = format!("POST {JOIN_PATH} HTTP/1.1\r\nhost: {door}\r\ncontent-type: application/toml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len());
    tls.write_all(raw.as_bytes()).map_err(|e| format!("could not be reached: TLS to {door}: {e}"))?;
    let mut buf = Vec::new();
    // A peer that closes without close_notify still said everything it had.
    match tls.read_to_end(&mut buf) {
        Ok(_) => {}
        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && !buf.is_empty() => {}
        Err(e) => return Err(format!("could not be reached: reading from {door}: {e}")),
    }
    let r = Response::parse(&buf)?;
    if r.status != 200 {
        return Err(r.body.trim().to_string());
    }
    JoinReply::parse(&r.body).map_err(|e| e.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::elect::Capacity;
    use crate::join::Slot;

    const H: Duration = Duration::from_secs(3600);
    const FRONT: Ipv4Addr = Ipv4Addr::new(10, 13, 8, 179);
    const MAIN: Ipv4Addr = Ipv4Addr::new(10, 13, 9, 90);
    const TWIN: Ipv4Addr = Ipv4Addr::new(10, 13, 9, 87);

    fn door(ca: &InstallerCa) -> Door {
        let mut l = Ledger::new("gunnar.rs");
        l.lay(FRONT, Slot { name: "front".into(), kind: Kind::Front, overlay: crate::OVERLAY_FRONT, endpoint: Some("95.111.211.189:51820".into()), utility: None });
        l.lay(MAIN, Slot { name: "main".into(), kind: Kind::Gunnar, overlay: crate::OVERLAY_BOX_A, endpoint: None, utility: None });
        l.lay(TWIN, Slot { name: "twin".into(), kind: Kind::Gunnar, overlay: crate::OVERLAY_BOX_B, endpoint: None, utility: None });
        let mut d = Door::new(l).with_ca(ca.clone());
        d.feed.insert("monetize_pubkey".into(), "eb16f371b856".into());
        d.front = Some(crate::join::FrontTunnel { tunnel_pubkey: "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=".into(), endpoint: "95.111.211.189:51820".into() });
        d
    }

    fn req(kind: Kind, key: &str, cores: u32) -> JoinRequest {
        let (csr, _) = mtls::box_request().unwrap();
        JoinRequest { estate: "gunnar.rs".into(), kind, tunnel_pubkey: key.into(), capacity: Some(Capacity { cores, ram_bytes: 4 << 30, data_bytes: 10 << 30, disk_mib_s: 500 }), csr_pem: csr }
    }

    fn post(d: &mut Door, r: &JoinRequest, who: &Presented, from: Ipv4Addr) -> Response {
        let body = r.render().unwrap();
        let raw = format!("POST {JOIN_PATH} HTTP/1.1\r\ncontent-length: {}\r\n\r\n{body}", body.len());
        respond(d, &parse_request(raw.as_bytes()).unwrap().unwrap(), who, from)
    }

    /// 0.4.0, the one-medium ruling: vm1 boots the SAME gunnar medium, the
    /// front slot makes it the front, and its own tunnel key is the relay's —
    /// fed to the boxes as the `front` peer, dialled on its laid address.
    #[test]
    fn vm1_on_the_gunnar_medium_becomes_the_front_and_its_key_is_the_relay_fed_to_the_boxes() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let mut d = door(&ca);
        d.front = None;
        d.ledger.slots.get_mut(&FRONT).unwrap().endpoint = None;
        assert_eq!(d.front_tunnel(), None, "no front key before vm1 presented one");
        let r = d.join(&req(Kind::Gunnar, "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=", 2), &Presented::Bootstrap, FRONT).unwrap();
        assert_eq!(r.role, Some(Role::Front), "the slot, not the medium, makes the front");
        d.join(&req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN).unwrap();
        let twin = d.join(&req(Kind::Gunnar, "KT", 1), &Presented::Bootstrap, TWIN).unwrap();
        let front = twin.front().expect("the front is fed");
        assert_eq!(front.tunnel_pubkey, "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=");
        assert_eq!(front.endpoint.as_deref(), Some("10.13.8.179:51820"));
        assert_eq!(twin.peers.iter().filter(|p| p.role == Role::Front).count(), 1, "vm1 listed once, as the front: {:?}", twin.peers);
        assert_eq!(twin.peers.len(), 3);
        // 0.4.1: the front certifies the twin on the utility network, so its
        // reply names every box's utility address (here the ledger key).
        let fr = d.reply_for("front");
        let at = |role: Role| fr.peers.iter().find(|p| p.role == role).and_then(|p| p.utility);
        assert_eq!((at(Role::Twin), at(Role::Primary), at(Role::Front)), (Some(TWIN), Some(MAIN), None), "{:?}", fr.peers);
        // A box slot still demands its own kind.
        let mut e = door(&ca);
        assert!(matches!(e.join(&req(Kind::Front, "K", 1), &Presented::Bootstrap, MAIN), Err(JoinError::Forged(_))));
    }

    #[test]
    fn joins_by_address_elect_the_weakest_sign_each_box_feed_the_estate_and_withdraw_bootstrap_at_completion() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let mut d = door(&ca);
        let r = post(&mut d, &req(Kind::Front, "KF", 2), &Presented::Bootstrap, FRONT);
        assert_eq!(JoinReply::parse(&r.body).unwrap().role, Some(Role::Front), "{}", r.body);
        let r = post(&mut d, &req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN);
        let reply = JoinReply::parse(&r.body).unwrap();
        assert_eq!((reply.me.name.as_str(), reply.role), ("main", None), "admitted, election pending");
        assert!(reply.cert_pem.as_deref().unwrap_or("").contains("BEGIN CERTIFICATE"), "the box's own certificate on its first join");
        assert_eq!(reply.feed.get("monetize_pubkey").map(String::as_str), Some("eb16f371b856"), "the estate value is fed, not baked");
        let front = reply.front().expect("the front's tunnel is fed on every reply, before any election");
        assert_eq!((front.tunnel_pubkey.as_str(), front.endpoint.as_deref()), ("xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=", Some("95.111.211.189:51820")), "the lay's relay, fed — not on the medium");
        // main asks again, with ITS OWN certificate: a read, no new certificate
        let again = JoinReply::parse(&post(&mut d, &req(Kind::Gunnar, "KM", 4), &Presented::Box("main".into()), MAIN).body).unwrap();
        assert!(again.cert_pem.is_none());
        let r = post(&mut d, &req(Kind::Gunnar, "KT", 1), &Presented::Bootstrap, TWIN);
        let reply = JoinReply::parse(&r.body).unwrap();
        assert_eq!((reply.me.name.as_str(), reply.role, reply.primary_overlay), ("twin", Some(Role::Twin), Some(crate::OVERLAY_BOX_A)));
        assert!(d.ledger.complete());
        assert!(d.signal().contains("bootstrap = \"withdrawn\""));
        let r = post(&mut d, &req(Kind::Gunnar, "X", 4), &Presented::Bootstrap, MAIN);
        assert!(r.status == 403 && r.body.contains("REFUSED [join-withdrawn]"), "{}", r.body);
        let p = std::env::temp_dir().join(format!("door-{}.toml", std::process::id()));
        d.save(&p).unwrap();
        let back = Door::load(&p).unwrap();
        assert_eq!(back, d, "the ledger round-trips; the CA key is not in it");
        assert!(!std::fs::read_to_string(&p).unwrap().contains("PRIVATE KEY"), "the CA key never lands in the ledger file");
        let _ = std::fs::remove_file(&p);
    }

    /// The whole wire over loopback: TLS 1.3, the bootstrap certificate, the
    /// box's own certificate on the second ask, a stranger refused.
    #[test]
    fn a_box_joins_over_real_mutual_tls_then_asks_again_as_itself() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let door_cert = ca.issue_door(Ipv4Addr::LOCALHOST, H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let mut l = Ledger::new("gunnar.rs");
        l.lay(Ipv4Addr::LOCALHOST, Slot { name: "main".into(), kind: Kind::Gunnar, overlay: crate::OVERLAY_BOX_A, endpoint: None, utility: None });
        let d = std::sync::Arc::new(std::sync::Mutex::new(Door::new(l).with_ca(ca.clone())));
        let scfg = mtls::server_config(&ca.cert_pem, &door_cert).unwrap();
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let SocketAddr::V4(addr) = listener.local_addr().unwrap() else { panic!() };
        let dd = d.clone();
        let server = std::thread::spawn(move || {
            for _ in 0..3 {
                let (s, _) = listener.accept().unwrap();
                let _ = serve_one(&mut dd.lock().unwrap(), scfg.clone(), s);
            }
        });
        let (csr, box_key) = mtls::box_request().unwrap();
        let r = JoinRequest { estate: "gunnar.rs".into(), kind: Kind::Gunnar, tunnel_pubkey: "KM".into(), capacity: None, csr_pem: csr };
        let first = post_join(addr, &ca.cert_pem, &boot, &r, Duration::from_secs(5)).unwrap();
        let mine = Issued { cert_pem: first.cert_pem.clone().expect("a certificate on the first join"), key_pem: box_key };
        let second = post_join(addr, &ca.cert_pem, &mine, &JoinRequest { csr_pem: String::new(), ..r.clone() }, Duration::from_secs(5)).unwrap();
        assert_eq!((second.me.name.as_str(), second.cert_pem.is_none()), ("main", true));
        let stranger = InstallerCa::mint("stranger", H).unwrap().issue_bootstrap(H).unwrap();
        assert!(post_join(addr, &ca.cert_pem, &stranger, &r, Duration::from_secs(5)).is_err(), "another CA's certificate never reaches the door's logic");
        server.join().unwrap();
        assert_eq!(d.lock().unwrap().ledger.admitted.len(), 1);
    }

    fn not_after(pem: &str) -> i64 {
        let der = x509_parser::pem::parse_x509_pem(pem.as_bytes()).unwrap().1;
        der.parse_x509().unwrap().validity().not_after.timestamp()
    }

    #[test]
    fn the_box_certificate_lifetime_is_the_callers_with_a_24h_default() {
        let ca = InstallerCa::mint("gunnar.rs", 48 * H).unwrap();
        let now = time::OffsetDateTime::now_utc().unix_timestamp();
        let mut d = door(&ca);
        assert_eq!(d.box_cert_valid(), Duration::from_secs(24 * 3600));
        let r = JoinReply::parse(&post(&mut d, &req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN).body).unwrap();
        let left = not_after(&r.cert_pem.unwrap()) - now;
        assert!((24 * 3600 - 60..=24 * 3600 + 60).contains(&left), "default 24 h, got {left} s");
        let mut d = door(&ca).with_box_cert_valid(36 * H);
        let r = JoinReply::parse(&post(&mut d, &req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN).body).unwrap();
        let left = not_after(&r.cert_pem.unwrap()) - now;
        assert!((36 * 3600 - 60..=36 * 3600 + 60).contains(&left), "the caller's 36 h, got {left} s");
        let p = std::env::temp_dir().join(format!("door-valid-{}.toml", std::process::id()));
        d.save(&p).unwrap();
        assert!(!std::fs::read_to_string(&p).unwrap().contains("box_cert_valid"), "a process parameter, not ledger state");
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn forget_clears_a_reinstalled_boxs_admission_keeps_the_verdict_and_lets_it_rejoin_once() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let mut d = door(&ca);
        post(&mut d, &req(Kind::Front, "KF", 2), &Presented::Bootstrap, FRONT);
        post(&mut d, &req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN);
        post(&mut d, &req(Kind::Gunnar, "KT", 1), &Presented::Bootstrap, TWIN);
        assert!(d.ledger.complete());
        let verdict = d.verdict.clone().expect("elected");
        // a name never laid: refused by name
        let e = d.forget("stranger").unwrap_err();
        assert_eq!(e, JoinError::NotLaid("stranger".into()));
        assert!(e.to_string().starts_with("REFUSED [door-forget-unknown]"), "{e}");
        // the twin is re-installed
        let old = d.forget("twin").unwrap().expect("twin had joined");
        assert_eq!(old.tunnel_pubkey, "KT");
        assert_eq!(d.forget("twin"), Ok(None), "forgetting twice is nothing");
        assert!(!d.ledger.complete() && d.signal().contains("bootstrap = \"open\""));
        assert_eq!(d.role_of("twin"), None, "not admitted, no role");
        // its OLD certificate is refused until it rejoins
        let r = post(&mut d, &req(Kind::Gunnar, "KT", 1), &Presented::Box("twin".into()), TWIN);
        assert!(r.status == 403 && r.body.contains("join-forged"), "{}", r.body);
        // bootstrap is open for THAT slot only
        let r = post(&mut d, &req(Kind::Gunnar, "X", 4), &Presented::Bootstrap, MAIN);
        assert!(r.status == 403 && r.body.contains("join-spent"), "{}", r.body);
        let r = post(&mut d, &req(Kind::Gunnar, "KT2", 8), &Presented::Bootstrap, TWIN);
        let reply = JoinReply::parse(&r.body).unwrap();
        assert!(reply.cert_pem.as_deref().unwrap_or("").contains("BEGIN CERTIFICATE"), "a new certificate for the new box");
        assert_eq!(reply.role, Some(Role::Twin), "the verdict is kept even though the new box is stronger");
        assert_eq!(d.verdict, Some(verdict));
        assert_eq!(d.ledger.admitted["twin"].tunnel_pubkey, "KT2", "the new key is recorded");
        assert!(d.ledger.complete() && d.signal().contains("bootstrap = \"withdrawn\""));
    }
}