nordisk-pairing 0.5.26

//! **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 feed key a re-admitted PRIMARY is given** (0.5.8, [`Door::readmit`]):
/// its twin's control plane, `<utility>:<port>` — where a fresh main is filled
/// from. The same word holger's box configuration uses (`recharge_from`).
pub const RECHARGE_FROM_FEED: &str = "recharge_from";

/// What [`Door::readmit`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Readmitted {
    pub name: String,
    /// `(public, utility)` of the dead box, and of the new one.
    pub was: (Ipv4Addr, Ipv4Addr),
    pub now: (Ipv4Addr, Ipv4Addr),
    /// The dead box had joined (and its admission was cleared).
    pub forgotten: bool,
    /// The twin's control plane a re-admitted primary is fed.
    pub recharge_from: Option<String>,
}

/// **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>,
    /// **The boxes an IN-PLACE re-admission is re-admitting now** ([`Door::readmit_kept`],
    /// 0.5.26): each may show the certificate of ITS OWN slot while its admission is forgotten
    /// ([`Door::rejoin_in_place`]). A process parameter of one run, never in the ledger file.
    #[serde(skip)]
    pub in_place: std::collections::BTreeSet<String>,
}

/// `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))
    }

    /// **Re-admit ONE box of a closed estate** (0.5.8) — a box that DIED and
    /// was laid again from the same medium, at the address the provider gave
    /// the new machine. The dead main of gunnar and of holger is the reason:
    /// its twin holds everything, and a fresh main must join the estate it
    /// left, as the primary it was, told where its twin is.
    ///
    /// * The slot keeps its name, kind and overlay address; it moves to
    ///   `public` (the address the installer dials) with `utility` beside it.
    ///   Another box already laid at `public` is refused by name.
    /// * The old admission is forgotten ([`Door::forget`]): the ledger is not
    ///   complete, so the bootstrap certificate is accepted again — for this
    ///   slot only — and the new box gets a certificate of its own.
    /// * The VERDICT is kept: the new box takes the old one's role back.
    /// * What the dead box OFFERED is dropped from the feed (`offer.<name>.*`):
    ///   the new box mints its own and offers them anew.
    /// * A re-admitted PRIMARY with a twin is fed [`RECHARGE_FROM_FEED`] = the
    ///   twin's control plane (`<utility>:`[`crate::CONTROL_PORT`]) in its OWN
    ///   feed — where it is filled from; the product decides what that means.
    pub fn readmit(&mut self, name: &str, public: Ipv4Addr, utility: Ipv4Addr) -> Result<Readmitted, String> {
        let bad = |why: String| Err(format!("REFUSED [door-readmit] {why}"));
        let Some((&old_public, slot)) = self.ledger.slot_of(name) else {
            return bad(format!("`{name}` is not a box this estate laid"));
        };
        let mut slot = slot.clone();
        let old_utility = slot.utility_or(old_public);
        if slot.kind.is_product() && self.verdict.is_none() {
            return bad(format!("the estate has no verdict yet: `{name}` is re-admitted into a CLOSED estate, and this one never elected (run the install)"));
        }
        if public != old_public {
            if let Some(other) = self.ledger.slots.get(&public) {
                return bad(format!("{public} is where `{}` is laid; two boxes are never laid at one address", other.name));
            }
        }
        self.ledger.slots.remove(&old_public);
        slot.utility = Some(utility);
        self.ledger.slots.insert(public, slot);
        let forgotten = self.ledger.admitted.remove(name).is_some();
        let prefix = crate::bootstrap::fed_offer_key(name, "");
        self.feed.retain(|k, _| !k.starts_with(&prefix));
        let primary = self.verdict.as_ref().is_some_and(|v| v.primary == name);
        let recharge_from = if primary {
            self.verdict.as_ref().and_then(|v| self.ledger.slot_of(&v.twin)).map(|(k, s)| format!("{}:{}", s.utility_or(*k), crate::CONTROL_PORT))
        } else {
            None
        };
        if let Some(at) = &recharge_from {
            self.own_feed.entry(name.to_string()).or_default().insert(RECHARGE_FROM_FEED.to_string(), at.clone());
        }
        Ok(Readmitted { name: name.to_string(), was: (old_public, old_utility), now: (public, utility), forgotten, recharge_from })
    }

    /// **Re-admit ONE box that was UPGRADED IN PLACE** (0.5.9, `estate
    /// upgrade`): the same server, the same addresses, its data kept — its
    /// slot and role as [`Door::readmit`] keeps them, its admission forgotten
    /// (the upgrade's first boot is a new installation, which clears its
    /// pairing state), its old offers dropped; but a primary is NOT fed
    /// [`RECHARGE_FROM_FEED`] (and one left by an earlier recharge is taken
    /// back): it holds everything it held.
    pub fn readmit_kept(&mut self, name: &str) -> Result<Readmitted, String> {
        let Some((&public, slot)) = self.ledger.slot_of(name) else {
            return Err(format!("REFUSED [door-readmit] `{name}` is not a box this estate laid"));
        };
        let utility = slot.utility_or(public);
        let mut r = self.readmit(name, public, utility)?;
        if let Some(own) = self.own_feed.get_mut(name) {
            own.remove(RECHARGE_FROM_FEED);
        }
        r.recharge_from = None;
        self.in_place.insert(name.to_string());
        Ok(r)
    }

    /// **A box of an in-place re-admission that shows ITS OWN certificate** (0.5.26, gunnar.rs
    /// twin 2026-10-25): it was fed by an earlier run of this re-admission that ended before its
    /// close (that run's door, and the admission it made, lived in that run only), so it asks
    /// AS ITSELF, with no signing request. The installer dialled the slot's address over the
    /// installer link, the handshake proved the certificate is this CA's, for the slot's name,
    /// and the box is one [`Door::readmit_kept`] named: admitted again, with the tunnel key it
    /// shows NOW. Anything else stays refused as before (`join-forged`): a box not in this
    /// re-admission, another slot's certificate, a slot already admitted.
    pub fn rejoin_in_place(&mut self, req: &JoinRequest, cn: &str, from: Ipv4Addr) -> Result<JoinReply, JoinError> {
        if req.estate != self.ledger.estate {
            return Err(JoinError::WrongEstate { want: self.ledger.estate.clone(), got: req.estate.clone() });
        }
        let slot = self.ledger.slots.get(&from).ok_or(JoinError::Unknown(from))?.clone();
        if slot.name != cn || !self.in_place.contains(cn) || self.ledger.admitted.contains_key(cn) {
            return Err(JoinError::Forged(format!("{from} is `{}`'s address, and it presents the certificate of `{cn}`", slot.name)));
        }
        if req.kind != slot.kind && slot.kind != Kind::Front {
            return Err(JoinError::Forged(format!("{from} presents kind {:?}; the slot `{}` laid there is {:?}", req.kind, slot.name, slot.kind)));
        }
        self.ledger.admitted.insert(slot.name.clone(), crate::join::Admitted { from, tunnel_pubkey: req.tunnel_pubkey.clone(), capacity: req.capacity });
        self.elect_if_ready();
        Ok(self.reply_for(&slot.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 and durably (0.5.17, G4: edda appliance-pid1 `durable::write_atomic` — staged
    /// under its own name, fsync, rename, fsync of the directory): a reader never sees half a
    /// ledger, a power cut leaves the old one or the new one, and two writers never share a
    /// staging file.
    pub fn save(&self, path: &Path) -> Result<(), String> {
        let text = toml::to_string(self).map_err(|e| e.to_string())?;
        appliance_pid1::durable::write_atomic(path, text.as_bytes(), None).map_err(|e| format!("{}: {e}", 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;
        }
        // The product boxes, with their kind: an election is between exactly two
        // boxes of ONE product (a gunnar and a holger never pair).
        let products: Vec<(String, Ipv4Addr, crate::elect::Capacity, Kind)> = self
            .ledger
            .admitted
            .iter()
            .filter_map(|(name, a)| {
                let (_, slot) = self.ledger.slot_of(name)?;
                slot.kind.is_product().then_some((name.clone(), slot.overlay, a.capacity?, slot.kind))
            })
            .collect();
        if let [a, b] = &products[..] {
            if a.3 != b.3 {
                return;
            }
            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),
            // P3a: a product's own box kind is not elected; it stands alone as its own primary.
            (Kind::Box(_), _) => Some(Role::Primary),
            (k, Some(v)) if k.is_product() && v.twin == name => Some(Role::Twin),
            (k, Some(v)) if k.is_product() && 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.5.4: a HOLGER estate is gunnar's shape — two holger boxes are elected
    /// the same way (the weakest is the twin), and a gunnar box and a holger box
    /// never pair.
    #[test]
    fn two_holger_boxes_are_elected_and_a_gunnar_never_pairs_with_a_holger() {
        let ca = InstallerCa::mint("holger.rs", H).unwrap();
        let mut d = door(&ca);
        for s in [MAIN, TWIN] {
            d.ledger.slots.get_mut(&s).unwrap().kind = Kind::Holger;
        }
        d.join(&req(Kind::Gunnar, "KF", 2), &Presented::Bootstrap, FRONT).unwrap();
        d.join(&req(Kind::Holger, "KM", 4), &Presented::Bootstrap, MAIN).unwrap();
        let twin = d.join(&req(Kind::Holger, "KT", 1), &Presented::Bootstrap, TWIN).unwrap();
        assert_eq!(twin.role, Some(Role::Twin), "the weakest holger box is the twin");
        assert_eq!(d.role_of("main"), Some(Role::Primary));
        let mut fresh = door(&ca);
        fresh.ledger.slots.get_mut(&MAIN).unwrap().kind = Kind::Holger;
        assert!(matches!(fresh.join(&req(Kind::Gunnar, "X", 4), &Presented::Bootstrap, MAIN), Err(JoinError::Forged(_))), "a holger slot demands a holger box");

        let mut mixed = door(&ca);
        mixed.ledger.slots.get_mut(&TWIN).unwrap().kind = Kind::Holger;
        mixed.join(&req(Kind::Gunnar, "KM", 4), &Presented::Bootstrap, MAIN).unwrap();
        mixed.join(&req(Kind::Holger, "KT", 1), &Presented::Bootstrap, TWIN).unwrap();
        assert_eq!((mixed.role_of("main"), mixed.role_of("twin")), (None, None), "no election across products");
    }

    /// ★ P3a (A28 H13): a box of a product-named kind (`Kind::Box`) is admitted, never joins
    /// an election, and stands alone as its own primary; two boxes of one box kind are not a
    /// primary and a twin; a gunnar box in a box slot is forged; and a gunnar election beside
    /// it is unchanged.
    #[test]
    fn a_box_kind_stands_alone_as_its_own_primary_and_never_joins_an_election() {
        let ca = InstallerCa::mint("lake.rs", H).unwrap();
        let be = Kind::named("be").unwrap();
        let mut d = door(&ca);
        for s in [MAIN, TWIN] {
            d.ledger.slots.get_mut(&s).unwrap().kind = be;
        }
        d.join(&req(Kind::Gunnar, "KF", 2), &Presented::Bootstrap, FRONT).unwrap();
        let main = d.join(&req(be, "KM", 4), &Presented::Bootstrap, MAIN).unwrap();
        assert_eq!(main.role, Some(Role::Primary), "a box kind is fed its role at once: it waits for no twin");
        let twin = d.join(&req(be, "KT", 1), &Presented::Bootstrap, TWIN).unwrap();
        assert_eq!(twin.role, Some(Role::Primary), "two boxes of one box kind are two primaries, not an election");
        assert_eq!(d.verdict, None, "no election among box kinds");
        let mut fresh = door(&ca);
        fresh.ledger.slots.get_mut(&MAIN).unwrap().kind = be;
        assert!(matches!(fresh.join(&req(Kind::Gunnar, "X", 4), &Presented::Bootstrap, MAIN), Err(JoinError::Forged(_))), "a box slot demands its own kind");
        let mut other = door(&ca);
        other.ledger.slots.get_mut(&MAIN).unwrap().kind = be;
        assert!(matches!(other.join(&req(Kind::named("fe").unwrap(), "X", 4), &Presented::Bootstrap, MAIN), Err(JoinError::Forged(_))), "another box kind is forged too");
        assert!(d.readmit("main", Ipv4Addr::new(10, 13, 8, 99), Ipv4Addr::new(10, 13, 9, 99)).is_ok(), "a box kind is re-admitted with no verdict");
    }

    /// 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);
    }

    /// 0.5.8: the DEAD MAIN of a closed estate, laid again at new addresses,
    /// joins its old slot on the bootstrap certificate — the primary it was,
    /// fed where its twin is — and every other box's next answer names it at
    /// its new utility address. A re-admitted twin is fed no `recharge_from`.
    #[test]
    fn a_dead_main_is_readmitted_at_its_new_address_as_the_primary_and_told_where_its_twin_is() {
        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);
        d.feed.insert("offer.main.ticket_pubkey".into(), "dead-mains-key".into());
        d.feed.insert("offer.front.plugin_door".into(), "10.13.8.179:8475".into());
        let verdict = d.verdict.clone().expect("elected");
        assert!(d.readmit("stranger", MAIN, MAIN).unwrap_err().contains("[door-readmit]"));
        assert!(d.readmit("main", TWIN, TWIN).unwrap_err().contains("where `twin` is laid"), "two boxes at one address");
        let mut unelected = door(&ca);
        assert!(unelected.readmit("main", MAIN, MAIN).unwrap_err().contains("never elected"));

        let (new_public, new_util) = (Ipv4Addr::new(127, 80, 99, 40), Ipv4Addr::new(10, 13, 9, 140));
        let r = d.readmit("main", new_public, new_util).unwrap();
        assert_eq!(r, Readmitted { name: "main".into(), was: (MAIN, MAIN), now: (new_public, new_util), forgotten: true, recharge_from: Some(format!("{TWIN}:{}", crate::CONTROL_PORT)) });
        assert!(!d.ledger.slots.contains_key(&MAIN) && d.ledger.slots[&new_public].name == "main", "the slot moved");
        assert_eq!(d.verdict, Some(verdict.clone()), "the verdict is kept");
        assert!(!d.feed.contains_key("offer.main.ticket_pubkey"), "the dead box's offers are gone");
        assert!(d.feed.contains_key("offer.front.plugin_door"), "the others' are kept");
        assert!(!d.ledger.complete(), "the bootstrap certificate is open for that slot");
        assert!(post(&mut d, &req(Kind::Gunnar, "X", 1), &Presented::Bootstrap, TWIN).body.contains("join-spent"), "…and only that slot");

        let reply = d.join(&req(Kind::Gunnar, "KM2", 2), &Presented::Bootstrap, new_public).unwrap();
        assert!(reply.cert_pem.as_deref().unwrap_or("").contains("BEGIN CERTIFICATE"), "a certificate of its own");
        assert_eq!(reply.role, Some(Role::Primary), "the primary it was, even weaker than before");
        assert_eq!(reply.feed.get(RECHARGE_FROM_FEED).map(String::as_str), Some(format!("{TWIN}:{}", crate::CONTROL_PORT).as_str()));
        assert_eq!(reply.primary_utility, Some(new_util));
        assert!(d.ledger.complete());
        let twin = d.reply_for("twin");
        assert_eq!(twin.primary_utility, Some(new_util), "the twin's next answer names the new main");
        assert!(!twin.feed.contains_key(RECHARGE_FROM_FEED), "only the re-admitted primary is told");
        let p = twin.peers.iter().find(|p| p.name == "main").unwrap();
        assert_eq!((p.utility, p.tunnel_pubkey.as_str()), (Some(new_util), "KM2"), "the relay learns the new key");

        let t = d.readmit("twin", TWIN, TWIN).unwrap();
        assert_eq!(t.recharge_from, None, "a twin is not filled from anybody");
        assert!(!d.feed_for("twin").contains_key(RECHARGE_FROM_FEED));
    }

    /// **An UPGRADED box is re-admitted where it stands, its data kept**
    /// (`estate upgrade`, korp.reinstall=keep): the same slot at the same
    /// address, its role back, the bootstrap certificate open for it — and a
    /// primary is NOT told to be filled from its twin: it kept everything, and
    /// a holger box told `recharge_from` serves nothing until a lane refills a
    /// box that is not fresh (which the lane refuses). A `recharge_from` left
    /// from an earlier recharge is taken back.
    #[test]
    fn an_upgraded_box_is_readmitted_in_place_and_a_primary_is_never_told_to_recharge() {
        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);
        d.feed.insert("offer.main.ticket_pubkey".into(), "the-old-build".into());
        // An earlier recharge left its word in the main's own feed.
        d.own_feed.entry("main".into()).or_default().insert(RECHARGE_FROM_FEED.into(), "stale".into());
        let r = d.readmit_kept("main").unwrap();
        assert_eq!(r, Readmitted { name: "main".into(), was: (MAIN, MAIN), now: (MAIN, MAIN), forgotten: true, recharge_from: None });
        assert!(!d.feed_for("main").contains_key(RECHARGE_FROM_FEED), "a kept primary is filled from nobody");
        assert!(!d.feed.contains_key("offer.main.ticket_pubkey"), "the upgraded box offers anew");
        assert!(!d.ledger.complete(), "the bootstrap certificate is open for that slot");
        let reply = d.join(&req(Kind::Gunnar, "KM2", 4), &Presented::Bootstrap, MAIN).unwrap();
        assert_eq!(reply.role, Some(Role::Primary));
        assert!(!reply.feed.contains_key(RECHARGE_FROM_FEED));
        assert!(d.readmit_kept("stranger").unwrap_err().contains("[door-readmit]"));
    }

    /// ★ 0.5.26 (gunnar.rs twin, 2026-10-25): a box of an IN-PLACE re-admission that shows ITS
    /// OWN certificate — fed by an earlier run of that re-admission that ended before its close —
    /// is admitted again as itself, with the tunnel key it shows now. A box no re-admission
    /// names, another slot's certificate and a slot already admitted stay `join-forged`, and the
    /// door's own join still refuses the forgotten certificate.
    #[test]
    fn a_box_of_an_in_place_readmission_that_shows_its_own_certificate_is_admitted_as_itself() {
        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);
        let own = |key: &str| JoinRequest { csr_pem: String::new(), ..req(Kind::Gunnar, key, 1) };
        // Nobody named: forged, as before.
        d.forget("twin").unwrap();
        assert!(matches!(d.rejoin_in_place(&own("KT2"), "twin", TWIN), Err(JoinError::Forged(_))), "no re-admission names the twin");
        d.readmit_kept("twin").unwrap();
        assert!(matches!(d.rejoin_in_place(&own("KT2"), "main", TWIN), Err(JoinError::Forged(_))), "the main's certificate at the twin's address");
        assert!(matches!(d.join(&own("KT2"), &Presented::Box("twin".into()), TWIN), Err(JoinError::Forged(_))), "the door's own join still refuses it");
        let reply = d.rejoin_in_place(&own("KT2"), "twin", TWIN).unwrap();
        assert_eq!((reply.me.name.as_str(), reply.role), ("twin", Some(Role::Twin)), "its role back");
        assert!(reply.cert_pem.is_none(), "it keeps the certificate it has");
        assert_eq!(d.ledger.admitted["twin"].tunnel_pubkey, "KT2", "the tunnel key it shows now");
        assert!(d.ledger.complete());
        assert!(matches!(d.rejoin_in_place(&own("KT3"), "twin", TWIN), Err(JoinError::Forged(_))), "admitted already: nothing to re-key");
        assert_eq!(d.ledger.admitted["twin"].tunnel_pubkey, "KT2");
    }

    #[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\""));
    }

    /// ★ 0.5.17 (G4): the ledger is written durably, each writer staging its own file — two
    /// writers of one ledger (the installer's sink and a re-admission) never collide on one
    /// staging name, and the ledger is always whole.
    #[test]
    fn two_writers_of_one_ledger_never_collide_and_the_ledger_stays_whole() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let d = std::sync::Arc::new(door(&ca));
        let dir = std::env::temp_dir().join(format!("door-durable-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("ledger.toml");
        let failed: Vec<String> = std::thread::scope(|s| {
            let hs: Vec<_> = (0..8).map(|_| s.spawn(|| (0..40).filter_map(|_| d.save(&path).err()).collect::<Vec<_>>())).collect();
            hs.into_iter().flat_map(|h| h.join().unwrap()).collect()
        });
        assert!(failed.is_empty(), "{} saves failed, e.g. {:?}", failed.len(), failed.first());
        assert!(Door::load(&path).is_ok(), "the ledger is whole");
        let left: Vec<_> = std::fs::read_dir(&dir).unwrap().map(|e| e.unwrap().file_name()).collect();
        assert_eq!(left.len(), 1, "no staging file is left: {left:?}");
        let _ = std::fs::remove_dir_all(&dir);
    }
}