gunnar-grpc 3.0.0

//! **The gate: the allow-all hole that shipped in v1, closed.**
//!
//! Measure, never assert is the whole shape of this file. A gate that refuses *everything*
//! passes a naive "unauthenticated calls are refused" suite exactly as well as
//! a correct one, and a handler hard-wired to `Ok` passes an "an admin may act"
//! suite exactly as well. So every claim here is asserted in both directions,
//! and the two constants that could hide a dead check are proved dead-or-alive
//! explicitly:
//!
//! - `may_stubbed_to_deny_turns_every_allow_red`: with
//!   [`MemoryControlPlane::force_may`]`(Some(false))`, calls that succeeded
//!   above now fail. So the success came *through* the grant decision, not past
//!   it.
//! - `may_stubbed_to_allow_turns_every_denial_green`: with `Some(true)`, calls
//!   that were `PERMISSION_DENIED` above now succeed. So the denial came from
//!   the grant decision, not from a gate that refuses everything.
//! - `every_rpc_refuses_an_unauthenticated_caller` drives every declared RPC
//!   **with no interceptor in the way**, so a handler that forgot its own check
//!   goes red.
//! - `a_valid_credential_is_not_refused` proves the `UNAUTHENTICATED` above is
//!   not a constant either.
//!
//! Nothing here is signed or verified by a stand-in: every credential is a real
//! `sshsig` over a real ed25519 key, minted in-process.

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

use gunnar_grpc::auth::{audited_methods, AuthFailure, AuthGate, MAX_CLOCK_SKEW_MS};
use gunnar_grpc::{
    sign_credential, sign_credential_for, AccessService, AccountsService, AdminService, AuthMode,
    Caller, ControlPlane, EventBus, EventsService, ForgeService, MemoryControlPlane, MirrorService,
    PolicyService, PullsService, RefsService, ReposService, ServerConfig, ServiceContext,
    UsageService, WebhooksService, AUDIT_TARGET, CREDENTIAL_METADATA_KEY,
};
use gunnar_proto::access_server::Access as _;
use gunnar_proto::accounts_server::Accounts as _;
use gunnar_proto::admin_server::Admin as _;
use gunnar_proto::events_server::Events as _;
use gunnar_proto::sso_server::Sso as _;
// Clans (2026-10-05), kept apart from the other services' imports.
use gunnar_proto::clans_server::Clans as _;
use gunnar_proto::forge_server::Forge as _;
use gunnar_proto::mirror_server::Mirror as _;
use gunnar_proto::policy_server::Policy as _;
use gunnar_proto::pull_requests_server::PullRequests as _;
use gunnar_proto::refs_server::Refs as _;
use gunnar_proto::repos_server::Repos as _;
use gunnar_proto::usage_batch_server::UsageBatch as _;
use gunnar_proto::usage_server::Usage as _;
use gunnar_proto::*;
use tonic::{Code, Request};
use tracing::subscriber::with_default;
use tracing_subscriber::layer::SubscriberExt;

// ───────────────────────────────────────────────────────────────────────────
// Fixture
// ───────────────────────────────────────────────────────────────────────────

/// **Every RPC the protos declare, read out of the protos** (GUN-023).
///
/// This was `const DECLARED_RPCS: usize = 72 + …`, a number somebody wrote
/// down. The protos declared 84 non-metered RPCs, `auth::METHODS` listed 70,
/// and the two assertions below compared 70 against 72 and were green — so
/// twelve RPCs (`Access`'s four group verbs, `Forge.GetIssue`,
/// `Forge.ListIssues`, `Mirror.SetCredential`, `Mirror.ResyncReplication`,
/// `Usage.Activity` and all three `Webhooks`) were in no list, were never
/// driven by the sweep, and would have gone on being undriven if one of them
/// had lost its `require` call. The census that exists to prove nothing is
/// missed was itself missing things.
///
/// A count cannot fix that, because a count is what was wrong. The set is
/// derived from the `.proto` files — `include_str!`, so they are embedded at
/// compile time and this cannot read a different tree from the one that built
/// `gunnar-proto` — and compared as a SET. A new RPC now shows up by name.
fn declared_rpcs() -> BTreeSet<String> {
    const PROTOS: &[&str] = &[
        include_str!("../../gunnar-proto/proto/gunnar/v1/access.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/accounts.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/enterprise.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/admin.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/bootstrap.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/clans.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/events.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/forge.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/mirror.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/policy.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/pulls.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/refs.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/repos.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/sso.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/usage.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/usage_batch.proto"),
        include_str!("../../gunnar-proto/proto/gunnar/v1/webhooks.proto"),
        // `entitlement.proto` is compiled only with the feature, so the
        // community build must not be held to its four RPCs: absent code has
        // no refusal to check.
        #[cfg(feature = "metered")]
        include_str!("../../gunnar-proto/proto/gunnar/v1/entitlement.proto"),
        // `billing.proto` likewise (2026-09-17): the tenant's own renewal
        // path, which the community appliance does not carry at all.
        #[cfg(feature = "metered")]
        include_str!("../../gunnar-proto/proto/gunnar/v1/billing.proto"),
        // `sealed.proto` likewise (2026-10-09): the named sealed slot.
        #[cfg(feature = "metered")]
        include_str!("../../gunnar-proto/proto/gunnar/v1/sealed.proto"),
    ];
    let mut out = BTreeSet::new();
    for text in PROTOS {
        let mut service: Option<&str> = None;
        for line in text.lines() {
            let line = line.trim();
            if let Some(rest) = line.strip_prefix("service ") {
                service = rest.split_whitespace().next();
            } else if line == "}" {
                service = None;
            } else if let Some(rest) = line.strip_prefix("rpc ") {
                let (Some(service), Some(method)) = (service, rest.split('(').next()) else {
                    continue;
                };
                out.insert(format!("{service}.{}", method.trim()));
            }
        }
    }
    // The parser is the weak link, so it is measured rather than trusted: a
    // scanner that found nothing, or that found only one service, would make
    // every assertion below vacuous.
    assert!(
        out.len() > 60 && out.contains("Bootstrap.Enroll") && out.contains("Repos.Delete"),
        "the proto scanner did not read the service blocks: {out:?}"
    );
    out
}

const NS: &str = "nordisk";
const REPO: &str = "gunnar";

fn mint() -> ssh_key::PrivateKey {
    ssh_key::PrivateKey::random(
        &mut rand::rand_core::UnwrapErr(rand::rngs::SysRng),
        ssh_key::Algorithm::Ed25519,
    )
    .expect("mint an ed25519 key")
}

fn openssh_of(key: &ssh_key::PrivateKey) -> String {
    key.public_key()
        .to_openssh()
        .expect("render the public key")
}

struct Harness {
    plane: Arc<MemoryControlPlane>,
    ctx: ServiceContext,
}

impl Harness {
    /// A server with authentication ON, one namespace, one repo, and nothing
    /// else. Callers are added by the tests, so a test that forgets to grant
    /// anything gets a refusal rather than an accident.
    fn new() -> Harness {
        let plane = Arc::new(MemoryControlPlane::with_repo(NS, REPO));
        let control: Arc<dyn ControlPlane> = plane.clone();
        let ctx = ServiceContext::new(
            control,
            Arc::new(EventBus::new(64)),
            ServerConfig::default(),
        );
        assert_eq!(
            ctx.config.auth,
            AuthMode::Required,
            "the fixture must run the SHIPPED default; a test suite that quietly ran the \
             escape hatch would prove nothing"
        );
        Harness { plane, ctx }
    }

    /// Register a principal with a freshly minted key. Returns the key and the
    /// principal id; no grant is given.
    fn principal(&self, name: &str) -> (String, ssh_key::PrivateKey) {
        let key = mint();
        let principal = self
            .plane
            .add_principal(name, name, PrincipalKind::User)
            .expect("principal added");
        self.plane
            .add_key(&principal.id, &openssh_of(&key), name, None, "", None)
            .expect("key registered");
        (principal.id, key)
    }

    fn gate(&self) -> AuthGate {
        AuthGate::new(Arc::clone(&self.ctx.control), self.ctx.config.auth)
    }

    /// The caller the gate produces for this key, through the real gate and
    /// over a real signature, so no test invents an identity the server would
    /// not.
    fn caller_for(&self, key: &ssh_key::PrivateKey, nonce: &str) -> Caller {
        let credential = sign_credential_for(key, now_ms(), nonce, "Admin.Health").expect("sign");
        let mut request = Request::new(());
        request.metadata_mut().insert(
            CREDENTIAL_METADATA_KEY,
            credential.to_header_value().parse().expect("ascii"),
        );
        self.gate()
            .authenticate_bound(request.metadata(), now_ms(), Some("Admin.Health"))
            .expect("a freshly signed credential from a registered key must verify")
    }

    fn authed<T>(&self, caller: &Caller, message: T) -> Request<T> {
        let mut request = Request::new(message);
        request.extensions_mut().insert(caller.clone());
        request
    }

    fn repos(&self) -> ReposService {
        ReposService::new(self.ctx.clone())
    }
    fn accounts(&self) -> AccountsService {
        AccountsService::new(self.ctx.clone())
    }
    fn clans(&self) -> gunnar_grpc::ClansService {
        gunnar_grpc::ClansService::new(self.ctx.clone())
    }
    fn refs(&self) -> RefsService {
        RefsService::new(self.ctx.clone())
    }
    fn access(&self) -> AccessService {
        AccessService::new(self.ctx.clone())
    }
    fn enterprise(&self) -> gunnar_grpc::EnterpriseService {
        gunnar_grpc::EnterpriseService::new(self.ctx.clone())
    }
    fn sso(&self) -> gunnar_grpc::SsoService {
        gunnar_grpc::SsoService::new(self.ctx.clone())
    }
    fn policy(&self) -> PolicyService {
        PolicyService::new(self.ctx.clone())
    }
    fn events(&self) -> EventsService {
        EventsService::new(self.ctx.clone())
    }
    fn admin(&self) -> AdminService {
        AdminService::new(self.ctx.clone())
    }
    fn usage(&self) -> UsageService {
        UsageService::new(self.ctx.clone())
    }
    fn usage_batch(&self) -> gunnar_grpc::UsageBatchService {
        gunnar_grpc::UsageBatchService::new(self.ctx.clone())
    }
    fn mirror(&self) -> MirrorService {
        MirrorService::new(self.ctx.clone())
    }
    fn forge(&self) -> ForgeService {
        ForgeService::new(self.ctx.clone())
    }
    fn pulls(&self) -> PullsService {
        PullsService::new(self.ctx.clone())
    }
    fn webhooks(&self) -> WebhooksService {
        WebhooksService::new(self.ctx.clone())
    }
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("after 1970")
        .as_millis() as i64
}

fn path() -> Option<RepoPath> {
    Some(RepoPath::new(NS, REPO))
}

// ───────────────────────────────────────────────────────────────────────────
// Authentication: the credential itself
// ───────────────────────────────────────────────────────────────────────────

#[test]
fn a_registered_key_authenticates_and_an_unregistered_one_does_not() {
    let harness = Harness::new();
    let (principal_id, registered) = harness.principal("alice");
    let stranger = mint();
    let gate = harness.gate();

    let mut metadata = tonic::metadata::MetadataMap::new();
    let good = sign_credential_for(&registered, now_ms(), "nonce-registered-01", "Admin.Health")
        .expect("sign");
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        good.to_header_value().parse().unwrap(),
    );
    let caller = gate
        .authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
        .expect("a registered key must authenticate");
    assert_eq!(caller.principal_id, principal_id);
    assert!(caller.authenticated);
    assert_eq!(
        caller.fingerprint,
        registered
            .public_key()
            .fingerprint(ssh_key::HashAlg::Sha256)
            .to_string(),
        "the caller must carry the fingerprint that was actually presented"
    );

    // The same shape, a key nobody registered. If this passed, the whole
    // scheme would be decoration.
    let mut metadata = tonic::metadata::MetadataMap::new();
    let forged = sign_credential_for(&stranger, now_ms(), "nonce-stranger-01", "Admin.Health")
        .expect("sign");
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        forged.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::NotVerified),
        "an unregistered key authenticated"
    );
}

#[test]
fn a_credential_signed_over_different_bytes_does_not_verify() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();

    // Sign one challenge, present another: same key, same signature, a nonce
    // that was not the one signed. This is the check that proves the signature
    // is over the challenge, rather than simply being there.
    let mut credential =
        sign_credential_for(&key, now_ms(), "nonce-original-000", "Admin.Health").expect("sign");
    credential.nonce = "nonce-substituted".into();

    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::NotVerified),
        "a signature over other bytes was accepted"
    );
}

#[test]
fn a_credential_is_single_use_and_expires() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();

    let credential =
        sign_credential_for(&key, now_ms(), "nonce-replay-0001", "Admin.Health").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );

    gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
        .expect("first use must be accepted");
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::Replayed),
        "the same credential was accepted twice; it is not single-use"
    );

    // …and one issued outside the window is refused before the signature is
    // even looked at.
    let stale = sign_credential_for(
        &key,
        now_ms() - MAX_CLOCK_SKEW_MS - 1_000,
        "nonce-stale-00001",
        "Admin.Health",
    )
    .expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        stale.to_header_value().parse().unwrap(),
    );
    assert!(
        matches!(
            gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
            Err(AuthFailure::Stale { .. })
        ),
        "a credential older than the skew window was accepted"
    );
}

#[test]
fn a_missing_or_malformed_credential_is_refused() {
    let harness = Harness::new();
    let gate = harness.gate();

    assert_eq!(
        gate.authenticate_bound(
            &tonic::metadata::MetadataMap::new(),
            now_ms(),
            Some("Admin.Health")
        ),
        Err(AuthFailure::Missing)
    );

    for bad in [
        "",
        "Bearer hunter2",
        "gunnar-sshsig-v1",
        "gunnar-sshsig-v1 SHA256:x 1 nonce",
        "gunnar-sshsig-v1 not-a-fingerprint 1 abcdefghijklmnop sig",
        "gunnar-sshsig-v1 SHA256:x not-a-number abcdefghijklmnop sig",
        // A nonce too short to be single-use.
        "gunnar-sshsig-v1 SHA256:x 1 short sig",
    ] {
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert(CREDENTIAL_METADATA_KEY, bad.parse().unwrap());
        assert!(
            matches!(
                gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
                Err(AuthFailure::Malformed(_))
            ),
            "{bad:?} was not refused as malformed"
        );
    }
}

#[test]
fn a_revoked_key_stops_authenticating() {
    // The point of reusing the ACL key table: revoking a key for git revokes it
    // for the control plane in the same instant, because there is one table.
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let fingerprint = key
        .public_key()
        .fingerprint(ssh_key::HashAlg::Sha256)
        .to_string();
    let gate = harness.gate();

    let before =
        sign_credential_for(&key, now_ms(), "nonce-before-revoke", "Admin.Health").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        before.to_header_value().parse().unwrap(),
    );
    gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
        .expect("before revocation the key works");

    harness.plane.revoke_key(&fingerprint).expect("revoked");

    let after =
        sign_credential_for(&key, now_ms(), "nonce-after-revoke0", "Admin.Health").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        after.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::NotVerified),
        "a revoked key still authenticated to the control plane"
    );
}

/// **The control plane has no anonymous caller, and never grows one.**
///
/// `Repo.visibility` now means a stranger with no key can *clone a public
/// repository over SSH*. That is a git-wire capability and it must not become a
/// control-plane one: an anonymous `Caller` reaching a handler would carry the
/// public-read rule of `AuthTable::may` into `Repos.Get`, `Refs.List` and
/// `Refs.ReadObject`, which is a control-plane read a stranger was never
/// supposed to have.
///
/// It is structurally impossible through `gunnar-core`: the reserved id cannot
/// be registered, so no key can be registered to it, so no credential can be
/// signed for it. **That is exactly why this test forges the answer**: a guard
/// against an unreachable state is worth nothing until it has been watched
/// refusing, and `AuthGate` takes `ControlPlane` as a trait, so any adapter can
/// return any principal id it likes.
#[test]
fn a_credential_that_resolves_to_the_anonymous_principal_is_refused() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();

    let authenticate = |nonce: &str| {
        let credential = sign_credential_for(&key, now_ms(), nonce, "Admin.Health").expect("sign");
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert(
            CREDENTIAL_METADATA_KEY,
            credential.to_header_value().parse().unwrap(),
        );
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
    };

    // Green first, so the refusal below cannot be a gate that refuses
    // everything: the same key, the same signature scheme, an ordinary
    // principal id.
    let ordinary = authenticate("nonce-ordinary-caller")
        .expect("an ordinary principal's credential must verify");
    assert_ne!(ordinary.principal_id, gunnar_grpc::ANONYMOUS_PRINCIPAL_ID);
    assert!(ordinary.authenticated);

    // Red: the identical credential, with the control plane naming the
    // anonymous principal.
    harness
        .plane
        .forge_principal_id(Some(gunnar_grpc::ANONYMOUS_PRINCIPAL_ID));
    assert_eq!(
        authenticate("nonce-anonymous-caller"),
        Err(AuthFailure::Anonymous),
        "a credential resolving to the anonymous principal produced a CALLER. Anonymous is a \
         read-only git identity and must never reach the control plane."
    );

    // The caller is told nothing that distinguishes it. Saying "you are
    // anonymous" would confirm the id means something here, and it does not.
    let status = AuthFailure::Anonymous.status();
    assert_eq!(status.code(), Code::Unauthenticated);
    assert_eq!(
        status.message(),
        AuthFailure::NotVerified.status().message()
    );

    // And the audit reason is its own, so an operator can tell this apart from
    // an ordinary bad signature in the log even though the client cannot on the
    // wire.
    assert_ne!(
        AuthFailure::Anonymous.reason(),
        AuthFailure::NotVerified.reason()
    );

    // Restoring the plane restores the caller: the instrument is not sticky,
    // and the refusal above was about the forged id and nothing else.
    harness.plane.forge_principal_id(None);
    authenticate("nonce-restored-caller").expect("the ordinary answer is back");
}

/// Not one declared RPC is named with the anonymous id, and no method is
/// exempt from the gate. A "public read over gRPC" bolted on later would show
/// up here first.
#[test]
fn no_rpc_is_reachable_without_a_caller_and_none_is_anonymous_scoped() {
    assert!(
        !gunnar_grpc::ANONYMOUS_PRINCIPAL_ID.is_empty(),
        "the reserved id must be a real string; an empty one would match a Caller with no \
         principal and refuse honest traffic"
    );
    // Every audited method is in `METHODS`, and `every_rpc_refuses_an_\
    // unauthenticated_caller` (below) drives all of them with no caller at all.
    // This asserts the *count* has not silently shrunk to nothing, which would
    // make that test vacuous.
    // `DECLARED_RPCS` in the protos, minus the two `Bootstrap` RPCs served with
    // no credential on purpose. Written as the difference rather than as 60, so a
    // new RPC has to be driven above or declared open by name — it cannot be
    // absorbed by a constant somebody bumps.
    let accounted: BTreeSet<String> = audited_methods()
        .iter()
        .chain(gunnar_grpc::UNAUTHENTICATED_METHODS)
        .map(|m| (*m).to_owned())
        .collect();
    let declared = declared_rpcs();
    assert_eq!(
        declared.difference(&accounted).collect::<Vec<_>>(),
        Vec::<&String>::new(),
        "these RPCs are declared in the protos and are in neither `auth::METHODS` nor \
         `UNAUTHENTICATED_METHODS`, so nothing checks that they refuse"
    );
    assert_eq!(
        accounted.difference(&declared).collect::<Vec<_>>(),
        Vec::<&String>::new(),
        "these are named in gunnar-grpc's lists and are not RPCs the protos declare"
    );
}

// ───────────────────────────────────────────────────────────────────────────
// Authorisation: the three directions
// ───────────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn an_admin_may_mutate_and_a_reader_may_not() {
    let harness = Harness::new();
    let (admin_id, admin_key) = harness.principal("admin");
    harness.plane.grant_server(&admin_id, GrantLevel::Admin);
    let (reader_id, reader_key) = harness.principal("reader");
    harness
        .plane
        .grant(&reader_id, NS, REPO, GrantLevel::Read)
        .expect("granted read on the repo");

    let admin = harness.caller_for(&admin_key, "nonce-admin-000001");
    let reader = harness.caller_for(&reader_key, "nonce-reader-00001");

    // 1. Authenticated, with the grant → allowed.
    harness
        .access()
        .add_principal(harness.authed(
            &admin,
            AddPrincipalRequest {
                name: "ci".into(),
                kind: PrincipalKind::User as i32,
                id: String::new(),
            },
        ))
        .await
        .expect("a server-Admin principal may add a principal");

    // 2. Authenticated, WITHOUT the grant → refused, and refused as
    //    PERMISSION_DENIED rather than UNAUTHENTICATED: the server knows
    //    exactly who this is.
    let status = harness
        .access()
        .add_principal(harness.authed(
            &reader,
            AddPrincipalRequest {
                name: "smuggled".into(),
                kind: PrincipalKind::User as i32,
                id: String::new(),
            },
        ))
        .await
        .expect_err("a read-only principal must not write the key table");
    assert_eq!(status.code(), Code::PermissionDenied);
    assert!(
        status.message().contains(&reader_id),
        "the refusal must name the principal it refused: {}",
        status.message()
    );

    // 3. The same reader IS allowed what it was granted, so the refusal above
    //    is not "this principal can do nothing".
    harness
        .repos()
        .get(harness.authed(&reader, GetRepoRequest { path: path() }))
        .await
        .expect("a Read grant on the repo must allow Repos.Get");

    // 4. …and is refused a repo it has no grant on, so the allow above is not
    //    "any authenticated caller may read anything".
    harness
        .repos()
        .create_namespace(harness.authed(
            &admin,
            CreateNamespaceRequest {
                name: "other".into(),
                description: String::new(),
            },
        ))
        .await
        .expect("admin creates a second namespace");
    harness
        .repos()
        .create(harness.authed(
            &admin,
            CreateRepoRequest {
                path: Some(RepoPath::new("other", "secret")),
                ..Default::default()
            },
        ))
        .await
        .expect("admin creates a repo in it");
    let status = harness
        .repos()
        .get(harness.authed(
            &reader,
            GetRepoRequest {
                path: Some(RepoPath::new("other", "secret")),
            },
        ))
        .await
        .expect_err("a grant on one repo must not reach another");
    assert_eq!(status.code(), Code::PermissionDenied);
}

/// **Blame and search are Read-gated, like every other `Refs` read.**
///
/// The two directions the sweep above cannot show: a caller *with* Read passes
/// the gate (the memory plane then answers `Unimplemented`, which is the gate
/// having let it THROUGH — a `PermissionDenied` here would mean the verb was
/// gated too tight for the operator it exists for), and a caller with no grant
/// at all is refused `PermissionDenied`. Written against the red where the two
/// new RPCs forgot their `require` call and answered anyone, and the red where
/// they demanded more than Read.
#[tokio::test]
async fn blame_and_search_need_read_and_no_more() {
    let harness = Harness::new();
    let (reader_id, reader_key) = harness.principal("reader");
    harness
        .plane
        .grant(&reader_id, NS, REPO, GrantLevel::Read)
        .expect("granted read on the repo");
    let reader = harness.caller_for(&reader_key, "nonce-reader-blame1");

    let (stranger_id, stranger_key) = harness.principal("stranger");
    let stranger = harness.caller_for(&stranger_key, "nonce-stranger-001");

    // A Read grant passes the gate on both verbs — the memory plane then answers
    // Unimplemented, which is the gate having admitted the caller.
    let code = harness
        .refs()
        .blame(harness.authed(
            &reader,
            BlameRequest {
                repo: path(),
                path: "README.md".into(),
                ..Default::default()
            },
        ))
        .await
        .expect_err("the memory plane does not implement blame")
        .code();
    assert_eq!(
        code,
        Code::Unimplemented,
        "a Read grant must let a caller reach Refs.Blame; got {code:?}"
    );
    let code = harness
        .refs()
        .search(harness.authed(
            &reader,
            SearchRequest {
                repo: path(),
                pattern: "needle".into(),
                ..Default::default()
            },
        ))
        .await
        .expect_err("the memory plane does not implement search")
        .code();
    assert_eq!(
        code,
        Code::Unimplemented,
        "a Read grant must reach Refs.Search"
    );

    // No grant at all → refused, and refused as PermissionDenied (the server
    // knows who this is) rather than reaching the store.
    let status = harness
        .refs()
        .blame(harness.authed(
            &stranger,
            BlameRequest {
                repo: path(),
                path: "README.md".into(),
                ..Default::default()
            },
        ))
        .await
        .expect_err("a caller with no read on the repo must not blame it");
    assert_eq!(status.code(), Code::PermissionDenied);
    assert!(
        status.message().contains(&stranger_id),
        "the refusal must name the principal it refused: {}",
        status.message()
    );
    let status = harness
        .refs()
        .search(harness.authed(
            &stranger,
            SearchRequest {
                repo: path(),
                pattern: "needle".into(),
                ..Default::default()
            },
        ))
        .await
        .expect_err("a caller with no read on the repo must not search it");
    assert_eq!(status.code(), Code::PermissionDenied);
}

/// **`Repos.Gc` needs `Write`, and `Write` is enough.**
///
/// Both directions, because either one alone is a guard that cannot fail. A
/// caller holding `Write` on the repository passes the gate — the memory plane
/// then answers `Unimplemented`, which is the gate having let it THROUGH, and a
/// `PermissionDenied` there is the bug this test exists for. A caller holding
/// only `Read` is refused, so "Write is enough" is not "anyone who is
/// authenticated is enough".
///
/// The upper direction is the one that matters. gc destroys nothing —
/// `storage.rs`: "every lazily-exploded object unlinked, the commit-and-tree
/// pool kept because that is the graph" — and gating it at `Admin` like its
/// destructive neighbours is what left 26 repositories on the estate never once
/// maintained. If someone later "tightens" this to `Admin` on the grounds that
/// every other reclaim verb is, this test goes red and says why.
#[tokio::test]
async fn gc_needs_write_and_no_more() {
    let harness = Harness::new();

    let (_writer_id, writer_key) = harness.principal("writer");
    harness
        .plane
        .grant(&_writer_id, NS, REPO, GrantLevel::Write)
        .expect("granted write on the repo");
    let writer = harness.caller_for(&writer_key, "nonce-writer-gc0001");

    let (reader_id, reader_key) = harness.principal("gc-reader");
    harness
        .plane
        .grant(&reader_id, NS, REPO, GrantLevel::Read)
        .expect("granted read on the repo");
    let reader = harness.caller_for(&reader_key, "nonce-gcreader-0001");

    // Write passes the gate. Unimplemented is the memory plane, not the gate.
    let code = harness
        .repos()
        .gc(harness.authed(&writer, GcRepoRequest { path: path() }))
        .await
        .expect_err("the memory plane does not implement gc")
        .code();
    assert_eq!(
        code,
        Code::Unimplemented,
        "a Write grant must let a caller reach Repos.Gc; got {code:?}. \
         PermissionDenied here means gc is gated above Write again."
    );

    // Read does not.
    let status = harness
        .repos()
        .gc(harness.authed(&reader, GcRepoRequest { path: path() }))
        .await
        .expect_err("a read-only principal must not gc the repository");
    assert_eq!(status.code(), Code::PermissionDenied);
    assert!(
        status.message().contains(&reader_id),
        "the refusal must name the principal it refused: {}",
        status.message()
    );

    // **But Read DOES reach `GcStatus`**, and that pairing is the design:
    // asking whether maintenance ran is not asking for it. A console that could
    // press the button but not watch it would be the worse half of the pair.
    let code = harness
        .repos()
        .gc_status(harness.authed(&reader, GcStatusRequest { path: path() }))
        .await
        .expect_err("the memory plane does not implement gc status")
        .code();
    assert_eq!(
        code,
        Code::Unimplemented,
        "a Read grant must let a caller reach Repos.GcStatus; got {code:?}"
    );

    // And a stranger reaches neither, so "Read is enough for GcStatus" is not
    // "no grant is needed".
    let (_stranger_id, stranger_key) = harness.principal("gc-stranger");
    let stranger = harness.caller_for(&stranger_key, "nonce-gcstranger-01");
    let status = harness
        .repos()
        .gc_status(harness.authed(&stranger, GcStatusRequest { path: path() }))
        .await
        .expect_err("a caller with no grant must not read the repository's gc");
    assert_eq!(status.code(), Code::PermissionDenied);
}

#[tokio::test]
async fn a_namespace_grant_does_not_reach_a_repo_in_another_namespace() {
    // the one-sandbox rule pins that grants do NOT inherit. The gRPC gate must not be the
    // place that quietly reintroduces inheritance.
    let harness = Harness::new();
    let (admin_id, admin_key) = harness.principal("admin");
    harness.plane.grant_server(&admin_id, GrantLevel::Admin);
    let admin = harness.caller_for(&admin_key, "nonce-admin-000002");

    let (tenant_id, tenant_key) = harness.principal("tenant");
    harness
        .plane
        .grant(&tenant_id, NS, "", GrantLevel::Admin)
        .expect("namespace-wide admin");
    let tenant = harness.caller_for(&tenant_key, "nonce-tenant-00001");

    // Inside the namespace: allowed.
    harness
        .repos()
        .create(harness.authed(
            &tenant,
            CreateRepoRequest {
                path: Some(RepoPath::new(NS, "second")),
                ..Default::default()
            },
        ))
        .await
        .expect("a namespace admin may create a repo in that namespace");

    // A different namespace: refused, even though the level is the same.
    harness
        .repos()
        .create_namespace(harness.authed(
            &admin,
            CreateNamespaceRequest {
                name: "elsewhere".into(),
                description: String::new(),
            },
        ))
        .await
        .expect("admin creates it");
    let status = harness
        .repos()
        .create(harness.authed(
            &tenant,
            CreateRepoRequest {
                path: Some(RepoPath::new("elsewhere", "sneaky")),
                ..Default::default()
            },
        ))
        .await
        .expect_err("a grant on one namespace must not reach another");
    assert_eq!(status.code(), Code::PermissionDenied);
}

// ───────────────────────────────────────────────────────────────────────────
// Measure, never assert: neither constant survives
// ───────────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn may_stubbed_to_deny_turns_every_allow_red() {
    // If `Repos.Get` had been hard-wired to succeed, this would still pass the
    // happy-path test above. Stub the grant decision to deny and it must fail:
    // that is the proof the success went THROUGH `Identity::may`.
    let harness = Harness::new();
    let (admin_id, admin_key) = harness.principal("admin");
    harness.plane.grant_server(&admin_id, GrantLevel::Admin);
    let admin = harness.caller_for(&admin_key, "nonce-admin-000003");

    harness
        .repos()
        .get(harness.authed(&admin, GetRepoRequest { path: path() }))
        .await
        .expect("the real answer allows this");

    harness.plane.force_may(Some(false));

    for (name, code) in [
        (
            "Repos.Get",
            harness
                .repos()
                .get(harness.authed(&admin, GetRepoRequest { path: path() }))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        (
            "Access.AddPrincipal",
            harness
                .access()
                .add_principal(harness.authed(
                    &admin,
                    AddPrincipalRequest {
                        name: "nope".into(),
                        kind: PrincipalKind::User as i32,
                        id: String::new(),
                    },
                ))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        (
            "Admin.Stats",
            harness
                .admin()
                .stats(harness.authed(&admin, StatsRequest {}))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // The PKI read consults the grant decision like everything else. The
        // certificate in the answer is public; the trusted-CA list and the
        // created/imported origin beside it are this box's state, and a
        // credential is not by itself a grant to read it.
        (
            "Admin.IssuerInfo",
            harness
                .admin()
                .issuer_info(harness.authed(&admin, gunnar_proto::IssuerInfoRequest {}))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // The estate's boxes (2026-10-04): addresses and principals of the
        // whole estate, a server-wide fact — a credential is not by itself a
        // grant to read it.
        (
            "Admin.Topology",
            harness
                .admin()
                .topology(harness.authed(&admin, gunnar_proto::TopologyRequest {}))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // The audit trail read (lane T11). Its answer is a count and a hash
        // and carries no entry — but who-audits-whom is this box's state, and
        // a credential is not by itself a grant to read it.
        (
            "Admin.AuditTrail",
            harness
                .admin()
                .audit_trail(harness.authed(&admin, gunnar_proto::AuditTrailRequest { at_seq: 0 }))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // The tunnel address decides what the hub routes to this box: held
        // to server Admin through `may`, like GrowStart.
        (
            "Admin.SetTunnelAddress",
            harness
                .admin()
                .set_tunnel_address(harness.authed(
                    &admin,
                    gunnar_proto::AssignTunnelAddressRequest {
                        address: "10.99.0.2/24".into(),
                    },
                ))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // Server ADMIN (lane T9): stubbed to deny, it must be refused before the
        // seam is asked to sign anything.
        (
            "Admin.IssueTwinCertificate",
            harness
                .admin()
                .issue_twin_certificate(harness.authed(
                    &admin,
                    gunnar_proto::IssueTwinCertificateRequest {
                        principal: "twin-box".into(),
                        csr_pem: String::new(),
                    },
                ))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
        // A capacity quote names the box's free space and its disk floor. An
        // authenticated caller is not by itself entitled to it: monetize is
        // told what a tenant may be sold, and a tenant is not.
        (
            "Admin.Capacity",
            harness
                .admin()
                .capacity(harness.authed(
                    &admin,
                    CapacityRequest {
                        requested_bytes: 1,
                        namespace: String::new(),
                        disks: false,
                    },
                ))
                .await
                .map(|_| ())
                .map_err(|e| e.code()),
        ),
    ] {
        assert_eq!(
            code,
            Err(Code::PermissionDenied),
            "{name} succeeded with the grant decision stubbed to deny, so it is not \
             consulting it"
        );
    }
}

#[tokio::test]
async fn may_stubbed_to_allow_turns_every_denial_green() {
    // The other half, and the one that catches a gate which simply refuses
    // everything: with the grant decision stubbed to allow, the calls that were
    // PERMISSION_DENIED must now go through.
    let harness = Harness::new();
    let (_, key) = harness.principal("nobody");
    let nobody = harness.caller_for(&key, "nonce-nobody-00001");

    let status = harness
        .access()
        .add_principal(harness.authed(
            &nobody,
            AddPrincipalRequest {
                name: "before".into(),
                kind: PrincipalKind::User as i32,
                id: String::new(),
            },
        ))
        .await
        .expect_err("a principal with no grants must be refused");
    assert_eq!(status.code(), Code::PermissionDenied);

    harness.plane.force_may(Some(true));

    harness
        .access()
        .add_principal(harness.authed(
            &nobody,
            AddPrincipalRequest {
                name: "after".into(),
                kind: PrincipalKind::User as i32,
                id: String::new(),
            },
        ))
        .await
        .expect(
            "with the grant decision stubbed to allow, the call must go through — otherwise \
             the refusal above came from a gate that refuses everything and proves nothing",
        );
    harness
        .repos()
        .get(harness.authed(&nobody, GetRepoRequest { path: path() }))
        .await
        .expect("and so must the read");
    // `Admin.Capacity` cannot SUCCEED here — the memory control plane has no
    // filesystem to measure and says `Unimplemented` — but it must no longer
    // be refused BY THE GATE, or the PERMISSION_DENIED asserted for it in
    // `may_stubbed_to_deny_turns_every_allow_red` would prove nothing but that
    // the handler refuses everyone.
    let status = harness
        .admin()
        .capacity(harness.authed(
            &nobody,
            CapacityRequest {
                requested_bytes: 1,
                namespace: String::new(),
                disks: false,
            },
        ))
        .await
        .expect_err("the test double holds no disk, so it cannot answer");
    assert_eq!(
        status.code(),
        Code::Unimplemented,
        "with the grant decision stubbed to allow, Admin.Capacity must get PAST the gate; it \
         answered {:?}",
        status.code()
    );
}

#[tokio::test]
async fn a_valid_credential_is_not_refused() {
    // The counterweight to `every_rpc_refuses_an_unauthenticated_caller`: the
    // UNAUTHENTICATED it asserts must not be a constant.
    let harness = Harness::new();
    let (admin_id, admin_key) = harness.principal("admin");
    harness.plane.grant_server(&admin_id, GrantLevel::Admin);
    let admin = harness.caller_for(&admin_key, "nonce-admin-000004");

    harness
        .admin()
        .health(harness.authed(&admin, HealthRequest {}))
        .await
        .expect("a verified admin must be served");
}

// ───────────────────────────────────────────────────────────────────────────
// Deny by default: every declared RPC, with no interceptor in the way
// ───────────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn every_rpc_refuses_an_unauthenticated_caller() {
    // These call the handlers DIRECTLY. The interceptor is not in the path, so
    // this is not testing `AuthGate`. It is testing that no handler in the
    // crate reaches the storage seam on its own. A handler that forgot its
    // `authorize` would answer here, and go red.
    let harness = Harness::new();
    let mut covered: BTreeSet<&'static str> = BTreeSet::new();

    macro_rules! refuses {
        ($name:literal, $call:expr) => {{
            covered.insert($name);
            let status = $call
                .await
                .map(|_| ())
                .expect_err(concat!($name, " served an unauthenticated caller"));
            assert_eq!(
                status.code(),
                Code::Unauthenticated,
                "{} answered {:?}, not UNAUTHENTICATED",
                $name,
                status.code()
            );
        }};
    }

    let repos = harness.repos();
    refuses!(
        "Repos.CreateNamespace",
        repos.create_namespace(Request::new(CreateNamespaceRequest::default()))
    );
    refuses!(
        "Repos.ListNamespaces",
        repos.list_namespaces(Request::new(ListNamespacesRequest::default()))
    );
    refuses!(
        "Repos.DeleteNamespace",
        repos.delete_namespace(Request::new(DeleteNamespaceRequest {
            name: NS.into(),
            delete_repos: true,
        }))
    );
    refuses!(
        "Repos.ArchiveOut",
        repos.archive_out(Request::new(ArchiveOutRequest {
            path: path(),
            from_generation: 0,
        }))
    );
    refuses!(
        "Repos.Create",
        repos.create(Request::new(CreateRepoRequest {
            path: Some(RepoPath::new(NS, "new")),
            ..Default::default()
        }))
    );
    refuses!(
        "Repos.Delete",
        repos.delete(Request::new(DeleteRepoRequest {
            path: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Repos.Undelete",
        repos.undelete(Request::new(UndeleteRepoRequest { path: path() }))
    );
    refuses!(
        "Repos.Purge",
        repos.purge(Request::new(PurgeRepoRequest { path: path() }))
    );
    refuses!(
        "Repos.Gc",
        repos.gc(Request::new(GcRepoRequest { path: path() }))
    );
    refuses!(
        "Repos.GcStatus",
        repos.gc_status(Request::new(GcStatusRequest { path: path() }))
    );
    refuses!(
        "Repos.ListDeleted",
        repos.list_deleted(Request::new(ListDeletedRequest::default()))
    );
    refuses!(
        "Repos.ListOrphans",
        repos.list_orphans(Request::new(gunnar_proto::ListOrphansRequest::default()))
    );
    refuses!(
        "Repos.ReclaimOrphan",
        repos.reclaim_orphan(Request::new(gunnar_proto::ReclaimOrphanRequest {
            path: path()
        }))
    );
    refuses!(
        "Repos.List",
        repos.list(Request::new(ListReposRequest::default()))
    );

    // `Accounts`: the per-user gate and the hard purge. Driven with no caller
    // for the reason every other line here is — a handler that forgot its own
    // `authorize` would answer, and answering `Accounts.Purge` to a stranger is
    // the worst hole in the surface.
    let accounts = harness.accounts();
    refuses!(
        "Accounts.Gate",
        accounts.gate(Request::new(gunnar_proto::GateAccountRequest {
            account: NS.into(),
            note: "because".into(),
        }))
    );
    refuses!(
        "Accounts.Ungate",
        accounts.ungate(Request::new(gunnar_proto::UngateAccountRequest {
            account: NS.into(),
        }))
    );
    refuses!(
        "Accounts.ListAccountGates",
        accounts.list_account_gates(Request::new(gunnar_proto::ListAccountGatesRequest {}))
    );
    refuses!(
        "Accounts.Purge",
        accounts.purge(Request::new(gunnar_proto::PurgeAccountRequest {
            account: NS.into(),
            reason: "because".into(),
            remove_principal: false,
            expect_empty: false,
        }))
    );
    refuses!(
        "Accounts.PurgeStatus",
        accounts.purge_status(Request::new(gunnar_proto::PurgeAccountStatusRequest {
            account: NS.into(),
        }))
    );
    refuses!(
        "Accounts.EnrollEmployee",
        accounts.enroll_employee(Request::new(gunnar_proto::EnrollEmployeeRequest {
            account: NS.into(),
            id: "employee".into(),
            name: "employee".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Accounts.ListEmployees",
        accounts.list_employees(Request::new(gunnar_proto::ListEmployeesRequest {
            account: NS.into(),
        }))
    );
    refuses!(
        "Accounts.ClaimOperator",
        accounts.claim_operator(Request::new(gunnar_proto::ClaimOperatorRequest {
            name: "rickard".into(),
            ..Default::default()
        }))
    );
    // Clans (2026-10-05): the person's own verbs, and none without a person.
    let clans = harness.clans();
    let clan = || "fellows".to_owned();
    refuses!(
        "Clans.Create",
        clans.create(Request::new(CreateClanRequest {
            name: clan(),
            contribution_bytes: 0
        }))
    );
    refuses!(
        "Clans.Get",
        clans.get(Request::new(GetClanRequest { clan: clan() }))
    );
    refuses!("Clans.List", clans.list(Request::new(ListClansRequest {})));
    refuses!(
        "Clans.Invite",
        clans.invite(Request::new(InviteToClanRequest {
            clan: clan(),
            principal_id: "bob".into()
        }))
    );
    refuses!(
        "Clans.Accept",
        clans.accept(Request::new(AcceptClanInviteRequest {
            clan: clan(),
            contribution_bytes: 0
        }))
    );
    refuses!(
        "Clans.Decline",
        clans.decline(Request::new(DeclineClanInviteRequest {
            clan: clan(),
            principal_id: String::new()
        }))
    );
    refuses!(
        "Clans.SetContribution",
        clans.set_contribution(Request::new(SetClanContributionRequest {
            clan: clan(),
            contribution_bytes: 1
        }))
    );
    refuses!(
        "Clans.Leave",
        clans.leave(Request::new(LeaveClanRequest { clan: clan() }))
    );
    refuses!(
        "Clans.RemoveMember",
        clans.remove_member(Request::new(RemoveClanMemberRequest {
            clan: clan(),
            principal_id: "bob".into()
        }))
    );
    refuses!(
        "Clans.SetAdmin",
        clans.set_admin(Request::new(SetClanAdminRequest {
            clan: clan(),
            principal_id: "bob".into(),
            admin: true
        }))
    );
    refuses!(
        "Clans.Dissolve",
        clans.dissolve(Request::new(DissolveClanRequest { clan: clan() }))
    );
    refuses!(
        "Clans.OfferRepo",
        clans.offer_repo(Request::new(OfferRepoRequest {
            clan: clan(),
            repo: "app".into(),
            account: "bob".into()
        }))
    );
    refuses!(
        "Clans.AnswerRepoOffer",
        clans.answer_repo_offer(Request::new(AnswerRepoOfferRequest {
            clan: clan(),
            repo: "app".into(),
            accept: true
        }))
    );
    refuses!(
        "Repos.Get",
        repos.get(Request::new(GetRepoRequest { path: path() }))
    );
    refuses!(
        "Repos.SetDefaultBranch",
        repos.set_default_branch(Request::new(SetDefaultBranchRequest {
            path: path(),
            default_branch: "refs/heads/x".into(),
        }))
    );
    refuses!(
        "Repos.SetVisibility",
        repos.set_visibility(Request::new(SetVisibilityRequest {
            path: path(),
            visibility: Visibility::Public as i32,
        }))
    );
    refuses!(
        "Repos.RecipientKeys",
        repos.recipient_keys(Request::new(gunnar_proto::RecipientKeysRequest {
            path: path(),
            principal_id: "someone".into(),
        }))
    );
    refuses!(
        "Repos.SetEncrypted",
        repos.set_encrypted(Request::new(gunnar_proto::SetEncryptedRequest {
            path: path(),
            encrypted: true,
        }))
    );
    refuses!(
        "Repos.SetNamespaceSettings",
        repos.set_namespace_settings(Request::new(gunnar_proto::SetNamespaceSettingsRequest {
            name: NS.into(),
            public: Some(true),
            events_public: None,
        }))
    );

    let refs = harness.refs();
    refuses!(
        "Refs.List",
        refs.list(Request::new(ListRefsRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.Get",
        refs.get(Request::new(GetRefRequest {
            repo: path(),
            ref_name: "refs/heads/main".into(),
            peel: false,
        }))
    );
    refuses!(
        "Refs.Log",
        refs.log(Request::new(LogRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.ReadObject",
        refs.read_object(Request::new(ReadObjectRequest {
            repo: path(),
            oid: "deadbeef".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.ReadLfsObject",
        refs.read_lfs_object(Request::new(ReadLfsObjectRequest {
            repo: path(),
            oid: "f0".repeat(32),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.Diff",
        refs.diff(Request::new(DiffRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.Blame",
        refs.blame(Request::new(BlameRequest {
            repo: path(),
            path: "README.md".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Refs.Search",
        refs.search(Request::new(SearchRequest {
            repo: path(),
            pattern: "needle".into(),
            ..Default::default()
        }))
    );

    let access = harness.access();
    refuses!(
        "Access.AddPrincipal",
        access.add_principal(Request::new(AddPrincipalRequest {
            name: "x".into(),
            kind: PrincipalKind::User as i32,
            id: String::new(),
        }))
    );
    refuses!(
        "Access.ListPrincipals",
        access.list_principals(Request::new(ListPrincipalsRequest::default()))
    );
    refuses!(
        "Access.RemovePrincipal",
        access.remove_principal(Request::new(RemovePrincipalRequest {
            principal_id: "principal-00000001".into(),
        }))
    );
    refuses!(
        "Access.RenamePrincipal",
        access.rename_principal(Request::new(RenamePrincipalRequest {
            principal_id: "principal-00000001".into(),
            name: "somebody else".into(),
        }))
    );
    refuses!(
        "Access.AddKey",
        access.add_key(Request::new(AddKeyRequest {
            principal_id: "principal-00000001".into(),
            openssh_public_key: "ssh-ed25519 AAAA x".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Access.RevokeKey",
        access.revoke_key(Request::new(RevokeKeyRequest {
            fingerprint: "SHA256:whatever".into(),
        }))
    );
    // The four GROUP verbs. They were in no list at all until GUN-023, so
    // this sweep never drove them.
    refuses!(
        "Access.AddGroupMember",
        access.add_group_member(Request::new(AddGroupMemberRequest {
            group_id: "reviewers".into(),
            member_id: "someone".into(),
            signed: None,
        }))
    );
    refuses!(
        "Access.RemoveGroupMember",
        access.remove_group_member(Request::new(RemoveGroupMemberRequest {
            group_id: "reviewers".into(),
            member_id: "someone".into(),
            signed: None,
        }))
    );
    refuses!(
        "Access.ListGroupMembers",
        access.list_group_members(Request::new(ListGroupMembersRequest {
            group_id: "reviewers".into(),
        }))
    );
    refuses!(
        "Access.ListGroupMembershipEvents",
        access.list_group_membership_events(Request::new(ListGroupMembershipEventsRequest {
            group_id: "reviewers".into(),
        }))
    );
    refuses!(
        "Access.ListKeys",
        access.list_keys(Request::new(ListKeysRequest::default()))
    );
    // The caller's own standing (2026-10-04): no grant, but never a stranger.
    refuses!(
        "Access.WhoAmI",
        access.who_am_i(Request::new(WhoAmIRequest::default()))
    );
    refuses!(
        "Access.AddCredential",
        access.add_credential(Request::new(AddCredentialRequest::default()))
    );
    refuses!(
        "Access.ListCredentials",
        access.list_credentials(Request::new(ListCredentialsRequest::default()))
    );
    refuses!(
        "Access.RevokeCredential",
        access.revoke_credential(Request::new(RevokeCredentialRequest::default()))
    );
    refuses!(
        "Access.RenameCredential",
        access.rename_credential(Request::new(RenameCredentialRequest::default()))
    );
    refuses!(
        "Access.GetEncryptionKey",
        access.get_encryption_key(Request::new(GetEncryptionKeyRequest::default()))
    );
    refuses!(
        "Access.CreateEncryptionKey",
        access.create_encryption_key(Request::new(CreateEncryptionKeyRequest::default()))
    );
    refuses!(
        "Access.AddUnlockMethod",
        access.add_unlock_method(Request::new(AddUnlockMethodRequest::default()))
    );
    refuses!(
        "Access.RemoveUnlockMethod",
        access.remove_unlock_method(Request::new(RemoveUnlockMethodRequest::default()))
    );
    refuses!(
        "Access.EnrolmentLink",
        access.enrolment_link(Request::new(EnrolmentLinkRequest::default()))
    );
    refuses!(
        "Access.MachineChallenge",
        access.machine_challenge(Request::new(MachineChallengeRequest::default()))
    );
    refuses!(
        "Access.RequestMachineEnrolment",
        access.request_machine_enrolment(Request::new(RequestMachineEnrolmentRequest::default()))
    );
    refuses!(
        "Access.MachineEnrolmentStatus",
        access.machine_enrolment_status(Request::new(MachineEnrolmentStatusRequest::default()))
    );
    refuses!(
        "Access.ListMachineEnrolments",
        access.list_machine_enrolments(Request::new(ListMachineEnrolmentsRequest::default()))
    );
    refuses!(
        "Access.ApproveMachineEnrolment",
        access.approve_machine_enrolment(Request::new(ApproveMachineEnrolmentRequest::default()))
    );
    refuses!(
        "Access.RefuseMachineEnrolment",
        access.refuse_machine_enrolment(Request::new(RefuseMachineEnrolmentRequest::default()))
    );
    refuses!(
        "Access.Check",
        access.check(Request::new(CheckRequest {
            namespace: NS.into(),
            repo: String::new(),
            level: GrantLevel::Admin as i32,
        }))
    );
    refuses!(
        "Access.Grant",
        access.grant(Request::new(GrantRequest {
            principal_id: "principal-00000001".into(),
            namespace: NS.into(),
            repo: REPO.into(),
            level: GrantLevel::Admin as i32,
        }))
    );
    refuses!(
        "Access.Revoke",
        access.revoke(Request::new(RevokeRequest {
            principal_id: "principal-00000001".into(),
            namespace: NS.into(),
            repo: REPO.into(),
        }))
    );
    refuses!(
        "Access.ListGrants",
        access.list_grants(Request::new(ListGrantsRequest::default()))
    );

    let policy = harness.policy();
    refuses!(
        "Policy.GetProtection",
        policy.get_protection(Request::new(GetProtectionRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Policy.SetProtection",
        policy.set_protection(Request::new(SetProtectionRequest {
            repo: path(),
            protection: Some(BranchProtection {
                ref_pattern: "refs/heads/main".into(),
                ..Default::default()
            }),
        }))
    );
    refuses!(
        "Policy.DeleteProtection",
        policy.delete_protection(Request::new(DeleteProtectionRequest {
            repo: path(),
            ref_pattern: "refs/heads/main".into(),
        }))
    );
    refuses!(
        "Policy.GetRefNamePolicy",
        policy.get_ref_name_policy(Request::new(GetRefNamePolicyRequest { repo: path() }))
    );
    refuses!(
        "Policy.SetRefNamePolicy",
        policy.set_ref_name_policy(Request::new(SetRefNamePolicyRequest {
            repo: path(),
            policy: Some(RefNamePolicy::default()),
        }))
    );
    refuses!(
        "Policy.ListGates",
        policy.list_gates(Request::new(ListGatesRequest { repo: path() }))
    );

    refuses!(
        "Events.Watch",
        harness
            .events()
            .watch(Request::new(WatchRequest::default()))
    );

    let admin = harness.admin();
    refuses!("Admin.Health", admin.health(Request::new(HealthRequest {})));
    // `Admin.Capacity` answers "could you serve a cap of N bytes" — the box's
    // free space, its disk floor and what a named tenant holds. That is the
    // whole filesystem's business, so it is refused to a caller with no
    // credential like every other server-wide read.
    refuses!(
        "Admin.Capacity",
        admin.capacity(Request::new(CapacityRequest {
            requested_bytes: 1,
            namespace: String::new(),
            disks: false,
        }))
    );
    refuses!("Admin.Stats", admin.stats(Request::new(StatsRequest {})));
    // The runtime data-set growth: two mutators and the credentialled status.
    // (`Bootstrap.GrowStatus` is the credential-free twin, in
    // UNAUTHENTICATED_METHODS with its reason.)
    refuses!(
        "Admin.GrowStart",
        admin.grow_start(Request::new(GrowStartRequest {
            go_ahead: b"{}".to_vec(),
            ..Default::default()
        }))
    );
    refuses!(
        "Admin.GrowStatus",
        admin.grow_status(Request::new(GrowStatusRequest {}))
    );
    refuses!(
        "Admin.GrowResume",
        admin.grow_resume(Request::new(GrowResumeRequest { give_up: false }))
    );
    refuses!(
        "Admin.StorageInfo",
        admin.storage_info(Request::new(StorageInfoRequest { repo: None }))
    );
    // **The PKI read** (2026-09-17, lane PKI-READ). Driven the day it was
    // declared, not three lanes later: the certificate it returns is public,
    // but the trusted-CA list, the created/imported origin and the adoption
    // date are this box's PKI state, and a caller with no credential at all
    // gets none of it.
    refuses!(
        "Admin.IssuerInfo",
        admin.issuer_info(Request::new(gunnar_proto::IssuerInfoRequest {}))
    );
    // **The audit trail read is behind the gate too** (lane T11), declared
    // here at the door rather than three lanes later. It returns no entry —
    // but "how many decisions has this box made, and do they still verify" is
    // a fact about the box, and an unauthenticated caller gets none of it.
    refuses!(
        "Admin.AuditTrail",
        admin.audit_trail(Request::new(gunnar_proto::AuditTrailRequest { at_seq: 0 }))
    );
    // **The estate's boxes** (2026-10-04): which boxes exist and where, a
    // stranger gets none of it.
    refuses!(
        "Admin.Topology",
        admin.topology(Request::new(gunnar_proto::TopologyRequest {}))
    );
    // **The tunnel address of an enrolled box** (2026-09-21, T17): server
    // Admin, so a caller with no credential is refused before the plane.
    refuses!(
        "Admin.SetTunnelAddress",
        admin.set_tunnel_address(Request::new(gunnar_proto::AssignTunnelAddressRequest {
            address: "10.99.0.2/24".into(),
        }))
    );
    // A twin's replication certificate (lane T9): it mints a credential into
    // the stream that carries every sealed secret, so a caller with nothing
    // gets nothing — not even the seam's own refusals.
    refuses!(
        "Admin.IssueTwinCertificate",
        admin.issue_twin_certificate(Request::new(gunnar_proto::IssueTwinCertificateRequest {
            principal: "twin-box".into(),
            csr_pem: String::new(),
        }))
    );
    refuses!(
        "Admin.VerifyChain",
        admin.verify_chain(Request::new(VerifyChainRequest {
            repo: Some(RepoPath::new(NS, REPO)),
            ..Default::default()
        }))
    );
    // **The generic server settings** (owner 2026-10-26): server Admin, so a caller with
    // nothing reads and writes nothing.
    refuses!(
        "Admin.GetSetting",
        admin.get_setting(Request::new(gunnar_proto::GetSettingRequest {
            name: "console.a-switch".into()
        }))
    );
    refuses!(
        "Admin.SetSetting",
        admin.set_setting(Request::new(gunnar_proto::SetSettingRequest {
            name: "console.a-switch".into(),
            value: b"on".to_vec()
        }))
    );

    #[cfg(feature = "metered")]
    {
        use gunnar_proto::entitlement_server::Entitlement as _;
        let entitlement = gunnar_grpc::EntitlementService::new(harness.ctx.clone());
        refuses!(
            "Entitlement.Get",
            entitlement.get(Request::new(GetEntitlementRequest {
                namespace: NS.into(),
            }))
        );
        refuses!(
            "Entitlement.List",
            entitlement.list(Request::new(ListEntitlementsRequest {}))
        );
        refuses!(
            "Entitlement.Set",
            entitlement.set(Request::new(SetEntitlementRequest { fact: None }))
        );

        // **`Billing.*`** (2026-09-17). Driven here for the same reason
        // everything else is: an RPC absent from this sweep is an RPC nothing
        // checks the deny-by-default behaviour of. The scope check that makes
        // `MintActorTicket` safe is in `tests/billing.rs`; what this asserts is
        // the floor under it — with no verified caller at all, neither verb
        // answers, and in particular the one that SIGNS does not sign.
        use gunnar_proto::billing_server::Billing as _;
        let billing = gunnar_grpc::BillingService::new(harness.ctx.clone());
        refuses!(
            "Billing.MintActorTicket",
            billing.mint_actor_ticket(Request::new(gunnar_proto::MintActorTicketRequest {
                namespace: NS.into(),
                purpose: String::new(),
                // An empty purpose is a renewal, and a renewal names no caps.
                // The shape has to be a request the handler would otherwise
                // ACCEPT, or the refusal this sweep asserts could be about the
                // arguments rather than about the missing credential.
                caps: Default::default(),
            }))
        );
        refuses!(
            "Billing.ApplianceIdentity",
            billing.appliance_identity(Request::new(gunnar_proto::ApplianceIdentityRequest {}))
        );
        // **`Sealed.*`** (2026-10-09): a slot is its owner's alone; a caller
        // with nothing reads and writes nothing. The owner rule itself is in
        // `service::sealed`'s unit tests.
        use gunnar_proto::sealed_server::Sealed as _;
        let sealed = gunnar_grpc::SealedService::new(harness.ctx.clone());
        refuses!(
            "Sealed.Put",
            sealed.put(Request::new(gunnar_proto::SealedPutRequest {
                name: "slot".into(),
                value: b"x".to_vec(),
                owner: String::new()
            }))
        );
        refuses!(
            "Sealed.Get",
            sealed.get(Request::new(gunnar_proto::SealedGetRequest {
                name: "slot".into()
            }))
        );
    }

    let usage = harness.usage();
    refuses!(
        "Usage.Read",
        usage.read(Request::new(ReadUsageRequest {
            namespace: NS.into(),
        }))
    );
    refuses!(
        "Usage.ReadAll",
        usage.read_all(Request::new(ReadAllUsageRequest {}))
    );
    refuses!(
        "Usage.Activity",
        usage.activity(Request::new(ActivityRequest {
            namespace: NS.into(),
            since_unix_ms: 0,
            until_unix_ms: 0,
        }))
    );
    refuses!(
        "Usage.AdmitPlan",
        usage.admit_plan(Request::new(gunnar_proto::AdmitPlanRequest {
            namespace: NS.into(),
        }))
    );
    // Many namespaces' meters in one call (2026-10-06): `ReadAll`'s grant.
    let usage_batch = harness.usage_batch();
    refuses!(
        "UsageBatch.ReadMany",
        usage_batch.read_many(Request::new(gunnar_proto::ReadManyUsageRequest::default()))
    );

    let mirror = harness.mirror();
    refuses!(
        "Mirror.Add",
        mirror.add(Request::new(AddMirrorRequest {
            repo: path(),
            remote_url: "ssh://example/x.git".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Mirror.Sync",
        mirror.sync(Request::new(SyncMirrorRequest { repo: path() }))
    );
    refuses!(
        "Mirror.Status",
        mirror.status(Request::new(MirrorStatusRequest::default()))
    );
    // **The three replication RPCs, added with the soft-delete work and never driven here.**
    //
    // They were never a HOLE — `MirrorServer` is mounted `with_interceptor`,
    // and only `Bootstrap` sits outside the gate — but nothing verified that,
    // so the guarantee was unchecked for the RPCs that decide WHERE a copy of
    // every object gets shipped. The count assertion at the end of this test is
    // what caught it, which is that assertion earning its place: a bare number
    // nobody reconciles against the protos would have let five RPCs sit
    // undriven indefinitely.
    refuses!(
        "Mirror.SetReplication",
        mirror.set_replication(Request::new(SetReplicationRequest::default()))
    );
    refuses!(
        "Mirror.RemoveReplication",
        mirror.remove_replication(Request::new(RemoveReplicationRequest {
            name: "standby".into(),
        }))
    );
    refuses!(
        "Mirror.DemoteReplication",
        mirror.demote_replication(Request::new(DemoteReplicationRequest {}))
    );
    refuses!(
        "Mirror.SetRole",
        mirror.set_role(Request::new(SetRoleRequest {
            role: ReplicationRole::Standby as i32,
        }))
    );
    refuses!(
        "Mirror.AbdicateReplication",
        mirror.abdicate_replication(Request::new(AbdicateReplicationRequest {}))
    );
    refuses!(
        "Mirror.ReplicationStatus",
        mirror.replication_status(Request::new(ReplicationStatusRequest {
            include_refs: false
        }))
    );
    refuses!(
        "Mirror.Remove",
        mirror.remove(Request::new(RemoveMirrorRequest { repo: path() }))
    );
    refuses!(
        "Mirror.SetCredential",
        mirror.set_credential(Request::new(SetMirrorCredentialRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Mirror.ResyncReplication",
        mirror.resync_replication(Request::new(ResyncReplicationRequest {
            name: "standby".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Mirror.FlushReplication",
        mirror.flush_replication(Request::new(FlushReplicationRequest::default()))
    );
    refuses!(
        "Mirror.LogHead",
        mirror.log_head(Request::new(gunnar_proto::LogHeadRequest::default()))
    );
    refuses!(
        "Mirror.LogFetch",
        mirror.log_fetch(Request::new(gunnar_proto::LogFetchRequest {
            from: 1,
            ..Default::default()
        }))
    );

    let forge = harness.forge();
    let repo = || format!("{NS}/{REPO}");
    refuses!(
        "Forge.Append",
        forge.append(Request::new(AppendEventRequest {
            event: Some(SignedForgeEvent {
                event: Some(gunnar_proto::ForgeEvent {
                    repo: repo(),
                    pull: 1,
                    actor: "someone".into(),
                    at_unix_ms: 0,
                    kind: Some(gunnar_proto::forge_event::Kind::Commented(
                        gunnar_proto::Commented {
                            body: "hi".into(),
                            ..Default::default()
                        }
                    )),
                }),
                key_fingerprint: "SHA256:nope".into(),
                signature: String::new(),
                attested: None,
            }),
        }))
    );
    refuses!(
        "Forge.GetReview",
        forge.get_review(Request::new(GetReviewRequest {
            repo: repo(),
            pull: 1,
        }))
    );
    refuses!(
        "Forge.ListReviews",
        forge.list_reviews(Request::new(ListReviewsRequest { repo: repo() }))
    );
    refuses!(
        "Forge.GetIssue",
        forge.get_issue(Request::new(GetIssueRequest {
            repo: repo(),
            number: 1,
        }))
    );
    refuses!(
        "Forge.ListIssues",
        forge.list_issues(Request::new(ListIssuesRequest { repo: repo() }))
    );

    // The three `Webhooks.*` RPCs, and the service was not even built here
    // before GUN-023: an entire mounted service outside the sweep.
    let webhooks = harness.webhooks();
    refuses!(
        "Webhooks.AddWebhook",
        webhooks.add_webhook(Request::new(AddWebhookRequest {
            name: "ci".into(),
            url: "https://example.invalid/hook".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Webhooks.ListWebhooks",
        webhooks.list_webhooks(Request::new(ListWebhooksRequest {}))
    );
    refuses!(
        "Webhooks.RemoveWebhook",
        webhooks.remove_webhook(Request::new(RemoveWebhookRequest { name: "ci".into() }))
    );

    // The five `PullRequests.*` RPCs. They authorise in `service::pulls` and
    // were missing from `auth::METHODS` until the two designs were merged, so
    // this sweep never drove them — the exact hole the census exists to close.
    let pulls = harness.pulls();
    refuses!(
        "PullRequests.Create",
        pulls.create(Request::new(CreatePullRequestRequest {
            repo: path(),
            title: "t".into(),
            description: String::new(),
            source_ref: "refs/heads/topic".into(),
            target_ref: "refs/heads/main".into(),
        }))
    );
    refuses!(
        "PullRequests.List",
        pulls.list(Request::new(ListPullRequestsRequest {
            repo: path(),
            state: 0,
        }))
    );
    refuses!(
        "PullRequests.Get",
        pulls.get(Request::new(GetPullRequestRequest {
            repo: path(),
            number: 1,
        }))
    );
    refuses!(
        "PullRequests.Close",
        pulls.close(Request::new(ClosePullRequestRequest {
            repo: path(),
            number: 1,
        }))
    );
    refuses!(
        "PullRequests.Merge",
        pulls.merge(Request::new(MergePullRequestRequest {
            repo: path(),
            number: 1,
        }))
    );

    // **The twelve that were served and censused nowhere** (2026-09-11,
    // GUN-023/GUN-024). Each one authorises in its handler — every one was read
    // — so none of them was an open door. What was missing is this: an RPC
    // absent from `auth::METHODS` is an RPC this sweep has never once driven,
    // so nothing would notice if one stopped authorising. That is the same hole
    // the `PullRequests.*` note above records, five RPCs and one release
    // earlier, which is what makes the hand-kept census the defect rather than
    // the omissions. `tests/census.rs` derives the list from the protos now.
    use gunnar_proto::webhooks_server::Webhooks as _;
    refuses!(
        "Access.AddGroupMember",
        access.add_group_member(Request::new(AddGroupMemberRequest {
            group_id: "team".into(),
            member_id: "someone".into(),
            signed: None,
        }))
    );
    refuses!(
        "Access.RemoveGroupMember",
        access.remove_group_member(Request::new(RemoveGroupMemberRequest {
            group_id: "team".into(),
            member_id: "someone".into(),
            signed: None,
        }))
    );
    refuses!(
        "Access.ListGroupMembers",
        access.list_group_members(Request::new(ListGroupMembersRequest {
            group_id: "team".into(),
        }))
    );
    refuses!(
        "Access.ListGroupMembershipEvents",
        access.list_group_membership_events(Request::new(ListGroupMembershipEventsRequest {
            group_id: String::new(),
        }))
    );
    refuses!(
        "Forge.GetIssue",
        forge.get_issue(Request::new(GetIssueRequest {
            repo: repo(),
            number: 1,
        }))
    );
    refuses!(
        "Forge.ListIssues",
        forge.list_issues(Request::new(ListIssuesRequest { repo: repo() }))
    );
    refuses!(
        "Mirror.SetCredential",
        mirror.set_credential(Request::new(SetMirrorCredentialRequest {
            repo: path(),
            ..Default::default()
        }))
    );
    refuses!(
        "Mirror.ResyncReplication",
        mirror.resync_replication(Request::new(ResyncReplicationRequest {
            name: "standby".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Mirror.LogHead",
        mirror.log_head(Request::new(gunnar_proto::LogHeadRequest::default()))
    );
    refuses!(
        "Mirror.LogFetch",
        mirror.log_fetch(Request::new(gunnar_proto::LogFetchRequest {
            from: 1,
            ..Default::default()
        }))
    );
    // **The two the sweep had drifted away from** (2026-09-16, lane AUDIT-TWIN). The drift
    // assertion at the end of this test was RED on master for `Mirror.SnapshotFetch` — the verb
    // that streams every sealed secret and every object on the box — and
    // `no_rpc_is_reachable_without_a_caller_and_none_is_anonymous_scoped` was RED for
    // `Mirror.LogRetain`, which was in neither `auth::METHODS` nor `UNAUTHENTICATED_METHODS` at
    // all. Both authorise in their handlers and always did; neither had ever been DRIVEN
    // without a credential, which is the whole point of this sweep and the third time that
    // omission has been found here.
    refuses!(
        "Mirror.SnapshotFetch",
        mirror.snapshot_fetch(Request::new(gunnar_proto::SnapshotFetchRequest::default()))
    );
    refuses!(
        "Mirror.LogRetain",
        mirror.log_retain(Request::new(gunnar_proto::LogRetainRequest {
            from: 1,
            allow_past_twin: false
        }))
    );
    refuses!(
        "Usage.Activity",
        usage.activity(Request::new(ActivityRequest {
            namespace: NS.into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Usage.AdmitPlan",
        usage.admit_plan(Request::new(gunnar_proto::AdmitPlanRequest {
            namespace: NS.into(),
        }))
    );
    let webhooks = harness.webhooks();
    refuses!(
        "Webhooks.AddWebhook",
        webhooks.add_webhook(Request::new(AddWebhookRequest {
            name: "hook".into(),
            url: "https://example/hook".into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Webhooks.ListWebhooks",
        webhooks.list_webhooks(Request::new(ListWebhooksRequest {}))
    );
    refuses!(
        "Webhooks.RemoveWebhook",
        webhooks.remove_webhook(Request::new(RemoveWebhookRequest {
            name: "hook".into(),
        }))
    );

    // The owner's enterprise list (2026-10-05): the account admin's audit
    // log, the operator's aggregates and look, the name filter.
    {
        use gunnar_proto::enterprise_server::Enterprise as _;
        let e = harness.enterprise();
        refuses!(
            "Enterprise.AuditLog",
            e.audit_log(Request::new(AuditLogRequest {
                account: NS.into(),
                ..Default::default()
            }))
        );
        refuses!(
            "Enterprise.ExportAuditLog",
            e.export_audit_log(Request::new(ExportAuditLogRequest {
                account: NS.into(),
                ..Default::default()
            }))
        );
        refuses!(
            "Enterprise.Aggregates",
            e.aggregates(Request::new(AggregatesRequest {}))
        );
        refuses!(
            "Enterprise.RequestOperatorAccess",
            e.request_operator_access(Request::new(RequestOperatorAccessRequest {
                account: NS.into(),
                reason: "why".into(),
                minutes: 0
            }))
        );
        refuses!(
            "Enterprise.ListOperatorAccess",
            e.list_operator_access(Request::new(ListOperatorAccessRequest {
                account: NS.into()
            }))
        );
        refuses!(
            "Enterprise.ApproveOperatorAccess",
            e.approve_operator_access(Request::new(ApproveOperatorAccessRequest::default()))
        );
        refuses!(
            "Enterprise.RefuseOperatorAccess",
            e.refuse_operator_access(Request::new(RefuseOperatorAccessRequest::default()))
        );
        refuses!(
            "Enterprise.RevokeOperatorAccess",
            e.revoke_operator_access(Request::new(RevokeOperatorAccessRequest::default()))
        );
        refuses!(
            "Enterprise.CheckName",
            e.check_name(Request::new(CheckNameRequest {
                kind: "repo".into(),
                name: "x".into(),
                account: String::new()
            }))
        );
        refuses!(
            "Enterprise.GetNameLists",
            e.get_name_lists(Request::new(GetNameListsRequest {}))
        );
        refuses!(
            "Enterprise.SetNameLists",
            e.set_name_lists(Request::new(SetNameListsRequest::default()))
        );
        refuses!(
            "Enterprise.ListNameReviews",
            e.list_name_reviews(Request::new(ListNameReviewsRequest { account: NS.into() }))
        );
        refuses!(
            "Enterprise.DecideNameReview",
            e.decide_name_review(Request::new(DecideNameReviewRequest::default()))
        );
    }

    // Company SSO (2026-10-05): the account admin's verbs and the relay's.
    let sso = harness.sso();
    refuses!(
        "Sso.SetProvider",
        sso.set_provider(Request::new(SetProviderRequest::default()))
    );
    refuses!(
        "Sso.GetProvider",
        sso.get_provider(Request::new(GetProviderRequest {
            account: NS.into(),
            ..Default::default()
        }))
    );
    refuses!(
        "Sso.RemoveProvider",
        sso.remove_provider(Request::new(RemoveProviderRequest { account: NS.into() }))
    );
    refuses!(
        "Sso.CreateScimToken",
        sso.create_scim_token(Request::new(CreateScimTokenRequest {
            account: NS.into(),
            label: String::new(),
            ..Default::default()
        }))
    );
    refuses!(
        "Sso.ListScimTokens",
        sso.list_scim_tokens(Request::new(ListScimTokensRequest { account: NS.into() }))
    );
    refuses!(
        "Sso.RevokeScimToken",
        sso.revoke_scim_token(Request::new(RevokeScimTokenRequest::default()))
    );
    refuses!(
        "Sso.ListDirectory",
        sso.list_directory(Request::new(ListDirectoryRequest { account: NS.into() }))
    );
    refuses!(
        "Sso.SetUserActive",
        sso.set_user_active(Request::new(SetUserActiveRequest::default()))
    );
    refuses!(
        "Sso.ImportUsers",
        sso.import_users(Request::new(ImportUsersRequest::default()))
    );
    refuses!(
        "Sso.ImportStatus",
        sso.import_status(Request::new(ImportStatusRequest::default()))
    );
    refuses!(
        "Sso.Discover",
        sso.discover(Request::new(DiscoverRequest {
            hint: "bob@acme.example".into()
        }))
    );
    refuses!(
        "Sso.Begin",
        sso.begin(Request::new(BeginRequest::default()))
    );
    refuses!(
        "Sso.Finish",
        sso.finish(Request::new(FinishRequest::default()))
    );
    refuses!(
        "Sso.SecondFactor",
        sso.second_factor(Request::new(SecondFactorRequest::default()))
    );
    refuses!("Sso.Scim", sso.scim(Request::new(ScimRequest::default())));
    refuses!(
        "Sso.VerifyDomain",
        sso.verify_domain(Request::new(VerifyDomainRequest {
            account: NS.into(),
            domain: "acme.example".into()
        }))
    );
    refuses!(
        "Sso.RebindUsers",
        sso.rebind_users(Request::new(RebindUsersRequest {
            account: NS.into(),
            ..Default::default()
        }))
    );

    // The sweep and the crate's own list of audited methods must agree, so a
    // new RPC cannot be added without landing in both.
    assert_eq!(
        covered,
        audited_methods(),
        "the deny-by-default sweep and gunnar_grpc::auth::METHODS have drifted"
    );
    // **The arithmetic, not a bare number.** The protos declare `DECLARED_RPCS`; two of
    // them (`Bootstrap.Enroll`, `Bootstrap.Status`) are served with no
    // credential on purpose and are listed in `UNAUTHENTICATED_METHODS` with
    // the reason. Everything else must be swept here.
    //
    // Written as the SUM so the next RPC cannot land quietly: adding one moves
    // the declared count, and the only ways to keep this green are to drive it
    // above or to say out loud, in UNAUTHENTICATED_METHODS, that it needs no
    // credential. A hardcoded 56 would have been satisfied by driving 56 of
    // anything.
    let declared = declared_rpcs();
    let swept: BTreeSet<String> = covered
        .iter()
        .chain(gunnar_grpc::UNAUTHENTICATED_METHODS)
        .map(|m| (*m).to_owned())
        .collect();
    assert_eq!(
        declared.difference(&swept).collect::<Vec<_>>(),
        Vec::<&String>::new(),
        "these RPCs are declared in the protos and this sweep never drove them, so nothing \
         checks that they refuse an unauthenticated caller. Drive them here, or say out loud \
         in UNAUTHENTICATED_METHODS that they need no credential."
    );
}

#[tokio::test]
async fn the_escape_hatch_serves_what_the_default_refuses() {
    // The escape hatch must actually be an escape hatch; otherwise the
    // sweep above would pass against a server that is simply broken.
    let plane = Arc::new(MemoryControlPlane::with_repo(NS, REPO));
    let control: Arc<dyn ControlPlane> = plane.clone();
    let ctx = ServiceContext::new(
        control,
        Arc::new(EventBus::new(64)),
        ServerConfig::default().insecure_allow_all_for_local_development(),
    );
    let served = AdminService::new(ctx)
        .health(Request::new(HealthRequest {}))
        .await
        .expect("the escape hatch serves an unauthenticated caller")
        .into_inner();
    assert!(
        !served.grpc_authentication_enabled,
        "and says so on the wire"
    );
}

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

/// Captures `tracing` events emitted on [`AUDIT_TARGET`], with their fields.
#[derive(Clone, Default)]
struct AuditLog(Arc<Mutex<Vec<String>>>);

impl AuditLog {
    fn lines(&self) -> Vec<String> {
        self.0.lock().unwrap().clone()
    }
}

struct AuditLayer(AuditLog);

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for AuditLayer {
    fn on_event(
        &self,
        event: &tracing::Event<'_>,
        _ctx: tracing_subscriber::layer::Context<'_, S>,
    ) {
        if event.metadata().target() != AUDIT_TARGET {
            return;
        }
        struct Visit(String);
        impl tracing::field::Visit for Visit {
            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
                self.0.push_str(&format!("{}={:?} ", field.name(), value));
            }
        }
        let mut visit = Visit(String::new());
        event.record(&mut visit);
        self.0 .0.lock().unwrap().push(visit.0);
    }
}

fn capture_audit() -> (AuditLog, impl tracing::Subscriber + Send + Sync) {
    let log = AuditLog::default();
    let subscriber = tracing_subscriber::registry().with(AuditLayer(log.clone()));
    (log, subscriber)
}

#[test]
fn a_mutation_emits_an_audit_record_naming_the_caller() {
    // Every gRPC service file used to have zero log lines, so the one
    // surface that most needed an audit trail had none at all.
    let (log, subscriber) = capture_audit();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .expect("runtime");

    with_default(subscriber, || {
        let harness = Harness::new();
        let (admin_id, admin_key) = harness.principal("admin");
        harness.plane.grant_server(&admin_id, GrantLevel::Admin);
        let admin = harness.caller_for(&admin_key, "nonce-admin-000005");
        log.0.lock().unwrap().clear();

        runtime.block_on(async {
            harness
                .repos()
                .create(harness.authed(
                    &admin,
                    CreateRepoRequest {
                        path: Some(RepoPath::new(NS, "audited")),
                        ..Default::default()
                    },
                ))
                .await
                .expect("created");
        });

        let lines = log.lines();
        let record = lines
            .iter()
            .find(|line| line.contains("Repos.Create"))
            .unwrap_or_else(|| panic!("no audit record for the mutation; captured {lines:#?}"));
        assert!(
            record.contains(&format!("principal={admin_id} ")),
            "the record does not name the caller: {record}"
        );
        assert!(
            record.contains("decision=\"allow\""),
            "the record does not carry the decision: {record}"
        );
        assert!(
            record.contains(&format!("scope=namespace:{NS} ")),
            "the record does not carry the scope: {record}"
        );
        assert!(
            record.contains("auth_mode=\"required\""),
            "the record does not say which mode served it: {record}"
        );
    });
}

#[test]
fn a_call_refused_before_the_handler_leaves_no_call_record() {
    // The other half of the audit claim: a request the gate turned away must
    // not produce a `grpc.call` record, because nothing was called. If it did,
    // the trail would show admin activity that never happened.
    let (log, subscriber) = capture_audit();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .expect("runtime");

    with_default(subscriber, || {
        let harness = Harness::new();
        log.0.lock().unwrap().clear();

        // (a) refused by the interceptor, before any handler exists.
        let mut gate = harness.gate();
        let refused = <AuthGate as tonic::service::Interceptor>::call(&mut gate, Request::new(()));
        assert_eq!(refused.err().map(|s| s.code()), Some(Code::Unauthenticated));

        // (b) and the handler, driven directly, also refuses.
        runtime.block_on(async {
            harness
                .repos()
                .create(Request::new(CreateRepoRequest {
                    path: Some(RepoPath::new(NS, "never")),
                    ..Default::default()
                }))
                .await
                .expect_err("refused");
        });

        let lines = log.lines();
        assert!(
            lines.iter().any(|l| l.contains("grpc.auth.refused")),
            "the refusal itself must be recorded; captured {lines:#?}"
        );
        assert!(
            !lines.iter().any(|l| l.contains("grpc.call")),
            "a refused call produced a control-plane call record: {lines:#?}"
        );
        assert!(
            !lines.iter().any(|l| l.contains("Repos.Create")),
            "a refused call named a method it never reached: {lines:#?}"
        );
    });

    // …and the capture itself is not broken: an allowed call DOES produce one.
    // Without this, the two assertions above would pass against a subscriber
    // that captured nothing at all (measured, not asserted).
    let (log, subscriber) = capture_audit();
    with_default(subscriber, || {
        let harness = Harness::new();
        let (admin_id, admin_key) = harness.principal("admin");
        harness.plane.grant_server(&admin_id, GrantLevel::Admin);
        let admin = harness.caller_for(&admin_key, "nonce-admin-000006");
        runtime.block_on(async {
            harness
                .admin()
                .stats(harness.authed(&admin, StatsRequest {}))
                .await
                .expect("stats");
        });
        assert!(
            log.lines().iter().any(|l| l.contains("grpc.call")),
            "the audit capture never sees anything, so the absence assertions above are \
             vacuous"
        );
    });
}

// ───────────────────────────────────────────────────────────────────────────
// The wiring: the gate is actually mounted
// ───────────────────────────────────────────────────────────────────────────

#[test]
fn the_gate_attaches_a_caller_to_the_request_it_lets_through() {
    // `router()` mounts this interceptor in front of all seven services. If it
    // did not attach a `Caller`, every handler would refuse a perfectly good
    // credential, and if it attached one without verifying, everything above
    // would be theatre. Both halves, on the real type.
    let harness = Harness::new();
    let (principal_id, key) = harness.principal("alice");
    let mut gate = harness.gate();

    let credential =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-wiring-00001", "Admin.Health")
            .expect("sign");
    let mut request = Request::new(());
    request.metadata_mut().insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );
    // What `RequestSpanLayer` puts there from the router's path.
    request
        .extensions_mut()
        .insert(gunnar_grpc::auth::CalledMethod("Admin.Health".to_owned()));
    let passed = <AuthGate as tonic::service::Interceptor>::call(&mut gate, request)
        .expect("a valid credential must pass the gate");
    let caller = passed
        .extensions()
        .get::<Caller>()
        .expect("the gate must attach the caller it verified");
    assert_eq!(caller.principal_id, principal_id);

    assert_eq!(
        <AuthGate as tonic::service::Interceptor>::call(&mut gate, Request::new(()))
            .err()
            .map(|s| s.code()),
        Some(Code::Unauthenticated),
        "the gate let a request with no credential through"
    );
}

// ───────────────────────────────────────────────────────────────────────────
// The method binding (GUN-001)
// ───────────────────────────────────────────────────────────────────────────
//
// The finding: the signed bytes were `fingerprint‖timestamp‖nonce`, so a
// credential lifted off a plaintext control plane could be re-sent at any other
// method — `Admin.Health` in, `Access.AddKey` out, permanent admin, and the
// audit line names the victim. Three things have to hold for that to be closed
// and each is asserted in both directions here:
//
//   1. a bound credential presented at another method is refused;
//   2. the method the gate checks against comes from the ROUTER's path, which
//      means `RequestSpanLayer` has to put it somewhere the interceptor can
//      read — the one mechanism the old comment said did not exist;
//   3. the method-blind v1 credential is refused at every method (5.3.3,
//      OWNER 2026-10-10; it used to be served until an opt-in flag).

/// The relay, on the real gate: sign for `Admin.Health`, present at
/// `Access.AddKey`.
///
/// Mutation-proven 2026-09-11: with the `signed != called` arm of
/// `authenticate_bound` deleted, this test goes green on the `Ok(_)` it is
/// asserting against — the credential verifies perfectly well, it is simply
/// being spent somewhere it was not issued for, which is the whole finding.
#[test]
fn a_credential_signed_for_one_method_is_refused_at_another() {
    let harness = Harness::new();
    let (principal_id, key) = harness.principal("alice");
    let gate = harness.gate();

    let credential =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-bound-000001", "Admin.Health")
            .expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );

    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Access.AddKey")),
        Err(AuthFailure::MethodMismatch {
            signed: "Admin.Health".to_owned(),
            called: "Access.AddKey".to_owned(),
        }),
        "a credential issued for Admin.Health was accepted at Access.AddKey"
    );

    // The other direction, and it matters: a gate that refused everything
    // would pass the assertion above without binding anything. The SAME
    // credential, at the method it names, authenticates — and names the right
    // principal.
    let caller = gate
        .authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
        .expect("the credential must be accepted at the method it was signed for");
    assert_eq!(caller.principal_id, principal_id);

    // …and the refusal was not free: a mismatch must NOT spend the nonce, or
    // an attacker who cannot use a captured credential could still burn it and
    // make the honest caller's own call fail.
    let replayed =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-bound-000002", "Admin.Health")
            .expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        replayed.to_header_value().parse().unwrap(),
    );
    assert!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Access.AddKey"))
            .is_err(),
        "the relay must be refused"
    );
    assert!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
            .is_ok(),
        "a refused relay spent the honest caller's nonce"
    );
}

/// The method is signed, not merely carried: rewriting the header's method
/// field breaks the signature rather than moving the credential.
#[test]
fn the_method_cannot_be_edited_out_of_a_bound_credential() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();

    let mut credential =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-rewrite-0001", "Admin.Health")
            .expect("sign");
    credential.method = Some("Access.AddKey".to_owned());
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Access.AddKey")),
        Err(AuthFailure::NotVerified),
        "the method was rewritten and the signature still verified"
    );

    // And dropping the method altogether — presenting the v2 signature under
    // the v1 scheme — does not downgrade into the unbound path either: that
    // path refuses everything, before the signature is even read.
    let mut credential =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-rewrite-0002", "Admin.Health")
            .expect("sign");
    credential.method = None;
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Access.AddKey")),
        Err(AuthFailure::Unbound),
        "a v2 signature presented as v1 was accepted, which is a downgrade"
    );
}

/// A bound credential with no method in scope is refused rather than served.
///
/// This is the fail-closed half. `authenticate` — the entry point with no
/// request behind it — cannot check the binding, and "cannot check" must never
/// read as "passes".
#[test]
fn a_bound_credential_is_refused_when_the_method_is_not_in_scope() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();

    let credential =
        gunnar_grpc::sign_credential_for(&key, now_ms(), "nonce-noscope-001", "Admin.Health")
            .expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        credential.to_header_value().parse().unwrap(),
    );
    assert!(
        matches!(
            gate.authenticate(&metadata, now_ms()),
            Err(AuthFailure::MethodMismatch { .. })
        ),
        "a bound credential was served by the path that cannot check the binding"
    );
}

/// **The shipped gate refuses the method-blind v1 credential at every method**
/// (OWNER 2026-10-10: require_bound ON in 5.3.3; stinky row 2, GUN-001).
///
/// The default gate — no builder call, no flag — handed a valid, fresh,
/// correctly signed `gunnar-sshsig-v1` header from a registered key, at every
/// method the control plane serves, plus the interceptor path with the router's
/// `CalledMethod` and the no-method-in-scope path. Every one is `Unbound`, and
/// each refusal is counted against the caller and method it was presented at,
/// none as served.
#[test]
fn the_shipped_gate_refuses_a_method_blind_credential_at_every_method() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness.gate();
    let v1 = |nonce: &str| {
        let credential = sign_credential(&key, now_ms(), nonce).expect("sign");
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert(
            CREDENTIAL_METADATA_KEY,
            credential.to_header_value().parse().unwrap(),
        );
        metadata
    };

    let mut served = Vec::new();
    for (i, method) in gunnar_grpc::auth::METHODS.iter().enumerate() {
        let answer = gate.authenticate_bound(
            &v1(&format!("nonce-v1-every-{i:06}")),
            now_ms(),
            Some(method),
        );
        if answer != Err(AuthFailure::Unbound) {
            served.push(format!("{method}: {answer:?}"));
        }
    }
    assert!(
        served.is_empty(),
        "the shipped gate did not refuse a method-blind credential at these methods: {served:#?}"
    );
    assert_eq!(
        gate.authenticate(&v1("nonce-v1-no-method-01"), now_ms()),
        Err(AuthFailure::Unbound),
        "with no method in scope a v1 credential must still be refused"
    );

    // The interceptor, with the method the router is about to dispatch to.
    let mut request = Request::new(());
    *request.metadata_mut() = v1("nonce-v1-intercept-1");
    request
        .extensions_mut()
        .insert(gunnar_grpc::auth::CalledMethod("Access.AddKey".to_owned()));
    let mut intercepting = gate.clone();
    let status = <AuthGate as tonic::service::Interceptor>::call(&mut intercepting, request)
        .expect_err("the interceptor let a method-blind credential through to Access.AddKey");
    assert_eq!(status.code(), Code::Unauthenticated);
    assert!(
        status.message().contains(gunnar_grpc::CREDENTIAL_SCHEME),
        "{}",
        status.message()
    );

    let report = gate.unbound_log().report();
    assert_eq!(
        report.accepted, 0,
        "nothing method-blind may be counted as served"
    );
    assert_eq!(
        report.refused as usize,
        gunnar_grpc::auth::METHODS.len() + 2,
        "every refusal must be counted"
    );
}

/// `/gunnar.v1.Access/AddKey` is `Access.AddKey`, and nothing else is anything.
#[test]
fn a_path_names_the_method_the_credential_names() {
    use gunnar_grpc::method_name_from_path;

    assert_eq!(
        method_name_from_path("/gunnar.v1.Access/AddKey").as_deref(),
        Some("Access.AddKey")
    );
    // Every audited method must be derivable from the path its service is
    // mounted at, or a credential could never match one of them.
    for method in audited_methods() {
        let (service, verb) = method.split_once('.').expect("Service.Method");
        assert_eq!(
            method_name_from_path(&format!("/gunnar.v1.{service}/{verb}")).as_deref(),
            Some(method),
            "the path for {method} does not round-trip to its own name"
        );
    }
    // …and the shapes that are not a gRPC method are None, so a credential
    // claiming one can never be "matched" against a route that does not exist.
    for junk in [
        "/",
        "",
        "/gunnar.v1.Access",
        "/gunnar.v1.Access/AddKey/extra",
        "/gunnar.v1.Access/",
    ] {
        assert_eq!(
            method_name_from_path(junk),
            None,
            "{junk:?} was read as a method name"
        );
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The replay guard does not forget (GUN-009)
// ───────────────────────────────────────────────────────────────────────────

/// **The finding.** A credential inside the freshness window is never pushed
/// out of the guard to make room for a newer one.
///
/// The guard used to run `while order.len() >= capacity { pop_front() }`, and
/// the oldest entry inside a five-minute window is still perfectly
/// presentable. An enrolled principal — a key and nothing else — could mint
/// their own credentials at ~27/s for five minutes, evict a captured one, and
/// replay it. That guard was the whole compensation for a credential that
/// bound no method, so GUN-001 and this were one hole.
///
/// Mutation-proven 2026-09-11: restore the eviction loop and the first
/// credential below is accepted a SECOND time, which is the replay.
#[test]
fn a_fresh_credential_is_never_evicted_to_make_room() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    // Four is the whole guard, so "the attacker fills it" is four calls
    // rather than 8192.
    let guard = gunnar_grpc::ReplayGuard::new(4);
    let now = now_ms();

    let captured = sign_credential(&key, now, "nonce-captured-0001").expect("sign");
    assert_eq!(
        guard.accept(&captured, now),
        gunnar_grpc::Accepted::Fresh,
        "the first use of a credential must be accepted"
    );

    // The attacker's own, valid, freshly signed, one after another. Nothing
    // here is forged: this is what any enrolled principal can do.
    for seq in 0..3 {
        let mine = sign_credential(&key, now, &format!("nonce-filler-{seq:07}")).expect("sign");
        assert_eq!(guard.accept(&mine, now), gunnar_grpc::Accepted::Fresh);
    }

    // The guard is now full of entries that are all still inside the window.
    // One more must be REFUSED rather than make room.
    let overflow = sign_credential(&key, now, "nonce-overflow-001").expect("sign");
    assert_eq!(
        guard.accept(&overflow, now),
        gunnar_grpc::Accepted::Full,
        "a full guard must refuse rather than forget something still presentable"
    );

    // …and the captured credential is still remembered, which is the whole
    // point. Before the fix this line read `Fresh`.
    assert_eq!(
        guard.accept(&captured, now),
        gunnar_grpc::Accepted::Replayed,
        "a captured credential was forgotten and could be replayed"
    );
}

/// The guard still forgets what it is allowed to forget, so a full one is a
/// transient state and not a permanently closed door.
///
/// Without this the fix above would be satisfied by a guard that refuses
/// everything after `capacity` credentials for the life of the process.
#[test]
fn the_guard_forgets_exactly_what_can_no_longer_be_presented() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let guard = gunnar_grpc::ReplayGuard::new(2);
    let now = now_ms();

    for seq in 0..2 {
        let filler = sign_credential(&key, now, &format!("nonce-window-{seq:07}")).expect("sign");
        assert_eq!(guard.accept(&filler, now), gunnar_grpc::Accepted::Fresh);
    }
    let blocked = sign_credential(&key, now, "nonce-blocked-0001").expect("sign");
    assert_eq!(guard.accept(&blocked, now), gunnar_grpc::Accepted::Full);

    // One window later, the two entries can no longer be presented at all —
    // `authenticate_bound` refuses them as stale before the guard is reached —
    // so forgetting them costs nothing, and the guard has room again.
    let later = now + MAX_CLOCK_SKEW_MS + 1;
    assert_eq!(guard.accept(&blocked, later), gunnar_grpc::Accepted::Fresh);
    assert_eq!(guard.len(), 1, "the expired pair must have been dropped");
}

/// A full guard is `RESOURCE_EXHAUSTED` and not `UNAUTHENTICATED`.
///
/// Nothing is wrong with the credential, and a client told "not authenticated"
/// rotates a key for no reason. The two answers are different facts about
/// different things and the status says which.
#[test]
fn a_full_guard_reports_the_server_and_not_the_caller() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let gate = harness
        .gate()
        .with_replay_guard(std::sync::Arc::new(gunnar_grpc::ReplayGuard::new(1)));
    let now = now_ms();

    for (seq, expected) in [(0, None), (1, Some(Code::ResourceExhausted))] {
        let credential = gunnar_grpc::sign_credential_for(
            &key,
            now,
            &format!("nonce-status-{seq:07}"),
            "Admin.Health",
        )
        .expect("sign");
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert(
            CREDENTIAL_METADATA_KEY,
            credential.to_header_value().parse().unwrap(),
        );
        let answer = gate.authenticate_bound(&metadata, now, Some("Admin.Health"));
        assert_eq!(
            answer.as_ref().err().map(|failure| failure.status().code()),
            expected,
            "credential {seq}"
        );
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Who still signs v1: the unbound-credential log
//
// Until 5.3.3 this log was the evidence for flipping an opt-in flag; with v1
// refused everywhere it is what names the client that just stopped working.
// Mutations, recorded because a green that has never been red is a comment
// (2026-09; the served-side rows went with the served path in 5.3.3):
//
// | mutation in `auth.rs` | what went red |
// |---|---|
// | the refused-side `note_unbound` removed | the refusal tests below (`refused` read 0) |
// | the roster's capacity check removed (unbounded) | `the_unbound_roster_is_bounded_…` (`distinct_dropped` stayed 0) |
// | the counters bounded along with the roster | the same test (`accepted` read 3 instead of 4) |
// ───────────────────────────────────────────────────────────────────────────

/// **A bound credential is served and leaves the log empty; a method-blind
/// one is refused, counted as refused, and named.**
///
/// Both directions, on ONE gate: a log that counted every credential would
/// name a fleet on v2 as still needing an update, and one that counted none
/// would leave the client that just broke unnamed. Presented at
/// `Access.AddKey`, the relay GUN-001 is about.
#[test]
fn a_bound_credential_is_not_logged_and_a_method_blind_one_is_refused_and_named() {
    let harness = Harness::new();
    let (_, key) = harness.principal("alice");
    let fingerprint = key
        .public_key()
        .fingerprint(ssh_key::HashAlg::Sha256)
        .to_string();
    let gate = harness.gate();

    // ── the bound credential: served, nothing to report ─────────────────
    let bound = sign_credential_for(&key, now_ms(), "nonce-bound-000001", "Admin.Health")
        .expect("sign a bound credential");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        bound.to_header_value().parse().unwrap(),
    );
    gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health"))
        .expect("a bound credential must be served");
    let report = gate.unbound_log().report();
    assert_eq!(
        report,
        gunnar_grpc::UnboundCredentialReport::default(),
        "a method-bound credential put something in the unbound log"
    );
    assert!(report.nothing_unbound_has_been_served());

    // ── the method-blind one: refused, counted, named ───────────────────
    let blind = sign_credential(&key, 1_700_000_000_000, "nonce-blind-000001").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        blind.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, 1_700_000_000_000, Some("Access.AddKey")),
        Err(AuthFailure::Unbound)
    );

    let report = gate.unbound_log().report();
    assert_eq!(
        report.accepted, 0,
        "a refused credential was counted as served"
    );
    assert_eq!(report.refused, 1, "the refusal was not counted");
    assert!(
        report.nothing_unbound_has_been_served(),
        "the verdict must follow the count, or the two readers disagree"
    );
    assert_eq!(report.last_seen_unix_ms, 1_700_000_000_000);
    assert_eq!(report.callers.len(), 1, "one caller, one method, one row");
    let row = &report.callers[0];
    assert!(row.refused);
    assert_eq!(row.fingerprint, fingerprint);
    assert_eq!(row.method, "Access.AddKey");
    assert_eq!(row.count, 1);
    assert!(
        row.principal_id.is_empty(),
        "the refusal happens before the key lookup, so no principal is known here — a row \
         that claimed one would be claiming the header's word for it"
    );
}

/// **The refusal comes before the key lookup, and after the clock.**
///
/// A forged method-blind credential (a key nobody registered) is refused as
/// `Unbound` — the server never looks the key up for a shape it serves to no
/// one — and is counted as refused: the counter answers "who is hitting the
/// closed door", junk included. A stale one is refused as `Stale` before the
/// binding is read, and is not counted at all. Neither is ever `accepted`.
#[test]
fn a_forged_method_blind_credential_is_refused_unread_and_a_stale_one_as_stale() {
    let harness = Harness::new();
    let (_, registered) = harness.principal("alice");
    let stranger = mint();
    let gate = harness.gate();

    let forged = sign_credential(&stranger, now_ms(), "nonce-forged-00001").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        forged.to_header_value().parse().unwrap(),
    );
    assert_eq!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::Unbound)
    );

    let stale =
        sign_credential(&registered, now_ms() - 3_600_000, "nonce-stale-000001").expect("sign");
    let mut metadata = tonic::metadata::MetadataMap::new();
    metadata.insert(
        CREDENTIAL_METADATA_KEY,
        stale.to_header_value().parse().unwrap(),
    );
    assert!(matches!(
        gate.authenticate_bound(&metadata, now_ms(), Some("Admin.Health")),
        Err(AuthFailure::Stale { .. })
    ));

    let report = gate.unbound_log().report();
    assert_eq!(report.accepted, 0, "nothing method-blind is ever served");
    assert_eq!(
        report.refused, 1,
        "the forged one is counted; the stale one never reached the binding"
    );
    assert_eq!(report.callers.len(), 1);
    assert!(report.callers[0].refused);
}

/// **A full roster never turns a non-zero count into an empty list.**
///
/// The roster is bounded and the counters are not, and that asymmetry is the
/// whole reason the flag decision can rest on `accepted == 0`: an unbounded
/// roster is a memory hole anyone with a key can fill, and a bounded *counter*
/// would read "nobody sends v1" on the one busy server where it is wrong.
///
/// Driven at capacity 2 rather than at the shipped 64 so the arithmetic is
/// visible: three distinct pairs, two listed, one dropped, three counted.
#[test]
fn the_unbound_roster_is_bounded_and_the_counter_is_not() {
    let log = gunnar_grpc::UnboundCredentialLog::new(2);
    assert!(log.record("SHA256:a", "alice", "Admin.Health", false, 10));
    assert!(log.record("SHA256:b", "bob", "Admin.Health", false, 20));
    // The same pair again: counted, not new, and not a second row.
    assert!(!log.record("SHA256:a", "alice", "Admin.Health", false, 30));
    // A third distinct pair, past the bound.
    assert!(log.record("SHA256:c", "carol", "Admin.Health", false, 40));

    let report = log.report();
    assert_eq!(
        report.accepted, 4,
        "every presentation must be counted, whether or not the roster could hold it"
    );
    assert_eq!(report.last_seen_unix_ms, 40);
    assert_eq!(report.callers.len(), 2, "the roster is bounded at 2");
    assert_eq!(
        report.distinct_dropped, 1,
        "and it must SAY that it dropped one, or the short list reads as the whole truth"
    );
    assert_eq!(
        report.callers[0].count, 2,
        "the repeat landed on alice's row"
    );
    assert_eq!(report.callers[1].count, 1);
    assert!(
        !report.nothing_unbound_has_been_served(),
        "a full roster must never make the verdict say the fleet is ready"
    );
}