gunnar-grpc 1.5.0

//! **gRPC authentication and authorisation: the v1 allow-all hole, closed.**
//!
//! # The mechanism: SSH-key-derived identity
//!
//! A caller proves it holds the private half of a key that is **already in
//! gunnar's ACL key table** — the same table the SSH listener authenticates git
//! push and fetch against — by signing a short, single-use challenge with it.
//! The signature is an **`sshsig`** (`SSHSIG`, the format `ssh-keygen -Y sign`
//! produces and the one `gunnar-policy` already verifies for signed commits),
//! so the credential can be produced by `ssh-agent`, by `ssh-keygen`, or by
//! `ssh-key` in-process, with no new key material anywhere.
//!
//! ```text
//! metadata:  gunnar-authorization: gunnar-sshsig-v2 <fingerprint> <issued_at_ms> <nonce> <method> <armor>
//!
//! signed:    gunnar-grpc-auth-v2\n<fingerprint>\n<issued_at_ms>\n<nonce>\n<method>\n
//! ```
//!
//! Verification is five steps and every one of them can refuse:
//!
//! 1. the header is present and parses as `gunnar-sshsig-v2` (or the
//!    method-blind `gunnar-sshsig-v1`, see [`CREDENTIAL_SCHEME_V1`]);
//! 2. `issued_at_ms` is inside [`MAX_CLOCK_SKEW_MS`] of now;
//! 3. `<method>` is the method this request is actually calling
//!    ([`CalledMethod`]);
//! 4. `<fingerprint>:<nonce>` has not been seen before ([`ReplayGuard`]);
//! 5. the `sshsig` verifies against the OpenSSH public key the **ACL key table**
//!    holds for that fingerprint, in namespace [`SSHSIG_NAMESPACE`].
//!
//! Only then does a [`Caller`] exist. Authorisation is a separate question and
//! is **never answered here**: it is delegated to
//! [`Identity::may`](crate::seam::Identity::may), which the real adapter
//! answers straight out of `gunnar_core::AuthTable` (the ACL model). There is one grant
//! decision in gunnar and this module is not a second one (one implementation).
//!
//! # Why not mTLS, and why not a static bearer token
//!
//! - **mTLS** means a second identity model — a CA, certificate issuance,
//!   renewal, revocation and a mapping from subject back to a gunnar principal
//!   — living beside the key table that already answers the same question.
//!   Two identity models that can disagree is exactly the twinning the one-sandbox rule had to
//!   undo once already.
//! - **A static bearer token** is one secret that grants everything. It has no
//!   principal behind it, so an audit line can only ever say "the token", and
//!   revoking one operator means rotating every operator. `SecretStore` (the secret-store seam)
//!   would give it a home, but a home for a shared password is still a shared
//!   password.
//! - **SSH-key identity** reuses the key table, the one fingerprint, the
//!   principal model and the ACL. Revoking a key over gRPC revokes the same
//!   key for git in the same instant, because there is only one table. Adding
//!   an operator is `gunnar key add`, which already exists.
//!
//! What it costs: a caller must sign, so a credential cannot be pasted into
//! `grpcurl` by hand. That is the price of not having a bearer secret, and it
//! is paid by the client library, once.
//!
//! # The challenge binds the method, and why it did not used to
//!
//! Until 2026-09-11 the signed bytes were `fingerprint‖timestamp‖nonce` and
//! nothing else. The argument for that was written down here: a tonic **server**
//! interceptor is handed `Request<()>` and cannot see the `:path`
//! (`GrpcMethod` is a *client*-side extension), so binding the method looked
//! like it meant verifying the signature inside every handler instead of once
//! at the edge — and a handler that forgot would be an open door. Single use
//! ([`ReplayGuard`]) was offered as the compensating control.
//!
//! **Single use is not method binding.** It stops a credential being used
//! twice; it does not stop it being used *once, somewhere else*. An on-path
//! attacker (the control plane is plaintext in every shipped medium variant)
//! takes the header off a live `Admin.Health` call, drops the original, and
//! sends it on with `:path = /gunnar.v1.Access/AddKey`. One RPC, one nonce,
//! permanent admin — and the audit line names the victim. That is GUN-001 of
//! the 2026-09-10 review.
//!
//! The premise was wrong in one place. The interceptor cannot read the `:path`,
//! but [`crate::trace::RequestSpanLayer`] — already wrapping the whole routes
//! table, already reading `request.uri().path()` for the span — *can*, and a
//! `tonic::Request` carries its `http::Extensions` through
//! `InterceptedService` untouched. So the layer puts the path in the extensions
//! as [`CalledMethod`] and the gate compares it with what the caller signed.
//! One check, at the edge, for all 88 RPCs, against the method the router is
//! actually about to dispatch to — not against a `&'static str` a handler
//! passes to `require` and could pass wrongly.
//!
//! # The v1 credential is refused (5.3.3, OWNER 2026-10-10)
//!
//! [`CREDENTIAL_SCHEME_V1`] is still **parsed**, so that it can be refused by
//! name, and refused it is — at every method, on every listener, with no flag
//! and no builder that turns the refusal off. A v1 credential binds no method,
//! so accepting it at all is GUN-001: a header lifted off `Admin.Health` opens
//! `Access.AddKey`. A **v2** credential is method-checked unconditionally: there
//! is no mode in which a bound credential is accepted at the wrong method.
//!
//! Until 5.3.3 the refusal was opt-in (`serve --grpc-require-bound-credentials`,
//! off by default) because the console was thought to still sign v1. That flag
//! is gone, not inverted: a switch that re-opens GUN-001 is the switch that gets
//! flipped during an incident and stays flipped.
//!
//! ## Every client signs v2 (measured 2026-10-10)
//!
//! - gunnar's own callers, through [`crate::client_signer::CredentialSigner`]
//!   ([`sign_credential_for`] on the method tonic is about to call):
//!   `gunnar-cli`, `gunnar-front` (operator door, off-site copy), the twin
//!   following its primary (`gunnar::twin`), the pairing door;
//! - `gunnar-server`'s replication to a browse replica (`Repos.Delete` /
//!   `Repos.Create`, `mirror.rs`), which signed v1 until this change and now
//!   signs each call for its method;
//! - `gunnar-monetize-guest`, whose host mints `gunnar-sshsig-v2` per method;
//! - outside this repository: `gunnar-ui` (`signer.rs` `credential_for_now`,
//!   v2 since 2026-09-11), monetize's `products/gunnar` (`signer.rs`, v2) and
//!   `private-gunnar-ops`' xtask (`sign_credential_for`).
//!
//! ## The migration window, for a real fleet
//!
//! Source is not deployment: a binary built before its client switched to v2
//! still signs v1. Such a client is refused with `UNAUTHENTICATED` naming the
//! scheme to send, and `Admin.Health` (`unbound_credentials_refused`, and a row
//! per fingerprint and method) names it from the server's side. Before
//! upgrading a server, read `gunnar control health` on the running one: a
//! non-zero `v1 seen` (served) there, after an uptime longer than the longest
//! interval any client calls at, names exactly the clients that this upgrade
//! will cut off — upgrade those first. A gunnar **primary** replicating to a
//! browse replica must be upgraded before (or with) the replica: an older
//! primary signs its `Repos.Delete`/`Repos.Create` v1, and an upgraded replica
//! refuses them (the replication worker retries until the primary is upgraded).
//! There is no rollback switch; the rollback is the previous binary.

use std::collections::BTreeSet;
use std::sync::{Arc, Mutex};

use gunnar_proto::{GrantLevel, RepoPath};
use tonic::{Request, Status};

use crate::seam::ControlPlane;

/// **The tag of the answer to a key that is being enrolled**
/// ([`AuthFailure::Enrolling`]): `UNAVAILABLE`, ask again. A caller (the
/// front's certify loop) and the request log key on it, as they key on
/// `[primary-growing]`.
pub const KEY_ENROLLING: &str = "[key-enrolling]";

/// The gRPC metadata key a credential travels in.
///
/// Not `authorization`: that name is owned by bearer/basic schemes and a proxy
/// in the path may rewrite or strip it. This one is gunnar's own.
pub const CREDENTIAL_METADATA_KEY: &str = "gunnar-authorization";

/// The credential scheme token **a client should send**. Versioned so a second
/// scheme can be added without guessing what an unlabelled header meant — which
/// is exactly what the method binding needed.
pub const CREDENTIAL_SCHEME: &str = "gunnar-sshsig-v2";

/// The method-blind scheme, still parsed so that it is refused by name. It
/// binds no method, and [`AuthGate`] refuses it at every method.
pub const CREDENTIAL_SCHEME_V1: &str = "gunnar-sshsig-v1";

/// The `sshsig` namespace gunnar signs gRPC challenges under.
///
/// A signature made for one namespace does not verify in another, so a commit
/// signature (`git`, which OpenSSH signs under `git`) can never be replayed as
/// a control-plane credential, and vice versa.
pub const SSHSIG_NAMESPACE: &str = "gunnar-grpc";

/// The first line of the **method-blind** signed challenge. Domain separation,
/// so the bytes gunnar asks a key to sign can never be mistaken for anything
/// else.
pub const CHALLENGE_PREAMBLE: &str = "gunnar-grpc-auth-v1";

/// The first line of the **method-bound** signed challenge.
///
/// A separate preamble, not a sixth line under the old one: a v1 verifier
/// handed v2 bytes must fail to verify rather than succeed on a prefix, and the
/// only way to guarantee that is for the first line to differ.
pub const CHALLENGE_PREAMBLE_V2: &str = "gunnar-grpc-auth-v2";

/// How far a credential's `issued_at` may be from the server's clock, in
/// either direction. Five minutes: enough for an unsynchronised laptop,
/// short enough that [`ReplayGuard`]'s memory stays small.
pub const MAX_CLOCK_SKEW_MS: i64 = nornir_auth_sshsig::MAX_CLOCK_SKEW_MS;

/// **The words every refusal for clock skew carries** — the `Stale` status and the bootstrap
/// gate's. A caller tells a skew refusal from any other by them (a twin whose own poll is
/// refused for skew asks the front for the time, `gunnar_server::twin`); monetize's guest keeps
/// the same phrase (`gunnar_monetize_guest::CLOCK_SKEW_PHRASE`).
pub const CLOCK_SKEW_PHRASE: &str = "away from this server's clock";

/// The shortest nonce accepted. A nonce is the only thing making a credential
/// single-use; a short one collides and starts refusing honest callers.
pub const MIN_NONCE_LEN: usize = nornir_auth_sshsig::MIN_NONCE_LEN;

/// The `tracing` target every audit record is emitted on.
///
/// A stable target rather than a message prefix, so an operator can route the
/// audit trail somewhere else (`RUST_LOG=gunnar::audit=info`) and a test can
/// assert on it without matching prose.
pub const AUDIT_TARGET: &str = "gunnar::audit";

/// The ACL principal id that means "nobody": `gunnar_core::ANONYMOUS_PRINCIPAL_ID`
///
///
/// **The control plane has no anonymous mode and must never grow one.** The
/// anonymous principal exists so a stranger can *clone a public repository over
/// SSH*; it is not a caller, it holds no key, and no credential can be signed
/// on its behalf. [`AuthGate::authenticate`] refuses it outright anyway; the
/// note there says why a check that "cannot fire" is still worth its three
/// lines.
///
/// Spelled as a literal because this crate deliberately does not link
/// `gunnar-core` (see [`Access`]). `gunnar-cli` links both and
/// `tests/anonymous_access.rs` asserts the two constants are the same string,
/// so the copy cannot drift in silence.
pub const ANONYMOUS_PRINCIPAL_ID: &str = "anonymous";

// ───────────────────────────────────────────────────────────────────────────
// Mode
// ───────────────────────────────────────────────────────────────────────────

/// Whether the control plane requires a credential.
///
/// [`AuthMode::Required`] is the default and the only mode a server should
/// ever be deployed in. The other exists because a developer poking at a
/// freshly built server with no key registered still has to be able to reach
/// it, and the honest answer to that is a named, loud switch rather than a
/// silent hole.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum AuthMode {
    /// Every RPC needs a verified credential and a grant. **The default.**
    #[default]
    Required,
    /// **DEVELOPMENT ONLY.** Every RPC is served to every caller, with no
    /// credential and no grant check. The v1 allow-all behaviour, kept as an
    /// escape hatch and never reached by default.
    ///
    /// Only [`crate::ServerConfig::insecure_allow_all_for_local_development`]
    /// constructs it, it logs at `warn!` when it is constructed and again at
    /// every server start, `Admin.Health` reports
    /// `grpc_authentication_enabled: false`, and every audit record it produces
    /// carries `auth_mode="allow-all-insecure"`.
    AllowAllInsecure,
}

impl AuthMode {
    /// True when a credential is required. This is what `Admin.Health` reports.
    pub fn is_required(&self) -> bool {
        matches!(self, AuthMode::Required)
    }

    /// The value that goes on every audit record.
    pub fn as_str(&self) -> &'static str {
        match self {
            AuthMode::Required => "required",
            AuthMode::AllowAllInsecure => "allow-all-insecure",
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// What is being asked for
// ───────────────────────────────────────────────────────────────────────────

/// The scope an RPC needs a grant on.
///
/// It mirrors `gunnar_core::Scope` (the ACL model), `Server` included: without it the
/// bootstrap admin could not create the first namespace. It is expressed in
/// `gunnar-proto` terms because this crate does not link `gunnar-core`. The
/// adapter translates; it does not re-decide.
#[derive(Clone, Debug, PartialEq)]
pub enum Access {
    /// The server itself: namespaces, principals, keys, grants, health.
    Server,
    /// One namespace, matched **exactly**: `team` does not reach `team/sub`
    /// (the one-sandbox rule, and grants do not inherit).
    Namespace(String),
    /// One repository.
    Repo(RepoPath),
}

impl Access {
    /// A repository, without cloning at the call site.
    pub fn repo(path: &RepoPath) -> Access {
        Access::Repo(path.clone())
    }

    /// A namespace.
    pub fn namespace(name: &str) -> Access {
        Access::Namespace(name.to_owned())
    }
}

impl std::fmt::Display for Access {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Access::Server => f.write_str("server"),
            Access::Namespace(name) => write!(f, "namespace:{name}"),
            Access::Repo(path) => write!(f, "repo:{}/{}", path.namespace, path.repo),
        }
    }
}

/// The level name that goes on an audit record.
pub fn level_name(level: GrantLevel) -> &'static str {
    match level {
        GrantLevel::Unspecified => "unspecified",
        GrantLevel::Read => "read",
        GrantLevel::Write => "write",
        GrantLevel::Admin => "admin",
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Who is asking
// ───────────────────────────────────────────────────────────────────────────

/// An authenticated caller, attached to a request by [`AuthGate`].
///
/// Its presence in a request's extensions is the *only* evidence a handler
/// accepts that a credential was verified. A handler cannot manufacture one
/// from metadata, and there is no `Default`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Caller {
    /// The ACL principal id the presented key belongs to.
    pub principal_id: String,
    /// The `SHA256:…` fingerprint of the key that was presented. Empty under
    /// [`AuthMode::AllowAllInsecure`], which is how an audit record can be told
    /// apart from a real one.
    pub fingerprint: String,
    /// The key belongs to a `Machine` principal (the ACL model): a peer server's deploy
    /// key. Read by the browse replica's mutator refusal, which lets exactly
    /// one write through and only from a machine (the browse replica).
    /// Always `false` for a certificate caller and under allow-all.
    pub machine: bool,
    /// False only under [`AuthMode::AllowAllInsecure`].
    pub authenticated: bool,
}

impl Caller {
    /// The synthetic caller the development escape hatch serves everything as.
    ///
    /// Named, not empty: an audit trail that said `principal=""` would look
    /// like a bug rather than like a decision.
    pub const INSECURE_PRINCIPAL: &'static str = "(unauthenticated: allow-all-insecure)";

    fn insecure() -> Caller {
        Caller {
            principal_id: Caller::INSECURE_PRINCIPAL.to_owned(),
            fingerprint: String::new(),
            machine: false,
            authenticated: false,
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The credential
// ───────────────────────────────────────────────────────────────────────────

/// **gunnar's credential scheme** — its wire format, byte for byte as it was
/// before the verifier moved to edda (`nornir-auth-sshsig`, workspace RULE 2:
/// monetize's control API checks callers with the same code). The constants
/// above ARE the scheme; this type hands them to the shared crate.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub struct GunnarScheme;

impl nornir_auth_sshsig::Scheme for GunnarScheme {
    const METADATA_KEY: &'static str = CREDENTIAL_METADATA_KEY;
    const SCHEME: &'static str = CREDENTIAL_SCHEME;
    const SCHEME_UNBOUND: Option<&'static str> = Some(CREDENTIAL_SCHEME_V1);
    const NAMESPACE: &'static str = SSHSIG_NAMESPACE;
    const PREAMBLE: &'static str = CHALLENGE_PREAMBLE_V2;
    const PREAMBLE_UNBOUND: &'static str = CHALLENGE_PREAMBLE;
    type Error = AuthFailure;
}

impl From<nornir_auth_sshsig::CredentialError> for AuthFailure {
    fn from(e: nornir_auth_sshsig::CredentialError) -> AuthFailure {
        match e {
            nornir_auth_sshsig::CredentialError::Missing => AuthFailure::Missing,
            nornir_auth_sshsig::CredentialError::Malformed(why) => AuthFailure::Malformed(why),
        }
    }
}

/// A parsed, not-yet-verified credential (`parse`, `to_header_value`,
/// `challenge`, `from_metadata` are the shared crate's, under [`GunnarScheme`]).
///
/// `method: None` is a [`CREDENTIAL_SCHEME_V1`] credential: it names no method
/// and could therefore be relayed to any of them. That is GUN-001, and it is
/// why [`AuthGate`] refuses it.
pub type Credential = nornir_auth_sshsig::Credential<GunnarScheme>;

/// **The one writer of the method-blind challenge bytes** ([`CREDENTIAL_SCHEME_V1`]).
pub fn challenge(fingerprint: &str, issued_at_unix_ms: i64, nonce: &str) -> String {
    nornir_auth_sshsig::challenge::<GunnarScheme>(fingerprint, issued_at_unix_ms, nonce)
}

/// **The one writer of the method-bound challenge bytes.**
pub fn challenge_bound(fingerprint: &str, issued_at_unix_ms: i64, nonce: &str, method: &str) -> String {
    nornir_auth_sshsig::challenge_bound::<GunnarScheme>(fingerprint, issued_at_unix_ms, nonce, method)
}

/// **The method a request is actually calling**, as `Service.Method`, put into
/// the request extensions by [`crate::trace::RequestSpanLayer`] and read by
/// [`AuthGate`].
///
/// The one fact the gate was missing. It comes from `request.uri().path()` —
/// the router's own dispatch key — and not from anything a handler passes, so a
/// handler that calls `require` with the wrong `&'static str` mislabels its
/// audit record and cannot loosen this check.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CalledMethod(pub String);

/// `/gunnar.v1.Access/AddKey` → `Access.AddKey`; the PEM armour on one header
/// line and back. The shared crate's, one spelling for every product.
pub use nornir_auth_sshsig::{armor_to_line, line_to_armor, method_name_from_path};

/// Why a credential was refused.
///
/// Deliberately coarse on the wire: [`AuthFailure::status`] tells a caller that
/// it was not authenticated and, for the fixable cases, what to fix, but never
/// whether the fingerprint is registered. That answer would turn the port into
/// a key-enumeration oracle.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AuthFailure {
    /// No credential at all.
    Missing,
    /// Present but unparseable.
    Malformed(String),
    /// Outside [`MAX_CLOCK_SKEW_MS`].
    Stale { skew_ms: i64 },
    /// This `(fingerprint, nonce)` has already been used.
    Replayed,
    /// The credential was signed for one method and presented at another.
    ///
    /// **The refusal GUN-001 asked for.** `signed` is what the caller put in
    /// the challenge; `called` is what the router was about to dispatch. Both
    /// are on the audit record because "a relay happened" and "a client has a
    /// bug" look the same from one field.
    MethodMismatch { signed: String, called: String },
    /// A method-blind [`CREDENTIAL_SCHEME_V1`] credential. Always refused.
    Unbound,
    /// [`ReplayGuard`] is full of credentials that are all still inside the
    /// freshness window.
    ///
    /// Not a denial of the caller: it is a statement about the server, and the
    /// status says so (`RESOURCE_EXHAUSTED`, not `UNAUTHENTICATED`). The
    /// alternative — forgetting the oldest still-presentable entry — is
    /// GUN-009.
    ReplayGuardFull,
    /// Signature did not verify, or the fingerprint is not registered.
    NotVerified,
    /// The identity table could not be read. **Not** a denial but a storage
    /// fault, and it must not be reported as "wrong key" (the ACL model: `may` returns
    /// `Result<bool>` for exactly this reason).
    Unavailable(String),
    /// **The credential verified against a key this box was FED to enrol and
    /// has not enrolled yet** (`Identity::awaiting_enrolment`): the front's
    /// certify key between the installer's feed and the primary's pairing loop
    /// enrolling it. Not a denial — the enrolment is in progress, and the
    /// caller asks again — so it is `UNAVAILABLE` tagged [`KEY_ENROLLING`],
    /// never `UNAUTHENTICATED` (cov-E r3 2026-10-06: the front's certify loop
    /// was told "did not verify against any registered key" for 5 minutes by a
    /// primary that was enrolling that very key, and the log gate failed the
    /// install on it).
    Enrolling,
    /// The credential verified against a key the identity table says belongs to
    /// the **anonymous** principal.
    ///
    /// That state is unreachable through `gunnar-core` — anonymous is not in
    /// the table and cannot be put there — so reaching it means the identity
    /// table has been corrupted or an adapter has invented a principal. Either
    /// way the answer is "no", loudly and with its own reason, rather than a
    /// caller who holds the public-read rule inside the control plane.
    Anonymous,
    /// The peer presented a certificate the **CA signed** — the handshake
    /// completed, so it is not a stranger — and no principal has registered it.
    ///
    /// Its own variant rather than [`AuthFailure::NotVerified`] because the two
    /// name different operator actions. `NotVerified` means *this credential is
    /// wrong*; this means *this credential is right and nobody has been given
    /// it*, which is fixed with `gunnar key add`, and an audit trail that
    /// cannot tell them apart cannot answer "did the certificate rollout
    /// finish".
    CertificateNotRegistered,
}

impl AuthFailure {
    /// The short reason that goes on an audit record.
    pub fn reason(&self) -> &'static str {
        match self {
            AuthFailure::Missing => "missing",
            AuthFailure::Malformed(_) => "malformed",
            AuthFailure::Stale { .. } => "stale",
            AuthFailure::Replayed => "replayed",
            AuthFailure::MethodMismatch { .. } => "method-mismatch",
            AuthFailure::Unbound => "unbound",
            AuthFailure::ReplayGuardFull => "replay-guard-full",
            AuthFailure::NotVerified => "not-verified",
            AuthFailure::Unavailable(_) => "unavailable",
            AuthFailure::Enrolling => "enrolling",
            AuthFailure::Anonymous => "anonymous-is-not-a-caller",
            AuthFailure::CertificateNotRegistered => "certificate-not-registered",
        }
    }

    /// The status a caller sees.
    pub fn status(&self) -> Status {
        match self {
            AuthFailure::Missing => Status::unauthenticated(format!(
                "this control plane requires a credential. Send `{CREDENTIAL_METADATA_KEY}: \
                 {CREDENTIAL_SCHEME} <fingerprint> <issued_at_ms> <nonce> <sshsig>`, signed \
                 with a key registered in gunnar's key table (the ACL model)."
            )),
            AuthFailure::Malformed(why) => {
                Status::unauthenticated(format!("malformed credential: {why}"))
            }
            AuthFailure::Stale { skew_ms } => Status::unauthenticated(format!(
                "the credential is {skew_ms} ms {CLOCK_SKEW_PHRASE}; the window is \
                 ±{MAX_CLOCK_SKEW_MS} ms"
            )),
            AuthFailure::Replayed => Status::unauthenticated(
                "this credential has already been used; each one is single-use",
            ),
            // The one refusal that names both halves. A caller seeing this has
            // either been relayed or has a signer that names the method
            // differently from the path it dials, and it cannot tell which
            // without being told both strings.
            AuthFailure::MethodMismatch { signed, called } => Status::unauthenticated(format!(
                "this credential was signed for {signed} and was presented at {called}; a \
                 credential is valid for one method only"
            )),
            // `resource_exhausted`, not `unauthenticated`. Nothing is wrong
            // with this credential; the server will not take on the obligation
            // to remember it, and a client that reads this as "my key is bad"
            // would rotate a key for no reason.
            AuthFailure::ReplayGuardFull => Status::resource_exhausted(
                "this control plane is already remembering as many single-use credentials as it \
                 will (the replay guard is full of credentials that have not yet expired). It \
                 refuses rather than forgetting one, because a forgotten credential can be \
                 replayed. Retry; if this persists, something is minting credentials far faster \
                 than a fleet does.",
            ),
            AuthFailure::Unbound => Status::unauthenticated(format!(
                "this control plane requires a method-bound credential. Send \
                 `{CREDENTIAL_METADATA_KEY}: {CREDENTIAL_SCHEME} <fingerprint> <issued_at_ms> \
                 <nonce> <method> <sshsig>` rather than the method-blind \
                 {CREDENTIAL_SCHEME_V1}, which binds no method and is refused at every one."
            )),
            // One message for "no such key" and "bad signature", on purpose.
            AuthFailure::NotVerified => {
                Status::unauthenticated("the credential did not verify against any registered key")
            }
            AuthFailure::Unavailable(why) => {
                Status::unavailable(format!("the identity table could not be read: {why}"))
            }
            // Only a caller whose signature verified against the fed key gets
            // this answer, so it tells nobody else anything.
            AuthFailure::Enrolling => Status::unavailable(format!(
                "{KEY_ENROLLING} this key was fed to this box by its installer and is being \
                 enrolled; ask again in a moment"
            )),
            // Same wording as `NotVerified`. Telling a caller *why* it was
            // singled out would confirm that the anonymous id means something
            // here, and it means nothing here on purpose.
            AuthFailure::Anonymous => {
                Status::unauthenticated("the credential did not verify against any registered key")
            }
            // Said plainly, unlike the two above. A peer that got through the
            // handshake holds a certificate this server's own CA signed, so
            // telling it that the certificate is unregistered discloses
            // nothing it could not deduce, and withholding it would send an
            // operator hunting a TLS fault that does not exist.
            AuthFailure::CertificateNotRegistered => Status::unauthenticated(
                "this client certificate is signed by the server's issuer but is not registered \
                 to any principal; register it with `gunnar key add`",
            ),
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Signing: the client half
// ───────────────────────────────────────────────────────────────────────────

/// Sign a **method-blind** ([`CREDENTIAL_SCHEME_V1`]) credential with an
/// OpenSSH private key.
///
/// **[`AuthGate`] refuses what this produces, at every method.** It is kept so
/// the refusal can be proven (and so [`ReplayGuard`], which does not care about
/// the method, can be exercised); a client signs [`sign_credential_for`].
pub fn sign_credential(
    key: &ssh_key::PrivateKey,
    issued_at_unix_ms: i64,
    nonce: &str,
) -> Result<Credential, SignError> {
    nornir_auth_sshsig::sign::<GunnarScheme>(&sshsig_key(key)?, issued_at_unix_ms, nonce, None)
}

/// [`sign_credential`], **bound to one method** (`Service.Method`, the
/// [`METHODS`] spelling). This is what `gunnar-cli` and the console sign.
pub fn sign_credential_for(key: &ssh_key::PrivateKey, issued_at_unix_ms: i64, nonce: &str, method: &str) -> Result<Credential, SignError> {
    nornir_auth_sshsig::sign::<GunnarScheme>(&sshsig_key(key)?, issued_at_unix_ms, nonce, Some(method))
}

/// What signing a credential can fail with: `nornir-auth-sshsig`'s own error.
///
/// **Two ssh-key versions meet here.** gunnar is on `ssh-key =0.7.0-rc.11` (the
/// RustCrypto bump, 2026-10-03 — the copy russh rides), and edda's
/// `nornir-auth-sshsig`, the ONE sshsig signer and verifier gunnar and monetize
/// share, is still on 0.6; so is everything that shares it (monetize's
/// control_auth, Skidbladnir's plugin-door-wire). Its error is the 0.6 crate's.
pub type SignError = nornir_auth_sshsig::ssh_key::Error;

/// gunnar's key as the type `nornir-auth-sshsig` signs with. **The bytes cross
/// the version line; the types do not** — the rule gunnar-ssh keeps for
/// fingerprints: the OpenSSH private-key encoding is the same bytes in both
/// versions (`gunnar-core/tests/crypto_kat.rs`, `gunnar-ssh/tests/host_key_kat.rs`),
/// so the key is handed over as that encoding and parsed on the other side. An
/// in-memory key always encodes; the arm exists so a failure is an error and
/// never a panic. Goes the day nornir-auth-sshsig moves to the same ssh-key.
fn sshsig_key(key: &ssh_key::PrivateKey) -> Result<nornir_auth_sshsig::ssh_key::PrivateKey, SignError> {
    let pem = key.to_openssh(ssh_key::LineEnding::LF).map_err(|_| SignError::FormatEncoding)?;
    nornir_auth_sshsig::ssh_key::PrivateKey::from_openssh(pem.as_bytes())
}

/// Verify a credential's signature against a registered OpenSSH public key
/// line — **the one verifier** (`nornir_auth_sshsig::verify_signature`): the
/// fingerprint is re-derived from the key handed in, and a signature made for
/// another namespace (`git`) is never accepted here.
///
/// ★ **A FIDO2 security key's signature counts only with a touch** (5.4 §3b):
/// an `sk-ssh-*` signature carries the authenticator's flags, which ssh-key
/// verifies but does not read — so one made with nobody present (a key
/// enrolled `no-touch-required`, malware driving a key left plugged in) is
/// refused here ([`crate::security_key::user_present`]).
pub fn verify_signature(openssh_public_key: &str, credential: &Credential) -> bool {
    nornir_auth_sshsig::verify_signature(openssh_public_key, credential) && touched(credential)
}

/// The credential's signature, if it is an sk one, was made with user presence.
fn touched(credential: &Credential) -> bool {
    let armor = nornir_auth_sshsig::line_to_armor(&credential.signature_line);
    nornir_auth_sshsig::ssh_key::SshSig::from_pem(armor.as_bytes())
        .is_ok_and(|sig| crate::security_key::user_present(sig.signature()))
}

// ───────────────────────────────────────────────────────────────────────────
// Bootstrap.Restore's own credential
// ───────────────────────────────────────────────────────────────────────────

/// The method a restore credential must be bound to.
pub const RESTORE_METHOD: &str = "Bootstrap.Restore";
/// The refusal word for a restore stream without a credential that verifies.
pub const RESTORE_UNAUTHENTICATED: &str = "restore-unauthenticated";
/// The refusal word for a box that was never fed a key to take a restore from.
pub const RESTORE_NO_KEY: &str = "restore-no-key";
/// Who owns a restore on a listener started with authentication OFF
/// ([`AuthMode::AllowAllInsecure`], reachable only from a program).
pub const RESTORE_OWNER_INSECURE: &str = "insecure-allow-all";

/// **Who is streaming this `Bootstrap.Restore`**, proved (h-v1-6a, 2026-09-28).
///
/// The restore stream is the box's whole future — seq 1 is the KEK, then the
/// identity table, the host key, the issuer CA and every file — and it arrives
/// on a service mounted outside [`AuthGate`], on the utility NIC, as plaintext
/// h2c. The gate cannot check it: it answers out of the identity table, and a
/// box being restored is fresh and has none. So this checks the same
/// `gunnar-sshsig-v2` credential the gate would, against `keys` instead — the
/// keys the box was FED before anybody could enrol (`ControlPlane::restore_keys`:
/// the installer's `admin_pubkey` and admin principals, `--restore-pubkey`):
///
/// 1. `keys` is not empty, or [`RESTORE_NO_KEY`] — FAIL CLOSED: a box nobody
///    fed takes no restore from anybody, and says so;
/// 2. the header is present and parses, else [`RESTORE_UNAUTHENTICATED`];
/// 3. it is inside [`MAX_CLOCK_SKEW_MS`];
/// 4. it is bound to [`RESTORE_METHOD`] — a method-blind v1 header, or one
///    lifted off another call, is refused;
/// 5. its signature verifies under one of `keys`;
/// 6. `(fingerprint, nonce)` is fresh in `replay` — last, so a forgery never
///    burns an honest caller's nonce.
///
/// Answers the OWNER: the key's fingerprint, which `restore_begin` records in
/// `txlog-restoring` so that only the same key resumes. The stream's CHUNKS
/// are not signed: a credential on the call is what every other RPC has, and
/// an on-path attacker on plaintext h2c could rewrite chunks under a signed
/// first one just as well.
pub fn authenticate_restore(
    mode: AuthMode,
    keys: &[String],
    replay: &ReplayGuard,
    metadata: &tonic::metadata::MetadataMap,
    now_unix_ms: i64,
) -> Result<String, Status> {
    if mode == AuthMode::AllowAllInsecure {
        return Ok(RESTORE_OWNER_INSECURE.to_owned());
    }
    if keys.is_empty() {
        return Err(Status::permission_denied(format!(
            "REFUSED [{RESTORE_NO_KEY}] {RESTORE_METHOD}: this box was never fed a key it may take a restore from, so it \
             takes none — a fresh box has no identity table, and a restore stream decides its KEK, its admin, its host \
             key and its issuer. Feed one: the installer's `admin_pubkey` / an admin `principal.<id>` (monetize's \
             identity is one), or start gunnar-server with `--restore-pubkey <openssh line>` (GUNNAR_RESTORE_PUBKEY). \
             Nothing was written; the box is still fresh"
        )));
    }
    authenticate_fed(RESTORE_METHOD, RESTORE_UNAUTHENTICATED, keys, replay, metadata, now_unix_ms)
}

/// The methods besides `Restore` that are served outside the gate and hand a
/// fresh box to whoever calls them (h-bootstrap-open, 2026-09-28): `BecomeTwin`
/// makes the box the permanent twin of the primary the caller NAMES (whose log
/// then brings its own KEK and admins), `InstallTwinCertificate` pins the FIRST
/// issuer it is handed. Each needs a credential bound to itself, from a FED key.
pub const BOOTSTRAP_BECOME_TWIN_METHOD: &str = "Bootstrap.BecomeTwin";
/// See [`BOOTSTRAP_BECOME_TWIN_METHOD`].
pub const BOOTSTRAP_INSTALL_TWIN_CERTIFICATE_METHOD: &str = "Bootstrap.InstallTwinCertificate";
/// The refusal word for a twin-making call without a credential that verifies.
pub const BOOTSTRAP_UNAUTHENTICATED: &str = "bootstrap-unauthenticated";
/// The refusal word for a box that was never fed a key to take such a call from.
pub const BOOTSTRAP_NO_KEY: &str = "bootstrap-no-key";

/// **Who is making this fresh box a twin, proved** (h-bootstrap-open) — the
/// same six steps as [`authenticate_restore`], for `method`, against the keys
/// the box was FED (`ControlPlane::restore_keys`). Answers the fingerprint.
/// The callers that make twins already hold such a key: the front's pairing
/// door (the console key, an admin principal of the feed) and
/// `gunnar control twin join -i <operator key>`. The installer's own path
/// (`pairing.rs`, in process) never crosses this door.
pub fn authenticate_bootstrap(
    mode: AuthMode,
    method: &str,
    keys: &[String],
    replay: &ReplayGuard,
    metadata: &tonic::metadata::MetadataMap,
    now_unix_ms: i64,
) -> Result<String, Status> {
    if mode == AuthMode::AllowAllInsecure {
        return Ok(RESTORE_OWNER_INSECURE.to_owned());
    }
    if keys.is_empty() {
        return Err(Status::permission_denied(format!(
            "REFUSED [{BOOTSTRAP_NO_KEY}] {method}: this box was never fed a key it may be made a twin by, so it takes \
             this from nobody — a fresh box has no identity table, and the twin it becomes follows whatever primary \
             the caller names, KEK and admins included. Feed one: the installer's `admin_pubkey` / an admin \
             `principal.<id>`, or start gunnar-server with `--restore-pubkey <openssh line>` (GUNNAR_RESTORE_PUBKEY). \
             Nothing was written; the box is as it was"
        )));
    }
    authenticate_fed(method, BOOTSTRAP_UNAUTHENTICATED, keys, replay, metadata, now_unix_ms)
}

/// Steps 2–6 of [`authenticate_restore`] for any `method` outside the gate.
fn authenticate_fed(
    method: &str,
    clause: &str,
    keys: &[String],
    replay: &ReplayGuard,
    metadata: &tonic::metadata::MetadataMap,
    now_unix_ms: i64,
) -> Result<String, Status> {
    let unauthenticated = |why: String| {
        Status::unauthenticated(format!(
            "REFUSED [{clause}] {method} needs `{CREDENTIAL_METADATA_KEY}: {CREDENTIAL_SCHEME} …` \
             bound to {method} and signed by a key this box was fed: {why}. Nothing was \
             written; the box is as it was"
        ))
    };
    let credential = Credential::from_metadata(metadata).map_err(|failure| match failure {
        AuthFailure::Missing => unauthenticated("the call carries no credential".to_owned()),
        AuthFailure::Malformed(why) => unauthenticated(format!("the credential is malformed ({why})")),
        other => unauthenticated(other.reason().to_owned()),
    })?;
    let skew = now_unix_ms - credential.issued_at_unix_ms;
    if skew.abs() > MAX_CLOCK_SKEW_MS {
        estate_clock::trigger::skew_refused(skew);
        return Err(unauthenticated(format!("the credential is {skew} ms {CLOCK_SKEW_PHRASE}")));
    }
    match credential.method.as_deref() {
        Some(m) if m == method => {}
        Some(other) => return Err(unauthenticated(format!("it was signed for {other}"))),
        None => return Err(unauthenticated(format!("it is a method-blind {CREDENTIAL_SCHEME_V1} credential"))),
    }
    if !keys.iter().any(|line| verify_signature(line, &credential)) {
        return Err(unauthenticated(format!(
            "{} is not one of the {} key(s) this box was fed, or its signature does not verify",
            credential.fingerprint,
            keys.len()
        )));
    }
    match replay.accept(&credential, now_unix_ms) {
        Accepted::Fresh => Ok(credential.fingerprint),
        Accepted::Replayed => Err(unauthenticated("this credential has already been spent".to_owned())),
        Accepted::Full => Err(AuthFailure::ReplayGuardFull.status()),
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Replay
// ───────────────────────────────────────────────────────────────────────────

/// Remembers accepted `(fingerprint, nonce)` pairs so each credential is used
/// once; a full guard REFUSES and never forgets a still-presentable entry
/// (GUN-009). The shared crate's (`nornir_auth_sshsig`).
pub use nornir_auth_sshsig::{Accepted, ReplayGuard, DEFAULT_REPLAY_CAPACITY};

// ───────────────────────────────────────────────────────────────────────────
// The unbound-credential log: who still sends v1, named
// ───────────────────────────────────────────────────────────────────────────

/// **Who is still presenting the method-blind credential, and when they last
/// did it.**
///
/// # Why this exists at all
///
/// Since 5.3.3 every method-blind credential is refused (see the module docs),
/// and a refusal nobody can see is how a security default gets switched back
/// off: a client that still signs [`CREDENTIAL_SCHEME_V1`] just stops working.
/// This names it — the fingerprint, the method it called, when, how often — so
/// an operator fixes the client instead of the server. `accepted` (served)
/// stays at zero on a 5.3.3 server; on an older one it is the reading that
/// says which clients an upgrade would cut off.
///
/// # The property that matters, and the one that is sacrificed
///
/// The roster of distinct callers is **bounded** ([`DEFAULT_UNBOUND_LOG_CAPACITY`])
/// and the totals are **not**. Past the bound a new `(fingerprint, method)`
/// pair increments [`UnboundCredentialReport::distinct_dropped`] and is not
/// listed — but [`UnboundCredentialReport::accepted`] and
/// [`UnboundCredentialReport::last_seen_unix_ms`] still move.
///
/// That asymmetry is deliberate and it is the whole reason this is safe to
/// build a decision on. The decision rests on `accepted == 0`, which a full
/// log can never falsify: an unbounded roster would be a memory hole that an
/// attacker fills by minting fingerprints, and a *bounded counter* would be a
/// flag that reads "nobody sends v1" on the one busy server where it is
/// wrong. The names are best-effort; the zero is exact.
///
/// # In-memory, and therefore about **this** process
///
/// Nothing here is persisted. A restart zeroes it, and that is the correct
/// reading rather than a gap: the question is "has a v1 credential arrived
/// since this listener came up", and an operator answers it by letting the
/// server run for longer than the longest interval any client of it calls at.
/// The durable half is the `grpc.auth.unbound` tracing event below, which goes
/// wherever the subscriber points it and survives the process.
#[derive(Debug)]
pub struct UnboundCredentialLog {
    inner: Mutex<UnboundInner>,
    capacity: usize,
}

#[derive(Debug, Default)]
struct UnboundInner {
    accepted: u64,
    refused: u64,
    last_seen_unix_ms: i64,
    distinct_dropped: u64,
    /// Keyed `fingerprint\0method`, so one client calling three methods is
    /// three rows: an operator chasing down the last v1 caller needs to know
    /// *what it does*, not only that it exists.
    seen: std::collections::BTreeMap<String, UnboundSighting>,
}

/// One `(fingerprint, method)` pair that has presented a v1 credential.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnboundSighting {
    /// The `SHA256:…` from the credential. **Not** proven to belong to anyone
    /// on a `refused` row: the refusal happens before the key lookup, so all that
    /// is known there is what the header claimed.
    pub fingerprint: String,
    /// The ACL principal, when the credential got far enough to name one.
    /// Empty on a `refused` row.
    pub principal_id: String,
    /// The method the router was about to dispatch to, as `Service.Method`.
    pub method: String,
    pub first_seen_unix_ms: i64,
    pub last_seen_unix_ms: i64,
    pub count: u64,
    /// True when the presentations on this row were refused rather than
    /// served — on a 5.3.3 server, every row: a client that has to be updated
    /// to sign v2, named.
    pub refused: bool,
}

/// A reading of [`UnboundCredentialLog`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UnboundCredentialReport {
    /// v1 credentials that **verified and named a caller**. Zero on a 5.3.3
    /// server, which serves none; on an older server, the clients an upgrade
    /// would cut off.
    pub accepted: u64,
    /// v1 credentials **refused**. Counted at presentation, before
    /// the signature is checked, so it includes junk as well as honest
    /// un-updated clients — it answers "who is hitting the closed door", not
    /// "who would have got in".
    pub refused: u64,
    /// When the last v1 credential of either kind arrived. Zero if none has.
    pub last_seen_unix_ms: i64,
    /// Distinct `(fingerprint, method)` pairs that did not fit the roster. The
    /// totals above still counted them.
    pub distinct_dropped: u64,
    /// The roster, oldest first-sighting first.
    pub callers: Vec<UnboundSighting>,
}

impl UnboundCredentialReport {
    /// **Has this listener served a method-blind credential?** Always no on a
    /// 5.3.3 server, which refuses them; read on an older one before upgrading.
    ///
    /// One writer for a question three readers ask (`Admin.Health`'s wire
    /// field, `gunnar control health`'s line, the tests), so they cannot come
    /// to three different verdicts about one report.
    ///
    /// It answers about *this process's uptime only* and says so at the call
    /// sites: a server up for ten seconds has seen nothing, which is not the
    /// same as nothing existing.
    pub fn nothing_unbound_has_been_served(&self) -> bool {
        self.accepted == 0
    }
}

/// How many distinct `(fingerprint, method)` pairs the roster holds.
///
/// Sixty-four. A fleet has operators, a console and a handful of machine accounts, and
/// the roster's job is to *name the stragglers* — a list long enough to hold
/// every client of a real deployment and short enough that filling it is
/// visible as [`UnboundCredentialReport::distinct_dropped`] rather than as
/// memory growth. The counters it sits beside are exact at any size, which is
/// what lets this number be small.
pub const DEFAULT_UNBOUND_LOG_CAPACITY: usize = 64;

/// What a sighting's `method` says when the called method was not in scope.
///
/// Reachable only from [`AuthGate::authenticate`], which has no request and
/// therefore no [`CalledMethod`]. Spelled out rather than left empty so a row
/// in `gunnar control health` reads as a fact about the call site and not as a
/// truncated string.
pub const UNKNOWN_METHOD: &str = "(no method in scope)";

impl Default for UnboundCredentialLog {
    fn default() -> Self {
        UnboundCredentialLog::new(DEFAULT_UNBOUND_LOG_CAPACITY)
    }
}

impl UnboundCredentialLog {
    pub fn new(capacity: usize) -> UnboundCredentialLog {
        UnboundCredentialLog {
            inner: Mutex::new(UnboundInner::default()),
            capacity: capacity.max(1),
        }
    }

    /// Record one presentation of a method-blind credential.
    ///
    /// Returns **true the first time** a given `(fingerprint, method)` pair is
    /// seen, so the caller can log loudly once and quietly for every repeat. A
    /// console polling `Admin.Health` every five seconds would otherwise put a
    /// `warn!` in the journal every five seconds, and an operator who has
    /// silenced that line has silenced the finding.
    pub fn record(
        &self,
        fingerprint: &str,
        principal_id: &str,
        method: &str,
        refused: bool,
        now_unix_ms: i64,
    ) -> bool {
        // A poisoned lock must not lose the sighting *and* must not panic in
        // the auth path: take the inner value and carry on, the same judgement
        // `ReplayGuard::accept` makes.
        let mut inner = match self.inner.lock() {
            Ok(inner) => inner,
            Err(poisoned) => poisoned.into_inner(),
        };
        if refused {
            inner.refused += 1;
        } else {
            inner.accepted += 1;
        }
        inner.last_seen_unix_ms = inner.last_seen_unix_ms.max(now_unix_ms);
        let key = format!("{fingerprint}\u{0}{method}");
        if let Some(row) = inner.seen.get_mut(&key) {
            row.count += 1;
            row.last_seen_unix_ms = row.last_seen_unix_ms.max(now_unix_ms);
            // A row first seen refused and later served is a row about a
            // caller that is getting in, and says so.
            row.refused = row.refused && refused;
            if row.principal_id.is_empty() {
                row.principal_id = principal_id.to_owned();
            }
            return false;
        }
        if inner.seen.len() >= self.capacity {
            inner.distinct_dropped += 1;
            // Still "new" to the caller: the loud line is the only trace this
            // pair leaves anywhere, so dropping it from the roster must not
            // also silence the log.
            return true;
        }
        inner.seen.insert(
            key,
            UnboundSighting {
                fingerprint: fingerprint.to_owned(),
                principal_id: principal_id.to_owned(),
                method: method.to_owned(),
                first_seen_unix_ms: now_unix_ms,
                last_seen_unix_ms: now_unix_ms,
                count: 1,
                refused,
            },
        );
        true
    }

    /// Read the log.
    pub fn report(&self) -> UnboundCredentialReport {
        let inner = match self.inner.lock() {
            Ok(inner) => inner,
            Err(poisoned) => poisoned.into_inner(),
        };
        let mut callers: Vec<UnboundSighting> = inner.seen.values().cloned().collect();
        callers.sort_by(|a, b| {
            a.first_seen_unix_ms
                .cmp(&b.first_seen_unix_ms)
                .then_with(|| a.fingerprint.cmp(&b.fingerprint))
                .then_with(|| a.method.cmp(&b.method))
        });
        UnboundCredentialReport {
            accepted: inner.accepted,
            refused: inner.refused,
            last_seen_unix_ms: inner.last_seen_unix_ms,
            distinct_dropped: inner.distinct_dropped,
            callers,
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The gate
// ───────────────────────────────────────────────────────────────────────────

/// The tonic interceptor that turns a credential into a [`Caller`], or refuses.
///
/// It runs **before any handler**, which is what makes "an unauthenticated call
/// never reaches the control plane" true by construction rather than by every
/// handler remembering. A handler still asks
/// [`ServiceContext::require`](crate::ServiceContext::require) for the grant:
/// two different questions, two different answers.
/// `tests/auth.rs::every_rpc_refuses_an_unauthenticated_caller` drives the
/// handlers directly, with no interceptor in the way, so a handler that forgot
/// its own check goes red.
#[derive(Clone)]
pub struct AuthGate {
    control: Arc<dyn ControlPlane>,
    mode: AuthMode,
    replay: Arc<ReplayGuard>,
    unbound: Arc<UnboundCredentialLog>,
    require_registered_certificate: bool,
}

impl AuthGate {
    pub fn new(control: Arc<dyn ControlPlane>, mode: AuthMode) -> AuthGate {
        AuthGate {
            control,
            mode,
            replay: Arc::new(ReplayGuard::default()),
            unbound: Arc::new(UnboundCredentialLog::default()),
            require_registered_certificate: false,
        }
    }

    /// Refuse a connection whose client certificate is not registered to a
    /// principal before the header is even read — see
    /// [`crate::ServerConfig::require_registered_client_certificate`].
    pub fn with_registered_certificate_required(mut self, require: bool) -> AuthGate {
        self.require_registered_certificate = require;
        self
    }

    /// Share one replay guard across every service on a listener. Two guards
    /// would let a credential be spent once per service.
    pub fn with_replay_guard(mut self, replay: Arc<ReplayGuard>) -> AuthGate {
        self.replay = replay;
        self
    }

    /// Share one [`UnboundCredentialLog`] with whatever reports it.
    ///
    /// The same reason the replay guard is shared: twelve services behind one
    /// gate is twelve clones of this struct, and twelve private logs would
    /// mean `Admin.Health` reporting only the v1 credentials that happened to
    /// arrive at `Admin` — a zero that means "not here" read as "nowhere".
    pub fn with_unbound_log(mut self, unbound: Arc<UnboundCredentialLog>) -> AuthGate {
        self.unbound = unbound;
        self
    }

    /// The log, for a caller that built the gate itself (the tests).
    pub fn unbound_log(&self) -> &Arc<UnboundCredentialLog> {
        &self.unbound
    }

    /// Record a refused method-blind credential, and say so in the log
    /// **once per distinct caller and method**.
    ///
    /// One writer for both the counter and the line, so a refusal can never
    /// be counted without being announced or announced without being counted
    /// — which is the drift that would let `Admin.Health` and the journal
    /// disagree about which client still has to be updated.
    fn note_unbound(&self, fingerprint: &str, method: &str, now_unix_ms: i64) {
        let first = self
            .unbound
            .record(fingerprint, "", method, true, now_unix_ms);
        if !first {
            // Every repeat is still counted; it is only the line that is
            // suppressed, and at `debug!` it is still reachable with
            // `RUST_LOG`.
            tracing::debug!(
                target: AUDIT_TARGET,
                event = "grpc.auth.unbound",
                fingerprint,
                method,
                refused = true,
                "a method-blind (gunnar-sshsig-v1) credential, refused again"
            );
            return;
        }
        tracing::warn!(
            target: AUDIT_TARGET,
            event = "grpc.auth.unbound",
            fingerprint,
            method,
            refused = true,
            "A METHOD-BLIND CREDENTIAL ({CREDENTIAL_SCHEME_V1}) was presented and REFUSED. It \
             binds no method, so a header lifted off this call could be re-sent at any other \
             one (GUN-001). This client has to be updated to sign {CREDENTIAL_SCHEME}. \
             `Admin.Health` counts these and names every caller."
        );
    }

    /// Verify a credential and name the caller, **with no method in scope**.
    ///
    /// **Nothing is served here.** A v2 credential cannot be checked without a
    /// method and is refused: "accept it because we cannot check" is the shape
    /// of every fail-open this review found. A v1 credential is refused
    /// everywhere. Callers that have a request want
    /// [`AuthGate::authenticate_request`], which does have the method.
    pub fn authenticate(
        &self,
        metadata: &tonic::metadata::MetadataMap,
        now_unix_ms: i64,
    ) -> Result<Caller, AuthFailure> {
        self.authenticate_bound(metadata, now_unix_ms, None)
    }

    /// [`AuthGate::authenticate`] with the method the router is about to
    /// dispatch to.
    pub fn authenticate_bound(
        &self,
        metadata: &tonic::metadata::MetadataMap,
        now_unix_ms: i64,
        called: Option<&str>,
    ) -> Result<Caller, AuthFailure> {
        if self.mode == AuthMode::AllowAllInsecure {
            return Ok(Caller::insecure());
        }
        let credential = Credential::from_metadata(metadata)?;

        let skew = now_unix_ms - credential.issued_at_unix_ms;
        if skew.abs() > MAX_CLOCK_SKEW_MS {
            // **The refusal is the evidence.** A credential outside the window
            // says this clock and its caller's disagree, and every caller of
            // this control plane is stamped by the front — so ask the front.
            // `estate_clock::trigger` holds the rules: at most one sync a
            // minute however many of these arrive (this check runs BEFORE the
            // signature, so a stranger can send them — the limit is the answer
            // to that), started on its own thread so this refusal is not
            // delayed, and carrying NOTHING from this request: `skew` is not
            // passed on, the correction comes only from the front's signed
            // answer. The window itself is unchanged; it is the detector.
            estate_clock::trigger::skew_refused(skew);
            return Err(AuthFailure::Stale { skew_ms: skew });
        }

        // ── the method binding ─────────────────────────────────────────────
        //
        // Before the key lookup and before the signature, for the same reason
        // staleness is: these are the checks that cost nothing and a request
        // that fails one of them must not reach the identity table. It is also
        // before `replay.accept`, so a relayed credential does not burn the
        // honest caller's nonce on its way to being refused.
        match (&credential.method, called) {
            (Some(signed), Some(called)) if signed != called => {
                return Err(AuthFailure::MethodMismatch {
                    signed: signed.clone(),
                    called: called.to_owned(),
                })
            }
            // A bound credential and no method to check it against. Only
            // reachable if the layer that supplies `CalledMethod` is not
            // mounted, which is a wiring fault in this crate — refuse rather
            // than serve a credential whose one restriction went unread.
            (Some(signed), None) => {
                return Err(AuthFailure::MethodMismatch {
                    signed: signed.clone(),
                    called: "(the called method was not in scope)".to_owned(),
                })
            }
            // A method-blind v1 credential, at any method: refused, always
            // (GUN-001; OWNER 2026-10-10). Recorded at the door it was turned
            // away from, because a client that still signs v1 has just stopped
            // working and this is the only place that knows which fingerprint
            // and which method — the refusal happens before the key lookup, so
            // no principal is named here and the report says so.
            (None, _) => {
                self.note_unbound(
                    &credential.fingerprint,
                    called.unwrap_or(UNKNOWN_METHOD),
                    now_unix_ms,
                );
                return Err(AuthFailure::Unbound);
            }
            (Some(_), Some(_)) => {}
        }

        let registered = self
            .control
            .registered_key(&credential.fingerprint)
            .map_err(|err| AuthFailure::Unavailable(err.to_string()))?;
        let Some(registered) = registered else {
            // A key the installer FED this box and its pairing loop has not
            // enrolled yet: `Unavailable`, ask again — but only for the holder
            // of that key. The signature is checked against the fed key, so a
            // stranger presenting its fingerprint is `NotVerified` as before.
            if let Some(fed) = self.control.awaiting_enrolment(&credential.fingerprint) {
                if verify_signature(&fed, &credential) {
                    return Err(AuthFailure::Enrolling);
                }
            }
            return Err(AuthFailure::NotVerified);
        };
        if !verify_signature(&registered.openssh, &credential) {
            return Err(AuthFailure::NotVerified);
        }

        // ── the control plane has no anonymous caller ──────────────────────
        //
        // Structurally unreachable: `gunnar_core::AuthTable` refuses to
        // register a principal under this id and refuses to attach a key to
        // one, so `registered_key` cannot name it. It is here anyway because
        // this crate takes `ControlPlane` as a TRAIT: any adapter, including a
        // future one and including a test double, can hand back whatever
        // principal id it likes, and "the SSH front end's `nobody` cannot
        // administer the server" is too important to rest on an invariant
        // enforced in a crate this one does not link. `tests/auth.rs` drives it
        // with exactly such an adapter, which is what makes it a guard that has
        // been seen red rather than a comment.
        if registered.principal_id == ANONYMOUS_PRINCIPAL_ID {
            return Err(AuthFailure::Anonymous);
        }

        // Last, so a forged signature is never given the chance to burn an
        // honest caller's nonce.
        match self.replay.accept(&credential, now_unix_ms) {
            Accepted::Fresh => {}
            Accepted::Replayed => return Err(AuthFailure::Replayed),
            Accepted::Full => return Err(AuthFailure::ReplayGuardFull),
        }

        Ok(Caller {
            principal_id: registered.principal_id,
            fingerprint: credential.fingerprint,
            machine: registered.machine,
            authenticated: true,
        })
    }

    /// [`AuthGate::authenticate`] with the **connection** in scope as well as
    /// the metadata.
    ///
    /// # The order, and why it is this way round
    ///
    /// A signed `gunnar-sshsig-v2` header wins whenever one is present, even on
    /// an mTLS connection. That is the locked resolution of the mTLS/certificate-identity
    /// tension: `sshsig` is the primary credential for git and for the control
    /// plane, and the certificate is a service credential. A connection can
    /// carry both — a fleet member with a certificate running a command on an
    /// operator's behalf — and the caller named in the audit record must be the
    /// one that *asked*, not the machine it arrived over.
    ///
    /// The certificate is consulted only when no header was sent at all. A
    /// **malformed** header is still a refusal, never a fall-through: silently
    /// downgrading to the connection's identity when a credential fails to
    /// parse is how a client bug becomes a privilege change nobody notices.
    ///
    /// # Why no replay guard on the certificate path
    ///
    /// [`ReplayGuard`] exists because a metadata credential is a bearer string
    /// once it is written down: capture it and it can be sent again. A
    /// certificate is not sent — it is *used*, inside a handshake that proves
    /// live possession of the private key, over a session that carries its own
    /// keys. There is nothing to replay, and a nonce store here would refuse
    /// the second RPC on every connection.
    pub fn authenticate_request<T>(
        &self,
        request: &Request<T>,
        now_unix_ms: i64,
    ) -> Result<Caller, AuthFailure> {
        if self.mode == AuthMode::AllowAllInsecure {
            return Ok(Caller::insecure());
        }
        // The replication door: the connection's certificate must be one a
        // principal HOLDS, not only one the issuer once signed — a superseded or
        // revoked twin certificate stops here, whatever header rides along.
        if self.require_registered_certificate {
            let Some(der) = crate::peer_certificate_der(request) else {
                return Err(AuthFailure::Missing);
            };
            self.authenticate_certificate(&der)?;
        }
        if request.metadata().get(CREDENTIAL_METADATA_KEY).is_some() {
            // The method the ROUTER will dispatch to, put here by
            // `trace::RequestSpanLayer` off `uri().path()`. Not a handler's
            // `&'static str`: see the module docs.
            let called = request
                .extensions()
                .get::<CalledMethod>()
                .map(|m| m.0.clone());
            return self.authenticate_bound(request.metadata(), now_unix_ms, called.as_deref());
        }
        let Some(der) = crate::peer_certificate_der(request) else {
            return Err(AuthFailure::Missing);
        };
        self.authenticate_certificate(&der)
    }

    /// The certificate half of [`AuthGate::authenticate_request`].
    ///
    /// Three steps where the header path has five, and the two that are gone
    /// were done by TLS: possession was proven in the handshake, and provenance
    /// by the issuer the listener pins. What is left is the same ACL table
    /// lookup the header path ends with.
    fn authenticate_certificate(&self, der: &[u8]) -> Result<Caller, AuthFailure> {
        let identity = self
            .control
            .registered_certificate(der)
            .map_err(|err| AuthFailure::Unavailable(err.to_string()))?;
        let Some(identity) = identity else {
            return Err(AuthFailure::CertificateNotRegistered);
        };
        // The anonymous principal again, for the same reason it is checked on
        // the header path: the
        // control plane has no anonymous caller, and `ControlPlane` is a trait
        // any adapter can implement.
        if identity.principal_id == ANONYMOUS_PRINCIPAL_ID {
            return Err(AuthFailure::Anonymous);
        }
        Ok(Caller {
            principal_id: identity.principal_id,
            fingerprint: identity.fingerprint,
            machine: false,
            authenticated: true,
        })
    }
}

impl tonic::service::Interceptor for AuthGate {
    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
        match self.authenticate_request(&request, crate::bus::now_unix_ms()) {
            Ok(caller) => {
                request.extensions_mut().insert(caller);
                Ok(request)
            }
            Err(failure) => {
                // The refusal record. It names no principal because there is
                // none, which is the difference a test can assert on between
                // this and a handler-level denial.
                let detail = match &failure {
                    AuthFailure::Malformed(why) => why.clone(),
                    AuthFailure::Unavailable(why) => why.clone(),
                    AuthFailure::Stale { skew_ms } => format!("{skew_ms} ms"),
                    // Both halves, because a relay and a client bug are
                    // the same line otherwise.
                    AuthFailure::MethodMismatch { signed, called } => {
                        format!("signed for {signed}, presented at {called}")
                    }
                    _ => String::new(),
                };
                // A key not installed YET (a fresh or restored box, callers that
                // retry): WARN once a minute, the rest at INFO (crate::quiet).
                // A key being enrolled (`Enrolling`): the holder of a fed key, told to
                // ask again — said at INFO, like a box growing its data set.
                let loud = if failure == AuthFailure::Enrolling {
                    None
                } else if crate::quiet::unverified(failure.reason()) {
                    crate::quiet::loud("grpc.auth.refused:not-verified")
                } else {
                    Some(0)
                };
                match loud {
                    Some(quietly) => tracing::warn!(
                        target: AUDIT_TARGET,
                        event = "grpc.auth.refused",
                        reason = failure.reason(),
                        detail = %detail,
                        quietly,
                        "gRPC call refused before reaching the control plane"
                    ),
                    None => tracing::info!(
                        target: AUDIT_TARGET,
                        event = "grpc.auth.refused",
                        reason = failure.reason(),
                        detail = %detail,
                        "gRPC call refused before reaching the control plane (said at WARN this minute already)"
                    ),
                }
                Err(failure.status())
            }
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The audit trail
// ───────────────────────────────────────────────────────────────────────────

/// **Where a decision is written down so that it can be checked later**
/// (GUN-044).
///
/// The three functions below have always emitted a `tracing` record on
/// [`AUDIT_TARGET`]. That is a log: it goes wherever the subscriber points it,
/// and nothing in it says whether a line has been removed from it since. This
/// seam is the durable half — an append-only, hash-chained file that can answer
/// "is this complete?" — and it is a TRAIT here for the same reason
/// [`ControlPlane`] is one: `gunnar-grpc` deliberately does not link
/// `gunnar-core`, and the chain's arithmetic belongs beside the rest of
/// gunnar's persistence, not inside a transport crate.
///
/// `None` — the default, a test router, a listener built without one — leaves
/// the `tracing` record exactly as it was. The seam adds a trail; it never
/// takes one away.
pub trait AuditLedger: Send + Sync + 'static {
    /// Record one decision. **Best effort by contract**: an implementation that
    /// cannot write must log and return, never panic and never fail the RPC. A
    /// control plane that stopped serving because its audit disk filled would
    /// be a denial of service with a compliance excuse.
    fn record(&self, record: AuditRecord<'_>);
}

/// One decision, in the words the `tracing` record uses.
///
/// Borrowed rather than owned, so a listener with no ledger wired pays nothing
/// but the construction of eight `&str`.
#[derive(Debug, Clone, Copy)]
pub struct AuditRecord<'a> {
    /// `Service.Method`.
    pub method: &'a str,
    /// The ACL principal id.
    pub principal: &'a str,
    /// The `SHA256:…` fingerprint of the key presented, or empty.
    pub fingerprint: &'a str,
    /// What was asked about, as [`Access`] displays it.
    pub scope: &'a str,
    /// The grant level, as [`level_name`] spells it.
    pub level: &'a str,
    /// [`AuthMode::as_str`].
    pub auth_mode: &'a str,
    /// `allow`, `deny` or `narrow`.
    pub decision: &'a str,
    /// Why, for a decision that has a why; empty otherwise.
    pub reason: &'a str,
}

impl<'a> AuditRecord<'a> {
    /// One record, from the parts the audit functions already hold.
    ///
    /// `scope` arrives rendered rather than as an [`Access`], so the
    /// `to_string` it costs happens at the call site — inside the `if let`
    /// that has already established there is a ledger to write to. A listener
    /// with none allocates nothing.
    fn of(
        method: &'a str,
        caller: &'a Caller,
        scope: &'a str,
        level: GrantLevel,
        mode: AuthMode,
        decision: &'a str,
        reason: &'a str,
    ) -> AuditRecord<'a> {
        AuditRecord {
            method,
            principal: &caller.principal_id,
            fingerprint: &caller.fingerprint,
            scope,
            level: level_name(level),
            auth_mode: mode.as_str(),
            decision,
            reason,
        }
    }
}

/// One audit record per authorised RPC. **The sharper point**: before this,
/// every gRPC service file had zero log lines, so an unauthenticated mutating
/// control plane also had no trail. A mutation must be traceable to a caller.
pub(crate) fn audit_allowed(
    method: &'static str,
    caller: &Caller,
    access: &Access,
    level: GrantLevel,
    mode: AuthMode,
    ledger: Option<&dyn AuditLedger>,
) {
    if let Some(ledger) = ledger {
        let scope = access.to_string();
        ledger.record(AuditRecord::of(
            method, caller, &scope, level, mode, "allow", "",
        ));
    }
    tracing::info!(
        target: AUDIT_TARGET,
        event = "grpc.call",
        method,
        principal = %caller.principal_id,
        fingerprint = %caller.fingerprint,
        scope = %access,
        level = level_name(level),
        auth_mode = mode.as_str(),
        decision = "allow",
        "gRPC control-plane call"
    );
}

/// **A call a principal made on ITSELF** (2026-10-04, the self-scope): its own
/// keys. Recorded with the scope `self:<principal>` and the decision `allow`,
/// and never as a server grant the caller does not hold — a trail that said
/// `scope=server level=admin` for a person adding their own laptop key would be
/// a lie an operator reads as an escalation.
pub(crate) fn audit_self(
    method: &'static str,
    caller: &Caller,
    mode: AuthMode,
    ledger: Option<&dyn AuditLedger>,
) {
    let scope = format!("self:{}", caller.principal_id);
    if let Some(ledger) = ledger {
        ledger.record(AuditRecord::of(
            method,
            caller,
            &scope,
            GrantLevel::Unspecified,
            mode,
            "allow",
            "the caller's own principal",
        ));
    }
    tracing::info!(
        target: AUDIT_TARGET,
        event = "grpc.call",
        method,
        principal = %caller.principal_id,
        fingerprint = %caller.fingerprint,
        scope = %scope,
        auth_mode = mode.as_str(),
        decision = "allow",
        reason = "the caller's own principal",
        "gRPC control-plane call on the caller's own principal"
    );
}

/// Why a listing was narrowed. Named once: the `tracing` record and the ledger
/// entry must say the same thing, or reading them side by side is misleading.
const NARROWED_REASON: &str = "no grant on the scope; narrowed to what the ACL allows row by row";

/// **The third decision**: the call was served, but on less than it asked for.
///
/// A listing by a caller with no grant on the scope is neither an allow nor a
/// deny, and recording it as either loses something an operator needs.
/// `decision="allow"` would claim a grant that does not exist; `decision="deny"`
/// would put a `warn!` in the log for every stranger who opened the front page
/// and drown the refusals that mean something. So it is `decision="narrow"`, at
/// `info!`, and "who read the public subset" stays greppable apart from "who was
/// turned away".
///
/// `level` is always `Read` today — a narrowing at any other level would be a
/// bug — but it is recorded rather than assumed, so the record does not have to
/// be re-read if that ever stops being true.
pub(crate) fn audit_narrowed(
    method: &'static str,
    caller: &Caller,
    access: &Access,
    level: GrantLevel,
    mode: AuthMode,
    ledger: Option<&dyn AuditLedger>,
) {
    if let Some(ledger) = ledger {
        let scope = access.to_string();
        ledger.record(AuditRecord::of(
            method,
            caller,
            &scope,
            level,
            mode,
            "narrow",
            NARROWED_REASON,
        ));
    }
    tracing::info!(
        target: AUDIT_TARGET,
        event = "grpc.call",
        method,
        principal = %caller.principal_id,
        fingerprint = %caller.fingerprint,
        scope = %access,
        level = level_name(level),
        auth_mode = mode.as_str(),
        decision = "narrow",
        reason = NARROWED_REASON,
        "gRPC control-plane listing narrowed rather than refused"
    );
}

/// The same record for a refusal, at `warn!`. A denied admin call is the line
/// an operator greps for.
pub(crate) fn audit_denied(
    method: &'static str,
    caller: &Caller,
    access: &Access,
    level: GrantLevel,
    mode: AuthMode,
    why: &str,
    ledger: Option<&dyn AuditLedger>,
) {
    if let Some(ledger) = ledger {
        let scope = access.to_string();
        ledger.record(AuditRecord::of(
            method, caller, &scope, level, mode, "deny", why,
        ));
    }
    tracing::warn!(
        target: AUDIT_TARGET,
        event = "grpc.call",
        method,
        principal = %caller.principal_id,
        fingerprint = %caller.fingerprint,
        scope = %access,
        level = level_name(level),
        auth_mode = mode.as_str(),
        decision = "deny",
        reason = why,
        "gRPC control-plane call refused"
    );
}

/// **An authorised call that also OVERRODE a guard** (2026-09-22, the retention guard's
/// `allow_past_twin`): a second record beside the call's own `allow`, with the words
/// `OVERRIDE:` and why, so the trail names who used the override and what it overrode — a
/// plain `allow` for `Mirror.LogRetain` would read the same whether or not a twin lost frames.
pub(crate) fn audit_override(
    method: &'static str,
    caller: &Caller,
    access: &Access,
    level: GrantLevel,
    mode: AuthMode,
    why: &str,
    ledger: Option<&dyn AuditLedger>,
) {
    let reason = format!("OVERRIDE: {why}");
    if let Some(ledger) = ledger {
        let scope = access.to_string();
        ledger.record(AuditRecord::of(method, caller, &scope, level, mode, "allow", &reason));
    }
    tracing::warn!(
        target: AUDIT_TARGET,
        event = "grpc.call",
        method,
        principal = %caller.principal_id,
        fingerprint = %caller.fingerprint,
        scope = %access,
        level = level_name(level),
        auth_mode = mode.as_str(),
        decision = "allow",
        reason = %reason,
        "gRPC control-plane call OVERRODE a guard"
    );
}

/// The verbs that open a READ: a call whose method starts with one of these
/// changes nothing ([`changes_state`]).
const READ_VERBS: &[&str] = &[
    "List", "Get", "Read", "Check", "Who", "Watch", "Verify", "Search", "Export", "Discover", "Diff",
    "Blame",
];

/// The calls whose verb does not say "read" and that change nothing a person
/// did: status and report reads, the twin's log reads, and the sign-in
/// ceremony itself (a sign-in is its own stamp, not an operation).
const NOT_OPERATIONS: &[&str] = &[
    "Access.WebChallenge",
    "Access.WebLogin",
    "Access.MachineEnrolmentStatus",
    "Accounts.PurgeStatus",
    "Admin.AuditTrail",
    "Admin.Capacity",
    "Admin.GrowStatus",
    "Admin.Health",
    "Admin.IssuerInfo",
    "Admin.Stats",
    "Admin.StorageInfo",
    "Admin.Topology",
    "Enterprise.Aggregates",
    "Enterprise.AuditLog",
    "Mirror.LogFetch",
    "Mirror.LogHead",
    "Mirror.ReplicationStatus",
    "Mirror.SnapshotFetch",
    "Mirror.Status",
    "Refs.Log",
    "Repos.ArchiveOut",
    "Repos.GcStatus",
    "Repos.RecipientKeys",
    "Sso.Begin",
    "Sso.Finish",
    "Sso.SecondFactor",
    "Usage.Activity",
    "Usage.AdmitPlan",
];

/// [`NOT_OPERATIONS`] of the metered build: two reads.
#[cfg(feature = "metered")]
const METERED_NOT_OPERATIONS: &[&str] = &["Billing.ApplianceIdentity", "Billing.MintActorTicket"];

/// **Does `method` (`Service.Method`) change something?** (2026-10-21) — what
/// "last operation" in the console's user list counts: a repository made or
/// removed, a setting, a key, a grant, an invite… and not a read. Unknown
/// methods count as changes, so a new mutator is never silently a read.
pub fn changes_state(method: &str) -> bool {
    if NOT_OPERATIONS.contains(&method) {
        return false;
    }
    #[cfg(feature = "metered")]
    if METERED_NOT_OPERATIONS.contains(&method) {
        return false;
    }
    let verb = method.rsplit_once('.').map_or(method, |(_, verb)| verb);
    !READ_VERBS.iter().any(|read| verb.starts_with(read))
}

/// Every method name that appears in an audit record, so the table-driven
/// deny test and the handlers cannot drift apart.
///
/// Kept as a set rather than a list of literals scattered through seven files:
/// `tests/auth.rs` asserts this covers exactly the RPCs the services expose.
pub fn audited_methods() -> BTreeSet<&'static str> {
    #[allow(unused_mut)]
    let mut all: BTreeSet<&'static str> = METHODS.iter().copied().collect();
    #[cfg(feature = "metered")]
    all.extend(METERED_METHODS.iter().copied());
    all
}

/// **The RPCs gunnar serves with NO credential. There are two, and there is a
/// reason for each.**
///
/// Both belong to `Bootstrap` (see [`crate::service::bootstrap`]). Neither can
/// be behind [`AuthGate`], because the gate answers out of the ACL key table and
/// on a server that has never been enrolled that table is empty: every
/// credential fails, including a correct one, so a gated `Bootstrap` would make
/// the first boot unreachable.
///
/// This list is the thing that keeps that from spreading. `tests/auth.rs`
/// asserts that [`METHODS`] and this are **disjoint** and that every method in
/// [`METHODS`] is refused without a credential, so adding a third
/// unauthenticated RPC means writing its name here, in the file that documents
/// why there are only two.
///
/// `Bootstrap.Enroll` is unauthenticated but it is **not** unguarded: it is
/// refused by `gunnar_core::AuthTable::install_first_identity` the moment
/// anything has been enrolled, and that refusal is inside the same
/// compare-and-swap that does the enrolling.
/// `Bootstrap.GrowStatus` (2026-09-14) is the third, and T6 of
/// `DATA-SET-GROWTH-FLOW.md` is its reason: while `/data` is unmounted for the
/// runtime data-set growth the identity table is served from RAM, and after a
/// crash in that window it is not served at all — so the actor OUTSIDE the box
/// that has to decide whether to re-attach the volumes must read the growth's
/// phase with nothing in hand. It reveals a phase word, the member disks'
/// names and sizes and the journal, and nothing else.
/// `Bootstrap.AssignTunnelAddress` (2026-09-21, T17) hands a fresh box its
/// WireGuard tunnel address, which used to be baked into the medium. It WRITES
/// only while the bootstrap window is open or to repeat the address held, and
/// otherwise reveals the box's WireGuard PUBLIC key and its tunnel address.
///
/// `Bootstrap.Restore` is on this list because it is served OUTSIDE the gate,
/// not because it takes no credential: since 2026-09-28 (h-v1-6a) it checks
/// its own, [`authenticate_restore`], against the keys the box was FED rather
/// than the identity table a fresh box does not have — and its resume answers
/// only the key that started it. Until then its one guard was "is the box
/// fresh?", and whoever reached :50051 first decided the KEK, the admin, the
/// host key and the issuer of a re-imaged box.
pub const UNAUTHENTICATED_METHODS: &[&str] =
    &[
        "Bootstrap.Enroll",
        "Bootstrap.Status",
        "Bootstrap.GrowStatus",
        "Bootstrap.BecomeTwin",
        "Bootstrap.Restore",
        "Bootstrap.AssignTunnelAddress",
        // The twin's two pairing doors (lane T9). Outside the gate for the reason
        // `BecomeTwin` is — a fresh box has no identity table to check against —
        // and guarded in the seam (refused unless the box IS a twin, the
        // certificate is for its own key, and once pinned the issuer is the same
        // one). `BecomeTwin` and `InstallTwinCertificate` ALSO take a credential
        // from a FED key (`authenticate_bootstrap`, h-bootstrap-open 2026-09-28);
        // `TwinCsr`'s answer is public data.
        "Bootstrap.TwinCsr",
        "Bootstrap.InstallTwinCertificate",
    ];

/// **The RPCs a fully-featured gunnar still answers `Unimplemented` for.**
/// There is ONE, and it is an architectural constraint rather than an
/// oversight.
///
/// `Mirror.Sync` fetches, and fetching runs over `gunnar-client` (nornir-git
/// ls-remote + fetch), which an S crate must never link (the dependency budget). The other three
/// `Mirror.*` verbs are a document write, a document read and a document
/// delete, and they are served; so are all four replication verbs on the same
/// service. The split is real and `MIRROR_GAP` in
/// `gunnar-server/src/control_plane.rs` is scoped to the verb that has it.
///
/// # Why this is a constant and not a paragraph
///
/// Because a refusal outlives its reason silently. `PROTECTION_GAP` sat beside
/// `MIRROR_GAP` saying all six `Policy.*` verbs were unimplemented, long after
/// `gunnar_core::PolicyStore` gave the policy document a home at
/// `refs/gunnar/policy` and every one of them started being served. `cargo
/// check` called the constant `never used`; nothing else noticed, and
/// `gunnar-ui/docs/coverage.md` went on quoting it to render six console panels
/// as disabled — for weeks, while a direct push to a "protected" branch was
/// refused by nothing at all.
///
/// So the refusal set is a value with two readers instead of prose with none:
///
/// * `gunnar-cli/tests/policy_mirror.rs::the_only_refusal_left_on_these_two_services`
///   drives EVERY `Policy.*` and `Mirror.*` verb against a real server and
///   asserts the set that actually answers `Unimplemented` is exactly this one
///   — so a refusal that is added, or one that quietly stops refusing, is named
///   rather than inferred;
/// * `gunnar-ui/crates/gunnar-ui-server/tests/contract.rs` reads this list out
///   of this file and asserts the `refused by gunnar` rows of the console's
///   coverage matrix are exactly it — so the doc cannot go stale again without
///   a test saying which row lies.
///
/// The census is exhaustive over `Policy`, `Mirror` and `Refs.Diff` — every
/// verb the console's matrix has ever claimed is refused. `Refs.Diff` is in it
/// for the same reason the others are: that row said `refused by gunnar,
/// replaced` while `CoreControlPlane::diff` had no refusal in it at all. If
/// another service ever grows a gap, its name goes here and the census test
/// above grows the call that proves it.
///
/// **Empty since 2026-09-08.** `Mirror.Sync` was the last entry — the one
/// verb that FETCHED, which the server crate could not do without linking a
/// git client the appliance may not carry (the dependency budget, +128 packages). `fetch_pack`
/// landed in `gunnar-wire` on gunnar's own pktline, 46 crates → 46, no gix;
/// `Mirror.Sync` now enqueues a job on the one runner and answers with it.
/// The list stays, empty, so the census test still runs every verb and would
/// name the day a refusal quietly comes back.
pub const UNIMPLEMENTED_METHODS: &[&str] = &[];

/// The authenticated RPCs of the gRPC control plane. Any new one must be added
/// here *and* given an `authorize`/`require` call; the test that drives every
/// method with no credential is what makes forgetting the second half go red.
///
/// [`UNAUTHENTICATED_METHODS`] is deliberately **not** in here.
///
/// # The count is not written down here any more, and that is the fix
///
/// This doc said "the 37 RPCs" while the list held 71 and the protos declared
/// 89. A hand-kept census drifts silently in exactly one direction — an RPC is
/// added and nothing here notices — and twelve had: the four `Access.*Group*`
/// verbs, `Forge.GetIssue`/`ListIssues`, `Mirror.SetCredential`/
/// `ResyncReplication`, `Usage.Activity` and all three `Webhooks.*`. Every one
/// of them DID authorise in its handler, checked one by one, so this was a
/// coverage hole and not an open door — but the sweep in
/// `tests/auth.rs` never drove them, so nothing would have noticed if one
/// stopped. That is precisely the hole the `PullRequests.*` note below records
/// being found in, a year and five RPCs earlier.
///
/// `tests/census.rs` now derives the expected set from the `.proto` files
/// themselves and compares it against this list, so the next omission is a
/// failing test rather than a number nobody re-counted.
pub const METHODS: &[&str] = &[
    // The four GROUP verbs (GUN-023). They authorise in the handler and always
    // did; what they were missing from was this list, so the deny-by-default
    // sweep never drove them and nothing would have noticed if one stopped.
    "Access.AddGroupMember",
    "Access.AddKey",
    "Access.AddPrincipal",
    // **Who am I, and may I?** (2026-10-04, the thin layer): the CALLER's own
    // standing, no grant beyond being authenticated — and never anonymously.
    "Access.Check",
    "Access.WhoAmI",
    // **Web credentials** (Stage L, 2026-10-04): a person's passkeys. A server
    // admin, or the principal itself (the self-scope).
    "Access.AddCredential",
    "Access.ListCredentials",
    "Access.RevokeCredential",
    "Access.RenameCredential",
    // **Invites** (Stage L): a pending passkey enrolment on an existing
    // principal — a server admin's, an account admin's for the account's
    // employees, a person's for themselves.
    "Access.CreateInvite",
    "Access.ListInvites",
    "Access.RevokeInvite",
    // **Is this invite still good?** (2026-10-19) Read-only, never spends it;
    // server Admin (the relay's own credential), as WebRegister.
    "Access.CheckInvite",
    // **A web login verified by gunnar** (Stage L): the console relays the
    // browser's WebAuthn answer; server Admin (the relay's own credential).
    "Access.WebChallenge",
    "Access.WebLogin",
    "Access.WebRegister",
    // **A machine asks, a person approves** (2026-10-05): the ops kit's web
    // credential. Asking and polling are the relay's (server Admin);
    // listing, approving and refusing a server admin's — and approving also a
    // PERSON's fresh step-up (or the bootstrap door's token), in the handler.
    "Access.RequestMachineEnrolment",
    "Access.MachineEnrolmentStatus",
    "Access.ListMachineEnrolments",
    "Access.ApproveMachineEnrolment",
    "Access.RefuseMachineEnrolment",
    "Access.Grant",
    "Access.ListGroupMembers",
    "Access.ListGroupMembershipEvents",
    "Access.ListGrants",
    "Access.ListGroupMembers",
    "Access.ListGroupMembershipEvents",
    "Access.ListKeys",
    "Access.ListPrincipals",
    "Access.RemoveGroupMember",
    "Access.RemovePrincipal",
    // A rename is what an operator reads in every listing afterwards; it takes
    // the gate AddPrincipal takes (2026-09-12).
    "Access.RenamePrincipal",
    "Access.Revoke",
    "Access.RevokeKey",
    // The per-USER verbs (2026-09-07). Every one is Admin at SERVER scope; see
    // `service::accounts` for why a namespace check would be self-granting.
    "Accounts.Gate",
    "Accounts.ListAccountGates",
    "Accounts.Purge",
    "Accounts.PurgeStatus",
    "Accounts.Ungate",
    // The account admin's own verbs (2026-10-03): enroll a NEW user as an
    // employee, and list them. Admin over the account's NAMESPACE, not the
    // server's: these are the customer's decisions.
    // **Company SSO** (2026-10-05). The provider, the SCIM bearers, the
    // directory and the CSV import: the ACCOUNT ADMIN's (Admin over the
    // account's namespace). Discover/Begin/Finish/SecondFactor/Scim: the
    // RELAY's (server Admin, the console's own credential), as WebLogin.
    "Sso.SetProvider",
    "Sso.GetProvider",
    "Sso.RemoveProvider",
    "Sso.CreateScimToken",
    "Sso.ListScimTokens",
    "Sso.RevokeScimToken",
    "Sso.ListDirectory",
    "Sso.SetUserActive",
    "Sso.ImportUsers",
    "Sso.Discover",
    "Sso.Begin",
    "Sso.Finish",
    "Sso.SecondFactor",
    "Sso.Scim",
    "Accounts.EnrollEmployee",
    "Accounts.ListEmployees",
    // The estate's operator (2026-10-04): server Admin — it makes the account the
    // first admin's passkey binds to, and adopts an ownerless namespace for it.
    "Accounts.ClaimOperator",
    "Admin.AuditTrail",
    "Admin.Capacity",
    // **The owner's enterprise list** (2026-10-05). The audit log, the
    // operator's look decided and the review queue: an ADMIN OF THE ACCOUNT
    // (never a server grant). Aggregates and the lists: server Read; asking
    // for a look and setting the lists: server Admin. CheckName: any
    // authenticated caller.
    "Enterprise.AuditLog",
    "Enterprise.ExportAuditLog",
    "Enterprise.Aggregates",
    "Enterprise.RequestOperatorAccess",
    "Enterprise.ListOperatorAccess",
    "Enterprise.ApproveOperatorAccess",
    "Enterprise.RefuseOperatorAccess",
    "Enterprise.RevokeOperatorAccess",
    "Enterprise.CheckName",
    "Enterprise.GetNameLists",
    "Enterprise.SetNameLists",
    "Enterprise.ListNameReviews",
    "Enterprise.DecideNameReview",
    "Admin.GrowResume",
    "Admin.GrowStart",
    "Admin.GrowStatus",
    "Admin.Health",
    // **"Which CA does this box mint with?"** (2026-09-17, lane PKI-READ).
    // Server `Read`, the bar its neighbours take; the reasoning, and the two
    // levels that were rejected, are on the handler in `service::admin`. It is
    // in this list from the hour it was declared, because the three omissions
    // recorded in this file — the four `Access.*Group*` verbs, `PullRequests.*`
    // and `Mirror.LogRetain` — were each an RPC that authorised correctly and
    // that the deny-by-default sweep had never once driven.
    "Admin.IssuerInfo",
    // The tunnel address of a box past its bootstrap (2026-09-21, T17):
    // server-wide Admin — it decides what the hub routes to this box.
    "Admin.SetTunnelAddress",
    // A twin's replication certificate (lane T9): server Admin, in this list
    // from the hour it was declared.
    "Admin.IssueTwinCertificate",
    "Admin.Stats",
    "Admin.StorageInfo",
    "Admin.Topology",
    "Admin.VerifyChain",
    // **Clans** (2026-10-05): the person's OWN verbs (the self-scope); gunnar
    // decides from the clan who its admin is — a server grant is not that.
    "Clans.Accept",
    // Round 2: a repository moved out of a clan — offered by the clan's
    // admin, answered by the offered account's (or withdrawn).
    "Clans.AnswerRepoOffer",
    "Clans.Create",
    "Clans.Decline",
    "Clans.Dissolve",
    "Clans.Get",
    "Clans.Invite",
    "Clans.Leave",
    "Clans.List",
    "Clans.OfferRepo",
    "Clans.RemoveMember",
    "Clans.SetAdmin",
    "Clans.SetContribution",
    "Events.Watch",
    "Forge.Append",
    "Forge.GetIssue",
    "Forge.GetReview",
    "Forge.ListIssues",
    "Forge.ListReviews",
    "Mirror.Add",
    "Mirror.Remove",
    "Mirror.Status",
    "Mirror.AbdicateReplication",
    "Mirror.DemoteReplication",
    "Mirror.FlushReplication",
    "Mirror.RemoveReplication",
    "Mirror.ReplicationStatus",
    "Mirror.ResyncReplication",
    // The transaction log the twin fetches (2026-09-15): head is a server
    // Read, the range is server Admin — it carries every sealed secret.
    "Mirror.LogHead",
    "Mirror.LogFetch",
    "Mirror.SnapshotFetch",
    // **The verb that MOVES the twin's retention** (added here 2026-09-16, lane AUDIT-TWIN).
    // It has always authorised in its handler — `Access::Server` + `Admin` — and what it was
    // missing from was this list, exactly as the group verbs and `PullRequests.*` were before
    // it; this is the third time. `no_rpc_is_reachable_without_a_caller_and_none_is_anonymous_scoped`
    // was RED on master naming it: declared in the protos, in neither this list nor
    // `UNAUTHENTICATED_METHODS`, so nothing checked that it refuses. What it would cost: a
    // caller who could move retention forward puts EVERY twin of this box into
    // `snapshot-needed` at will — the hole a twin may not cross is what this verb makes.
    "Mirror.LogRetain",
    "Mirror.SetCredential",
    "Mirror.SetReplication",
    "Mirror.SetRole",
    "Mirror.Sync",
    "Policy.DeleteProtection",
    "Policy.GetProtection",
    "Policy.GetRefNamePolicy",
    "Policy.ListGates",
    "Policy.SetProtection",
    "Policy.SetRefNamePolicy",
    // `PullRequests.*` audit under these names in `service::pulls` and were
    // MISSING from this list until the two designs were merged (2026-08-30).
    // Five RPCs that check a grant and were not in the census, so the
    // deny-by-default sweep in `tests/auth.rs` never drove them — which is the
    // exact hole this constant exists to close.
    "PullRequests.Close",
    "PullRequests.Create",
    "PullRequests.Get",
    "PullRequests.List",
    "PullRequests.Merge",
    "Refs.Blame",
    "Refs.Diff",
    "Refs.Get",
    "Refs.List",
    "Refs.Log",
    "Refs.ReadLfsObject",
    "Refs.ReadObject",
    "Refs.Search",
    "Repos.ArchiveOut",
    "Repos.Create",
    "Repos.CreateNamespace",
    "Repos.Delete",
    "Repos.DeleteNamespace",
    "Repos.Gc",
    "Repos.GcStatus",
    "Repos.Get",
    "Repos.List",
    "Repos.ListDeleted",
    "Repos.ListNamespaces",
    "Repos.ListOrphans",
    "Repos.Purge",
    // A member's SSH public keys, for an encrypted repository's key ring (2026-10-05).
    "Repos.RecipientKeys",
    "Repos.ReclaimOrphan",
    "Repos.SetDefaultBranch",
    // Encrypted repositories: the account admin turns the flag on/off (2026-10-05).
    "Repos.SetEncrypted",
    // The account admin's settings: namespace public, events public (2026-10-03).
    "Repos.SetNamespaceSettings",
    "Repos.SetVisibility",
    "Repos.Undelete",
    "Usage.Activity",
    "Usage.AdmitPlan",
    "Usage.Read",
    "Usage.ReadAll",
    // Many namespaces' meters in one call, only what changed (2026-10-06).
    "UsageBatch.ReadMany",
    "Webhooks.AddWebhook",
    "Webhooks.ListWebhooks",
    "Webhooks.RemoveWebhook",
];

/// **The metered appliance's RPCs**, audited like the rest — and, like the
/// rest, refused without a credential. A separate list because the service
/// is a separate compilation: the community build has no `Entitlement` and
/// therefore no entry here to sweep. `audited_methods` unions the two.
///
/// `Billing.*` (2026-09-17) is the second family in here. It is the tenant's
/// own renewal path: `MintActorTicket` needs `Admin` on the ONE namespace it
/// names, and `ApplianceIdentity` needs server `Admin` because it is operator
/// configuration. Both are listed here rather than being remembered, because
/// an RPC missing from this table is an RPC the deny-by-default sweep in
/// `tests/auth.rs` has never once driven — which is the hole `Mirror.LogRetain`
/// and the four `Access.*Group*` verbs were each found in, and the reason
/// `tests/census.rs` now derives the expected set from the protos.
#[cfg(feature = "metered")]
pub const METERED_METHODS: &[&str] = &[
    "Billing.ApplianceIdentity",
    "Billing.MintActorTicket",
    "Entitlement.Get",
    "Entitlement.List",
    "Entitlement.Set",
    "Sealed.Get",
    "Sealed.Put",
];

// ───────────────────────────────────────────────────────────────────────────
// The OTHER direction: a primary proving itself to its twin
// ───────────────────────────────────────────────────────────────────────────

/// **The `sshsig` namespace a primary signs its answers to a twin under.**
///
/// Deliberately NOT [`SSHSIG_NAMESPACE`]. The credential a twin sends and the
/// answer a primary sends back are signed by different keys for different
/// purposes in opposite directions, and `sshsig` verification is namespace-
/// bound, so separating them means a captured credential can never be presented
/// as a primary's answer and a captured answer can never be spent as a
/// credential. One constant, one direction.
pub const TWIN_ANSWER_NAMESPACE: &str = "gunnar-twin";

/// The version tag on the signed text, so the shape can change without an old
/// signature quietly satisfying a new reader.
const TWIN_ANSWER_V1: &str = "gunnar-log-head-v1";

/// The tag on an answer that also names the log's generation (stinky row 18).
/// A different tag, not an extra line on v1: a v1 signature can never be read
/// as a v2 one with the generation stripped, or the other way round.
const TWIN_ANSWER_V2: &str = "gunnar-log-head-v2";

/// **What a primary signs when a twin asks it for the head.**
///
/// Every field a twin acts on is in here, and the nonce the twin just minted is
/// first: the freshness, the identity and the numbers stand or fall together.
/// Signing only the nonce would leave the numbers forgeable; signing only the
/// numbers would let a recorded answer be replayed for ever at a twin whose
/// primary had moved on.
pub fn twin_answer_challenge(nonce: &str, log_id: &str, first_seq: u64, head_seq: u64) -> String {
    twin_answer_challenge_at(nonce, log_id, first_seq, head_seq, 0)
}

/// [`twin_answer_challenge`] of an answer that also names the log's
/// GENERATION (`LogHeadResponse.generation`). Generation 0 — a primary that has
/// never been restored, or one too old to know of generations — is the exact v1
/// text, so every older primary's answer still verifies as generation 0.
pub fn twin_answer_challenge_at(nonce: &str, log_id: &str, first_seq: u64, head_seq: u64, generation: u64) -> String {
    if generation == 0 {
        format!("{TWIN_ANSWER_V1}\n{nonce}\n{log_id}\n{first_seq}\n{head_seq}\n")
    } else {
        format!("{TWIN_ANSWER_V2}\n{nonce}\n{log_id}\n{first_seq}\n{head_seq}\n{generation}\n")
    }
}

/// **Sign a `Mirror.LogHead` answer** with this box's key. The armour comes back
/// in the one-line form ([`armor_to_line`]) the proto field carries.
pub fn sign_twin_answer(
    key: &ssh_key::PrivateKey,
    nonce: &str,
    log_id: &str,
    first_seq: u64,
    head_seq: u64,
) -> Result<String, ssh_key::Error> {
    sign_twin_answer_at(key, nonce, log_id, first_seq, head_seq, 0)
}

/// [`sign_twin_answer`] of an answer that also names the log's GENERATION
/// (`LogHeadResponse.generation`, stinky row 18).
pub fn sign_twin_answer_at(
    key: &ssh_key::PrivateKey,
    nonce: &str,
    log_id: &str,
    first_seq: u64,
    head_seq: u64,
    generation: u64,
) -> Result<String, ssh_key::Error> {
    let message = twin_answer_challenge_at(nonce, log_id, first_seq, head_seq, generation);
    let signature = key.sign(TWIN_ANSWER_NAMESPACE, ssh_key::HashAlg::Sha512, message.as_bytes())?;
    Ok(armor_to_line(&signature.to_pem(ssh_key::LineEnding::LF)?))
}

/// **Did the box this twin agreed to follow sign this answer?**
///
/// `pinned_public_key` is the `authorized_keys` line the twin recorded on its
/// first poll. The namespace is checked here as well as inside `verify`, for the
/// same reason it is on the credential path: a signature made for anything else
/// — a commit, a credential — must never be accepted as a primary's answer, and
/// that is important enough to be visible at the call site.
pub fn verify_twin_answer(
    pinned_public_key: &str,
    signature_line: &str,
    nonce: &str,
    log_id: &str,
    first_seq: u64,
    head_seq: u64,
) -> bool {
    verify_twin_answer_at(pinned_public_key, signature_line, nonce, log_id, first_seq, head_seq, 0)
}

/// [`verify_twin_answer`] of an answer that also names the log's generation.
pub fn verify_twin_answer_at(
    pinned_public_key: &str,
    signature_line: &str,
    nonce: &str,
    log_id: &str,
    first_seq: u64,
    head_seq: u64,
    generation: u64,
) -> bool {
    let Ok(key) = ssh_key::PublicKey::from_openssh(pinned_public_key) else {
        return false;
    };
    let armor = line_to_armor(signature_line);
    let Ok(signature) = ssh_key::SshSig::from_pem(armor.as_bytes()) else {
        return false;
    };
    if signature.namespace() != TWIN_ANSWER_NAMESPACE {
        return false;
    }
    let message = twin_answer_challenge_at(nonce, log_id, first_seq, head_seq, generation);
    key.verify(TWIN_ANSWER_NAMESPACE, message.as_bytes(), &signature).is_ok()
}

#[cfg(test)]
mod twin_answer_tests {
    use super::*;

    fn key() -> ssh_key::PrivateKey {
        ssh_key::PrivateKey::random(&mut rand::rand_core::UnwrapErr(rand::rngs::SysRng), ssh_key::Algorithm::Ed25519).unwrap()
    }

    fn pub_line(k: &ssh_key::PrivateKey) -> String {
        k.public_key().to_openssh().unwrap()
    }

    /// ★ **The answer a twin acts on is signed, fresh, and bound to every field
    /// it acts on.**
    ///
    /// Neutralise by making `verify_twin_answer` return `true` unconditionally:
    /// every case below but the first stops refusing.
    #[test]
    fn a_twin_answer_verifies_only_for_the_pinned_key_this_nonce_and_these_numbers() {
        let primary = key();
        let pinned = pub_line(&primary);
        let sig = sign_twin_answer(&primary, "nonce-1", "abc123", 1, 42).unwrap();

        assert!(verify_twin_answer(&pinned, &sig, "nonce-1", "abc123", 1, 42), "the real answer verifies");

        // A different box: the whole point.
        assert!(!verify_twin_answer(&pub_line(&key()), &sig, "nonce-1", "abc123", 1, 42), "another box's key must not verify");
        // A replayed answer, at the next poll.
        assert!(!verify_twin_answer(&pinned, &sig, "nonce-2", "abc123", 1, 42), "an answer for another poll is not this poll's");
        // The numbers and the identity are inside the signature, not beside it.
        assert!(!verify_twin_answer(&pinned, &sig, "nonce-1", "def456", 1, 42), "the log identity is signed");
        assert!(!verify_twin_answer(&pinned, &sig, "nonce-1", "abc123", 2, 42), "first_seq is signed");
        assert!(!verify_twin_answer(&pinned, &sig, "nonce-1", "abc123", 1, 43), "head_seq is signed");
        // Nothing at all.
        assert!(!verify_twin_answer(&pinned, "", "nonce-1", "abc123", 1, 42), "an unsigned answer is not signed");
    }

    /// ★ **A credential can never be spent as an answer, or the reverse.**
    ///
    /// The two directions are signed by different keys for different purposes;
    /// `sshsig` is namespace-bound and this is what that buys. Neutralise by
    /// setting `TWIN_ANSWER_NAMESPACE` to `SSHSIG_NAMESPACE`: the first
    /// assertion goes green, which is the hole.
    #[test]
    fn the_twin_namespace_and_the_credential_namespace_do_not_cross() {
        assert_ne!(TWIN_ANSWER_NAMESPACE, SSHSIG_NAMESPACE);
        let primary = key();
        let pinned = pub_line(&primary);
        let message = twin_answer_challenge("n", "id", 1, 2);
        // The same bytes, signed under the CREDENTIAL namespace, offered as an answer.
        let wrong = primary.sign(SSHSIG_NAMESPACE, ssh_key::HashAlg::Sha512, message.as_bytes()).unwrap();
        let line = armor_to_line(&wrong.to_pem(ssh_key::LineEnding::LF).unwrap());
        assert!(!verify_twin_answer(&pinned, &line, "n", "id", 1, 2), "a signature from the credential namespace is not an answer");
        // …and the real answer is not a credential.
        let answer = sign_twin_answer(&primary, "n", "id", 1, 2).unwrap();
        let armor = line_to_armor(&answer);
        let sig = ssh_key::SshSig::from_pem(armor.as_bytes()).unwrap();
        assert_ne!(sig.namespace(), SSHSIG_NAMESPACE);
    }
}

#[cfg(test)]
mod changes_state_tests {
    /// What the console reads as a person's "last operation" (2026-10-21): a
    /// change, never a read and never the sign-in ceremony itself.
    #[test]
    fn a_read_is_not_an_operation_and_a_change_is() {
        for read in ["Access.ListPrincipals", "Access.WhoAmI", "Repos.List", "Refs.Log", "Admin.Health", "Access.WebLogin", "Sso.Finish"] {
            assert!(!super::changes_state(read), "{read} is not an operation");
        }
        for write in ["Access.AddPrincipal", "Access.AddKey", "Repos.Create", "Repos.Delete", "Repos.SetVisibility", "Policy.SetProtection", "Mirror.LogRetain", "Sso.Scim"] {
            assert!(super::changes_state(write), "{write} is an operation");
        }
        for listed in super::NOT_OPERATIONS {
            assert!(super::audited_methods().contains(listed), "{listed} names no method");
        }
        #[cfg(feature = "metered")]
        for listed in super::METERED_NOT_OPERATIONS {
            assert!(super::audited_methods().contains(listed), "{listed} names no method");
            assert!(!super::changes_state(listed), "{listed} is a read");
        }
    }
}