gunnar-grpc 1.5.0

//! `Bootstrap`: the one service gunnar serves without a credential, and only
//! while nothing has ever been enrolled.
//!
//! # The whole security boundary, in one sentence
//!
//! `Enroll` is refused unless
//! `gunnar_core::AuthTable::bootstrap_window_open` is true, and the check that
//! decides that is inside the same compare-and-swap that writes the first
//! identity — so there is no interval between deciding and acting for a second
//! caller to slip through.
//!
//! # Why this service is NOT behind [`crate::auth::AuthGate`]
//!
//! It cannot be. The gate's whole job is to look a fingerprint up in the ACL key
//! table, and on a server that has never been enrolled the key table is empty:
//! every credential, including a correct one, fails. Putting `Bootstrap` behind
//! the gate would make the first boot unreachable, which is the defect this
//! service exists to fix.
//!
//! So it is mounted **beside** the gate, and its own refusal is the door. Two
//! properties keep that honest:
//!
//! * [`crate::auth::UNAUTHENTICATED_METHODS`] names exactly these two RPCs, and
//!   `tests/auth.rs` asserts that every other method in
//!   [`crate::auth::METHODS`] is refused without a credential. A third
//!   unauthenticated method cannot be added by accident; it has to be written
//!   into that list, where it is visible.
//! * the refusal lives in `gunnar-core`, under the CAS, not here. A handler
//!   that forgot to check would still be refused by the writer.
//! * `Restore` is outside the gate but NOT credential-free (h-v1-6a,
//!   2026-09-28): [`crate::auth::authenticate_restore`] checks the stream's
//!   `gunnar-sshsig-v2` header against the keys the box was fed, and the
//!   verified key is the owner the resume is held to.
//! * `BecomeTwin` and `InstallTwinCertificate` likewise (h-bootstrap-open,
//!   2026-09-28): [`crate::auth::authenticate_bootstrap`], a credential bound to
//!   the method from a FED key — a twin follows whatever primary it is told and
//!   pins the first issuer it is handed, so neither is anybody's to call.
//!
//! # What it says out loud
//!
//! Every call, allowed or refused, is one record on
//! [`crate::auth::AUDIT_TARGET`]. An enrolment is `warn!`, not `info!`: it
//! happens once in a server's life and it hands somebody administrative control
//! of the box. A refused enrolment is `warn!` too — on a server that has been
//! enrolled, somebody calling `Bootstrap.Enroll` is either a confused operator
//! or an attacker trying the obvious thing, and both are worth a line.

use gunnar_proto::bootstrap_server::Bootstrap;
use gunnar_proto::{
    AssignTunnelAddressRequest, AssignTunnelAddressResponse,
    BecomeTwinRequest, BecomeTwinResponse, BootstrapStatusRequest, BootstrapStatusResponse,
    EnrollRequest, EnrollResponse, GrowStatusRequest, GrowStatusResponse, InstallTwinCertificateRequest,
    InstallTwinCertificateResponse, LogChunk, RestoreResponse, TwinCsrRequest, TwinCsrResponse,
};
use tonic::{Request, Response, Status};

use crate::seam::{BootstrapState, ControlPlane};
use crate::service::BootstrapService;

/// The audited method name for enrolment.
pub const METHOD_ENROLL: &str = "Bootstrap.Enroll";
/// The audited method name for the status probe.
pub const METHOD_STATUS: &str = "Bootstrap.Status";
/// The audited method name for the runtime growth's credential-free status.
pub const METHOD_GROW_STATUS: &str = "Bootstrap.GrowStatus";
/// The audited method name for the twin door.
pub const METHOD_BECOME_TWIN: &str = "Bootstrap.BecomeTwin";
/// The audited method name for the restore door.
pub const METHOD_RESTORE: &str = "Bootstrap.Restore";
/// How long a `Bootstrap.Restore` stream waits for the box's intake while another stream still
/// holds it (an abandoned stream whose last apply has not returned).
pub const RESTORE_INTAKE_WAIT: std::time::Duration = std::time::Duration::from_secs(20);
/// How long a `Bootstrap.Restore` stream may go without a chunk before the box ends it (and
/// lets go of its intake): a sender whose connection died without a word.
pub const RESTORE_IDLE: std::time::Duration = std::time::Duration::from_secs(180);
/// The audited method name for the tunnel-address door.
pub const METHOD_ASSIGN_TUNNEL_ADDRESS: &str = "Bootstrap.AssignTunnelAddress";
/// The audited method name for the twin's certificate request (lane T9).
pub const METHOD_TWIN_CSR: &str = "Bootstrap.TwinCsr";
/// The audited method name for installing the twin's certificate (lane T9).
pub const METHOD_INSTALL_TWIN_CERTIFICATE: &str = "Bootstrap.InstallTwinCertificate";

/// One sentence naming the state, written once so the log line, the status
/// response and the start-up banner cannot describe the same server three
/// different ways (one implementation).
pub fn describe(state: &BootstrapState) -> String {
    if state.window_open {
        return "NO IDENTITY HAS EVER BEEN ENROLLED. Bootstrap.Enroll is open to any caller that \
                can reach this control port, with no credential. It closes permanently on the \
                first success."
            .to_owned();
    }
    if state.bootstrapped && state.keys == 0 {
        return format!(
            "bootstrapped at {} by \"{}\", but the identity table now holds {} principals and 0 \
             keys: nobody can authenticate against this server, and the enrolment window stays \
             shut. Add a key with `gunnar key add` while the server is down.",
            state.bootstrapped_at_unix_ms, state.bootstrapped_principal, state.principals
        );
    }
    if state.bootstrapped {
        return format!(
            "bootstrapped at {} by \"{}\"; the enrolment window is closed permanently. {} \
             principals, {} keys.",
            state.bootstrapped_at_unix_ms,
            state.bootstrapped_principal,
            state.principals,
            state.keys
        );
    }
    format!(
        "the enrolment window is closed: this identity table already holds {} principals and {} \
         keys but carries no bootstrap record, so it was written before the record existed.",
        state.principals, state.keys
    )
}

#[tonic::async_trait]
impl Bootstrap for BootstrapService {
    async fn enroll(
        &self,
        request: Request<EnrollRequest>,
    ) -> Result<Response<EnrollResponse>, Status> {
        let request = request.into_inner();
        if request.name.trim().is_empty() {
            return Err(Status::invalid_argument(
                "`name` is required: it is both the user's id and the account it owns, the \
                 <account> in <host>/<account>/<repo>",
            ));
        }
        if request.openssh_public_key.trim().is_empty() {
            return Err(Status::invalid_argument(
                "`openssh_public_key` is required: one authorized_keys line, e.g. \
                 \"ssh-ed25519 AAAAC3Nz… name@host\"",
            ));
        }
        let name = request.name.trim();

        match self
            .ctx
            .control
            .enroll_first_identity(name, request.openssh_public_key.trim())
        {
            Ok(enrolled) => {
                tracing::warn!(
                    target: crate::auth::AUDIT_TARGET,
                    event = "grpc.call",
                    method = METHOD_ENROLL,
                    principal = %enrolled.principal.id,
                    fingerprint = %enrolled.key.fingerprint,
                    account = %enrolled.account,
                    decision = "allow",
                    "THE UNAUTHENTICATED ENROLMENT WINDOW IS NOW CLOSED. This principal is \
                     gunnar-admin and is intrinsically admin of its own account. Nothing else \
                     on this server will ever be served without a credential."
                );
                let at = enrolled.bootstrapped_at_unix_ms;
                Ok(Response::new(EnrollResponse {
                    principal: Some(enrolled.principal),
                    key: Some(enrolled.key),
                    account: enrolled.account,
                    bootstrapped_at_unix_ms: at,
                    client_cert_pem: enrolled.client_cert_pem,
                    client_key_pem: enrolled.client_key_pem,
                    issuer_cert_pem: enrolled.issuer_cert_pem,
                }))
            }
            Err(err) => {
                tracing::warn!(
                    target: crate::auth::AUDIT_TARGET,
                    event = "grpc.call",
                    method = METHOD_ENROLL,
                    principal = %name,
                    decision = "deny",
                    reason = %err,
                    "Bootstrap.Enroll refused"
                );
                Err(Status::from(err))
            }
        }
    }

    async fn become_twin(
        &self,
        request: Request<BecomeTwinRequest>,
    ) -> Result<Response<BecomeTwinResponse>, Status> {
        // **Who is making this box a twin, proved** (h-bootstrap-open, 2026-09-28): a
        // credential bound to this method, from a key the box was FED — the same rule
        // `Restore` keeps, because a twin follows whatever primary it is told to.
        self.fed_caller(request.metadata(), crate::auth::BOOTSTRAP_BECOME_TWIN_METHOD, METHOD_BECOME_TWIN)?;
        let primary = request.into_inner().primary;
        // Already a twin before this call: `become_twin` answers the same key for the same
        // primary (and refuses any other), so nothing changes — said at INFO, not as news.
        let already = self.ctx.control.is_twin();
        match self.ctx.control.become_twin(primary.trim()) {
            Ok(key) if already => {
                let restart_required = self.ctx.control.twin_git_listener_needs_restart();
                tracing::info!(
                    target: crate::auth::AUDIT_TARGET,
                    event = "grpc.call",
                    method = METHOD_BECOME_TWIN,
                    primary = %primary,
                    restart_required,
                    decision = "allow",
                    "this box is already the twin of {}: asked again, it answered the same key (nothing changed)",
                    primary.trim()
                );
                Ok(Response::new(BecomeTwinResponse {
                    twin_public_key: key,
                    detail: format!("already a twin of {primary}; enrol this key on the primary in group `{}`", crate::TWIN_GROUP),
                    restart_required,
                }))
            }
            Ok(key) => {
                // **The socket that could not move** (2026-09-21, lane T4). The
                // git bind is chosen once, at start-up, and a role conferred
                // over the wire arrives after it. Nothing is served through it
                // — every git service on a twin is refused by name at the door
                // from this instant — but the socket is still on a primary's
                // address, and only a restart moves it. Said here rather than
                // left for somebody to find with `ss`.
                let restart_required = self.ctx.control.twin_git_listener_needs_restart();
                tracing::warn!(
                    target: crate::auth::AUDIT_TARGET,
                    event = "grpc.call",
                    method = METHOD_BECOME_TWIN,
                    primary = %primary,
                    restart_required,
                    decision = "allow",
                    "THIS BOX IS NOW A TWIN: it gave up its own log, froze, and follows the primary. \
                     Enrol the key below on the primary in group `twin`."
                );
                if restart_required {
                    tracing::warn!(
                        target: crate::auth::AUDIT_TARGET,
                        event = "grpc.call",
                        method = METHOD_BECOME_TWIN,
                        "this box was already listening when it was told it is a twin: its GIT LISTENER is still bound where a \
                         primary's was and only a restart moves it to loopback. It serves no git in the meantime — every git \
                         service is refused by name — but the socket is on the wrong address until the box comes back."
                    );
                }
                Ok(Response::new(BecomeTwinResponse {
                    twin_public_key: key,
                    detail: format!(
                        "a twin of {primary}; enrol this key on the primary in group `{}`{}",
                        crate::TWIN_GROUP,
                        if restart_required {
                            ". RESTART THIS BOX: its git listener was bound before it was told it is a twin, so the socket is \
                             still on a primary's address. No git is served through it — every request is refused by name — but \
                             only a restart puts it on loopback"
                        } else {
                            ""
                        }
                    ),
                    restart_required,
                }))
            }
            Err(err) => {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_BECOME_TWIN, primary = %primary, decision = "deny", reason = %err, "Bootstrap.BecomeTwin refused");
                Err(Status::from(err))
            }
        }
    }

    async fn assign_tunnel_address(
        &self,
        request: Request<AssignTunnelAddressRequest>,
    ) -> Result<Response<AssignTunnelAddressResponse>, Status> {
        let address = request.into_inner().address;
        let address = address.trim();
        match self.ctx.control.assign_tunnel_address(address) {
            Ok(answer) => {
                if answer.changed {
                    tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_ASSIGN_TUNNEL_ADDRESS, address = %answer.address, wg_public_key = %answer.wg_public_key, tunnel_up = answer.tunnel_up, decision = "allow", "THIS BOX'S TUNNEL ADDRESS WAS HANDED OVER: register its key on the hub at this address");
                }
                Ok(Response::new(answer))
            }
            Err(err) => {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_ASSIGN_TUNNEL_ADDRESS, address = %address, decision = "deny", reason = %err, "Bootstrap.AssignTunnelAddress refused");
                Err(Status::from(err))
            }
        }
    }

    async fn twin_csr(&self, _request: Request<TwinCsrRequest>) -> Result<Response<TwinCsrResponse>, Status> {
        match self.ctx.control.twin_csr() {
            Ok(csr_pem) => {
                tracing::info!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_TWIN_CSR, decision = "allow", "Bootstrap.TwinCsr: the twin's replication request (its key stays here)");
                Ok(Response::new(TwinCsrResponse { csr_pem }))
            }
            Err(err) => {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_TWIN_CSR, decision = "deny", reason = %err, "Bootstrap.TwinCsr refused");
                Err(Status::from(err))
            }
        }
    }

    async fn install_twin_certificate(
        &self,
        request: Request<InstallTwinCertificateRequest>,
    ) -> Result<Response<InstallTwinCertificateResponse>, Status> {
        // The issuer a twin pins is the one it trusts for good: only from a fed key.
        self.fed_caller(request.metadata(), crate::auth::BOOTSTRAP_INSTALL_TWIN_CERTIFICATE_METHOD, METHOD_INSTALL_TWIN_CERTIFICATE)?;
        let r = request.into_inner();
        let port = u16::try_from(r.replication_port)
            .map_err(|_| Status::invalid_argument(format!("replication_port {} is not a port", r.replication_port)))?;
        match self.ctx.control.install_twin_certificate(&r.cert_pem, &r.ca_pem, port, &r.replication_host) {
            Ok(days_left) => {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_INSTALL_TWIN_CERTIFICATE, days_left, replication_port = port, decision = "allow", "THIS TWIN IS PAIRED FOR MUTUAL TLS: it pins its primary's issuer and follows it over TLS 1.3 from the next attempt");
                Ok(Response::new(InstallTwinCertificateResponse {
                    days_left,
                    detail: format!("paired: {days_left} day(s) of certificate; the follower picks it up on its next attempt"),
                }))
            }
            Err(err) => {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_INSTALL_TWIN_CERTIFICATE, decision = "deny", reason = %err, "Bootstrap.InstallTwinCertificate refused");
                Err(Status::from(err))
            }
        }
    }

    async fn restore(
        &self,
        request: Request<tonic::Streaming<LogChunk>>,
    ) -> Result<Response<RestoreResponse>, Status> {
        // **Who is streaming, proved, before the box gives up anything** (h-v1-6a,
        // 2026-09-28). Checked against the keys this box was FED, never the identity
        // table a fresh box does not have; the answer is the owner `restore_begin`
        // records, so the resume branch serves only the key that started it.
        let keys = self.ctx.control.restore_keys().map_err(Status::from)?;
        let owner = crate::auth::authenticate_restore(self.ctx.config.auth, &keys, &self.ctx.replay, request.metadata(), crate::bus::now_unix_ms())
            .inspect_err(|status| {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, decision = "deny", reason = %status.message(), "Bootstrap.Restore refused: the stream's credential");
            })?;
        let mut stream = request.into_inner();
        // **The box's ONE intake for this stream** (restore-large-entry, 2026-10-07): the owner
        // check and the next sequence (`restore_begin`), and on the server the staging area an
        // entry larger than one piece is written into as it arrives. A stream the sender
        // abandoned may still hold it for a moment — its last apply has not returned — so this
        // waits a little for it instead of racing it over the same staged entry.
        let waited = std::time::Instant::now();
        let mut intake = loop {
            match self.ctx.control.restore_intake(&owner) {
                Ok(Some(intake)) => break intake,
                Ok(None) if waited.elapsed() < RESTORE_INTAKE_WAIT => tokio::time::sleep(std::time::Duration::from_millis(100)).await,
                Ok(None) => {
                    return Err(Status::unavailable(format!(
                        "another Bootstrap.Restore stream still holds this box after {} s (an abandoned stream whose last \
                         apply has not returned); nothing of this stream was taken — the next call resumes where the box is",
                        RESTORE_INTAKE_WAIT.as_secs()
                    )))
                }
                Err(err) => {
                    tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, owner = %owner, decision = "deny", reason = %err, "Bootstrap.Restore refused");
                    return Err(Status::from(err));
                }
            }
        };
        tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, owner = %owner, decision = "allow", "Bootstrap.Restore: a fresh box takes its twin's log");
        // The head the sender announced (the largest `restore_head` seen): a stream that ends
        // below it ended SHORT, however cleanly, and the box refuses to finish.
        let mut announced = 0u64;
        // **Which log is being streamed** (2026-09-21, lane T4). Every chunk
        // carries it, like `restore_head`; the first non-empty one is kept and
        // `restore_finish` writes it, so the restored box IS that history and
        // the twin it was built from keeps following it. A stream that changes
        // it mid-way is two senders or two logs, and is refused rather than
        // silently taking whichever arrived last.
        let mut announced_id = String::new();
        // **The room check, once, before the first byte of the restore lands.** A set that left
        // four members and arrives at two may hold less than what is coming; without this the
        // stream fills the volume and dies in the middle, an hour in, as an ENOSPC. The refusal
        // says how much is short and leaves the box fresh.
        let mut room_checked = false;
        // Every way out of this loop drops the intake, and dropping it keeps what it staged of an
        // unfinished entry: a stream that broke, one the sender cancelled at its deadline (tonic
        // reads a client's RST_STREAM as a CLEAN end) and one refused below all leave the box
        // holding every byte it took, durably.
        // **A sender that went silent lets go of the box** (restore-large-entry): one stream at a
        // time holds the intake, so a connection that died without a word (no RST reached this
        // box) must not hold it for ever. A sender with nothing more to say ends its stream.
        loop {
            let chunk = match tokio::time::timeout(RESTORE_IDLE, stream.message()).await {
                Ok(next) => match next? {
                    Some(chunk) => chunk,
                    None => break,
                },
                Err(_) => {
                    let (applied, _) = intake.counts();
                    let said = intake.end_of_stream().err().map(|e| format!(": {e}")).unwrap_or_default();
                    tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, applied, idle_s = RESTORE_IDLE.as_secs(), decision = "deny", "Bootstrap.Restore: the sender went silent; the stream is ended and the box stays restoring");
                    return Err(Status::deadline_exceeded(format!(
                        "no chunk arrived for {} s; this stream is ended and the box stays restoring — the next stream resumes where it is{said}",
                        RESTORE_IDLE.as_secs()
                    )));
                }
            };
            announced = announced.max(chunk.restore_head);
            let chunk_id = chunk.log_id.trim();
            if !chunk_id.is_empty() {
                if announced_id.is_empty() {
                    announced_id.push_str(chunk_id);
                } else if announced_id != chunk_id {
                    return Err(Status::invalid_argument(format!(
                        "this stream began as log {announced_id} and is now log {chunk_id}: one restore is one history, \
                         and a box built out of two of them is not a restore of either"
                    )));
                }
            }
            if !room_checked {
                room_checked = true;
                self.ctx.control.restore_room(chunk.restore_bytes).map_err(|err| {
                    tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, announced_bytes = chunk.restore_bytes, decision = "deny", reason = %err, "Bootstrap.Restore refused for room; nothing of the stream was written");
                    Status::from(err)
                })?;
            }
            intake.take(&chunk).map_err(Status::from)?;
        }
        let (applied, lost) = intake.counts();
        if let Err(err) = intake.end_of_stream() {
            tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, applied, decision = "deny", reason = %err, "Bootstrap.Restore: the stream ended inside an entry; the box stays restoring");
            return Err(Status::from(err));
        }
        drop(intake);
        let head = self.ctx.control.restore_finish(announced, &announced_id).map_err(|err| {
            tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, announced, applied, decision = "deny", reason = %err, "Bootstrap.Restore did not finish; the box stays restoring");
            Status::from(err)
        })?;
        tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = METHOD_RESTORE, head, applied, lost, decision = "allow", "Bootstrap.Restore finished; the box must restart");
        Ok(Response::new(RestoreResponse {
            head_seq: head,
            applied,
            lost,
            restart_required: true,
            detail: format!("restored to seq {head} ({applied} applied, {lost} replayed CAS lost); restart the box — it opens as the primary it was"),
        }))
    }

    async fn status(
        &self,
        _request: Request<BootstrapStatusRequest>,
    ) -> Result<Response<BootstrapStatusResponse>, Status> {
        // A read fault is UNAVAILABLE, never a cheerful "closed". An operator
        // who is told the door is shut because the disk could not be read has
        // been told the most dangerous possible lie about this server.
        let state = self.ctx.control.bootstrap_state().map_err(Status::from)?;
        let detail = describe(&state);
        let (facts, facts_detail) = fresh_facts(self.ctx.control.as_ref(), state.window_open);
        tracing::debug!(
            target: crate::auth::AUDIT_TARGET,
            event = "grpc.call",
            method = METHOD_STATUS,
            window_open = state.window_open,
            decision = "allow",
            "Bootstrap.Status"
        );
        Ok(Response::new(BootstrapStatusResponse {
            window_open: state.window_open,
            bootstrapped: state.bootstrapped,
            bootstrapped_at_unix_ms: state.bootstrapped_at_unix_ms,
            bootstrapped_principal: state.bootstrapped_principal,
            bootstrapped_fingerprint: state.bootstrapped_fingerprint,
            principals: state.principals,
            keys: state.keys,
            detail,
            serving_git: state.serving_git,
            readiness: state.readiness,
            build: state.build,
            role: state.role,
            // Read at answer time, and unauthenticated on purpose: see the
            // field's note in bootstrap.proto.
            server_unix_ms: estate_clock::now_unix_ms(),
            clock_last: estate_clock::report::last().map(crate::service::admin::clock_sync),
            clock_skew_syncs_suppressed: estate_clock::trigger::suppressed(),
            primary: self.ctx.control.primary_heartbeat(self.ctx.heartbeats.bounds().silent_after_ms),
            facts,
            facts_detail,
            recharge_from: state.recharge_from,
            restoring: state.restoring,
            log_head_seq: state.log_head_seq,
            restore_partial_seq: state.restore_partial.map(|(seq, _)| seq).unwrap_or_default(),
            restore_partial_bytes: state.restore_partial.map(|(_, bytes)| bytes).unwrap_or_default(),
            restore_snapshot_resumable: state.restore_snapshot_resumable,
            restore_snapshot_id: state.restore_snapshot.as_ref().map(|s| s.0.clone()).unwrap_or_default(),
            restore_snapshot_items: state.restore_snapshot.as_ref().map(|s| s.1).unwrap_or_default(),
            restore_snapshot_bytes: state.restore_snapshot.as_ref().map(|s| s.2).unwrap_or_default(),
            restore_snapshot_whole: state.restore_snapshot.as_ref().is_some_and(|s| s.3),
        }))
    }

    /// **The runtime growth's phase, without a credential.** The one reader
    /// `Admin.GrowStatus` also answers from; here for the actor outside the
    /// box (T6 in `DATA-SET-GROWTH-FLOW.md`), which owns the volumes and has
    /// nothing to sign with while the identity table is off the stripe — or,
    /// after a crash in that window, while there is no identity table to
    /// sign against at all.
    ///
    /// A read fault is UNAVAILABLE here as on `Status`: an outside actor told
    /// `idle` because the journal could not be read would leave the volumes
    /// detached.
    async fn grow_status(
        &self,
        _request: Request<GrowStatusRequest>,
    ) -> Result<Response<GrowStatusResponse>, Status> {
        let status = self.ctx.control.grow_status().map_err(Status::from)?;
        tracing::debug!(
            target: crate::auth::AUDIT_TARGET,
            event = "grpc.call",
            method = METHOD_GROW_STATUS,
            phase = %status.phase,
            decision = "allow",
            "Bootstrap.GrowStatus"
        );
        Ok(Response::new(status))
    }
}

/// **The machine, for the twin election, while the box is FRESH** (T17, 2026-09-21, for T15's
/// TD8). `xtask elect twin` ranks two boxes by cores, RAM, the data root's size and the
/// start-up disk test — facts that live on the authenticated `Admin.Capacity`. A fresh box
/// accepts nobody's key, and enrolling first is no way round it because `BecomeTwin` requires
/// a FRESH box: the election stopped with `elect/box-silent`.
///
/// So `Status` carries them while the window is open, and ONLY then. They come out of the
/// control plane's own `capacity` answer (asked for nothing: no namespace, no bytes), so the
/// numbers are exactly what `Admin.Capacity` would say — one source, never a second probe.
/// Nothing tenant-shaped is copied: no namespace, no data set, no root path, no verdict.
/// Once anybody has enrolled the facts are withheld and the sentence says where they are.
pub(crate) fn fresh_facts(control: &dyn ControlPlane, window_open: bool) -> (Option<gunnar_proto::BoxFacts>, String) {
    if !window_open {
        return (
            None,
            "withheld: this box is enrolled, so its machine is read with a credential — \
             Admin.Capacity carries the same numbers"
                .to_owned(),
        );
    }
    match control.capacity(&gunnar_proto::CapacityRequest::default()) {
        Ok(c) => {
            // `total_bytes` is a plain uint64 on Capacity and 0 when the statvfs failed
            // (`CAPACITY_UNMEASURED`); here that failure is an ABSENCE, never a zero.
            let measured = c.verdict != gunnar_proto::CapacityVerdict::CapacityUnmeasured as i32;
            let data_total_bytes = (measured && c.total_bytes > 0).then_some(c.total_bytes);
            (
                Some(gunnar_proto::BoxFacts {
                    cpu_millicores_total: c.cpu_millicores_total,
                    ram_bytes_total: c.ram_bytes_total,
                    data_total_bytes,
                    disk_test: c.disk_test,
                    disk_seq_write_bps: c.disk_seq_write_bps,
                    disk_seq_read_bps: c.disk_seq_read_bps,
                    disk_sync_milliops_per_s: c.disk_sync_milliops_per_s,
                    disk_read_how: c.disk_read_how,
                    disk_reason: c.disk_reason,
                    machine_reason: c.machine_reason,
                }),
                if data_total_bytes.is_some() {
                    "this box is fresh: its machine is here for the twin election, without a \
                     credential, until the first enrolment"
                        .to_owned()
                } else {
                    format!(
                        "this box is fresh, but its data root could not be measured, so \
                         data_total_bytes is absent: {}",
                        c.reason
                    )
                },
            )
        }
        // A plane that cannot answer Capacity cannot answer this either; said, not zeroed.
        Err(err) => (None, format!("this box is fresh, but its machine could not be read: {err}")),
    }
}

impl BootstrapService {
    /// [`crate::auth::authenticate_bootstrap`] against this box's fed keys, audited.
    fn fed_caller(&self, metadata: &tonic::metadata::MetadataMap, method: &str, audited: &str) -> Result<String, Status> {
        let keys = self.ctx.control.restore_keys().map_err(Status::from)?;
        crate::auth::authenticate_bootstrap(self.ctx.config.auth, method, &keys, &self.ctx.replay, metadata, crate::bus::now_unix_ms()).inspect_err(|status| {
            tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = audited, decision = "deny", reason = %status.message(), "{audited} refused: the caller's credential");
        })
    }
}