gunnar-ssh 3.0.0

//! The control plane this listener answers from: `gunnar-core`, directly
//! (the ACL model).
//!
//! # What this replaced
//!
//! Two local traits, `seam::AuthDirectory` and `seam::RepoResolver`, written
//! while `gunnar-core` was being built in parallel. They are gone, and this is
//! not a trait: there is one identity table and one catalog, they are
//! `gunnar-core`'s, and putting an interface in front of them would only be
//! somewhere for a second answer to live (one implementation).
//!
//! # The bug the seam had, which is why this is a rewrite and not a rename
//!
//! Both seam traits returned `Option`. `gunnar-core` returns `Result`, and
//! deliberately: the table lives in storage (the Arrow-IPC ref log), a read can fail, and
//! the ACL model spells out that `may(...)` must not be a bare `bool` because the
//! safe-looking invention for the failure case ("no access") turns a storage
//! fault into a silent permission denial. Any adapter from `Result` to `Option`
//! has to make exactly that invention.
//!
//! So the `Result` is carried all the way through
//! [`Dispatch::authorise`](crate::server::Dispatch::authorise) and lands as its
//! own [`DispatchError::Store`](crate::server::DispatchError::Store) variant:
//! separate from `Denied`, logged as a fault, and never reported to the client
//! as "you may not have this repository".

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use gunnar_core::{
    AuthTable, ControlStore, Fingerprint, HashKind, Level, NewRepo, Principal, PrincipalKind, Repo,
    RepoCatalog, RepoRef,
};

/// The identity table and the repository catalog, together.
///
/// Cheap to clone: both halves are handles onto the same stored documents.
#[derive(Clone)]
pub struct Control {
    auth: AuthTable,
    catalog: RepoCatalog,
    /// **Is this box a frozen STANDBY?** `true` refuses a client `git push`.
    ///
    /// Shared by clone with the server's replication slot, which flips it on a
    /// role change — so this front door and the control plane read ONE value
    /// (one implementation) and never disagree about whether a write should be refused. The
    /// default is an unshared `false`: a bare `Control` (a test, a build with no
    /// replication) is a writable primary, exactly as before this existed.
    writes_frozen: Arc<AtomicBool>,
    /// **Is this box a browse replica**? Shared with
    /// the replication slot the same way `writes_frozen` is; read for the
    /// refusal WORDING (a browse box's refusal names itself) and to override
    /// [`AnonymousPolicy::Refuse`] — the browse replica exists to serve
    /// anonymous reads, so it serves them whatever the flag says.
    browse: Arc<AtomicBool>,
    /// **Is this box a TWIN right now?** `true` refuses every git service —
    /// read and write, keyed and anonymous — at
    /// [`Dispatch::authorise_service`](crate::server::Dispatch::authorise_service).
    ///
    /// Shared by clone with `ServerStores`' own flag, the way `writes_frozen`
    /// and `browse` are, so the front door and the control plane read ONE value
    /// and can never disagree. It has to be live, not a snapshot: a box is made
    /// a twin over the wire by `Bootstrap.BecomeTwin` while it is already
    /// running and serving, and under the owner's one-medium ruling that is the
    /// ONLY way a twin is ever made. The default is an unshared `false` — a
    /// bare `Control` is a primary, exactly as before this existed.
    twin: Arc<AtomicBool>,
    /// **Does this box serve anonymous reads at all?** The primary of a
    /// browse-replica deployment says `Refuse`, and names where to go.
    anonymous: AnonymousPolicy,
    /// **The gate chain**: the server's answers to "may this namespace push
    /// right now", asked IN ORDER by `Dispatch::authorise_service`; the first
    /// `Refuse` wins. The server wires the disk-headroom gate first and, on a
    /// `metered` build with a monetize key, the entitlement gate second. Empty
    /// — a bare `Control`, a test — is every namespace Free with infinite disk,
    /// which is what a test that is not about the gates wants.
    plan: Vec<Arc<dyn crate::plan::PlanGate>>,
    /// **The owner's enterprise list at the push door** (2026-10-05): the
    /// name filter a first push's repository name is put to, and the
    /// accounts' audit logs its creation is written into. `None` — a bare
    /// `Control`, a test not about them — filters nothing and logs nothing.
    enterprise: Option<(
        gunnar_core::name_filter::NameFilterStore,
        Arc<gunnar_core::account_audit::AccountAuditLog>,
    )>,
    /// People's encryption keys (CRYPT-YUBI): the `gunnar-user-key` door and the
    /// recipient answer read and write them. `None`: a door without them.
    encryption_keys: Option<gunnar_core::encryption_key::EncryptionKeys>,
}

/// **The `--anonymous` knob** (the browse replica). Anonymous reads of
/// public repositories are either served here (`Allow`, the default and every
/// pre-2026-09 server) or refused with a line that names the browse replica
/// they are served by (`Refuse`). Strangers cost CPU and bandwidth on the box
/// that also takes the paying writes; the browse replica is where that cost
/// belongs, and the primary sends them there rather than quietly 404-ing.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum AnonymousPolicy {
    /// Serve anonymous reads of public repositories on this box.
    #[default]
    Allow,
    /// Refuse every anonymous request; `browse_host`, when named, goes into
    /// the refusal so `git clone` prints where to clone from instead.
    Refuse {
        /// `host[:port]` of the browse replica, for the refusal line.
        browse_host: Option<String>,
    },
}

impl Control {
    /// Wrap an already-open table and catalog. Writable (a primary) until
    /// [`Control::with_write_gate`] shares a standby's freeze flag in.
    pub fn new(auth: AuthTable, catalog: RepoCatalog) -> Self {
        Control {
            auth,
            catalog,
            writes_frozen: Arc::new(AtomicBool::new(false)),
            browse: Arc::new(AtomicBool::new(false)),
            twin: Arc::new(AtomicBool::new(false)),
            anonymous: AnonymousPolicy::Allow,
            plan: Vec::new(),
            enterprise: None,
            encryption_keys: None,
        }
    }

    /// The store of people's encryption keys this door reads and writes.
    #[must_use]
    pub fn with_encryption_keys(
        mut self,
        keys: gunnar_core::encryption_key::EncryptionKeys,
    ) -> Self {
        self.encryption_keys = Some(keys);
        self
    }

    /// People's encryption keys, when this door holds them.
    pub fn encryption_keys(&self) -> Option<&gunnar_core::encryption_key::EncryptionKeys> {
        self.encryption_keys.as_ref()
    }

    /// **The name filter and the accounts' audit logs** (2026-10-05): a
    /// repository a first push creates is named here too, and its creation is
    /// in its account's log.
    #[must_use]
    pub fn with_enterprise(
        mut self,
        names: gunnar_core::name_filter::NameFilterStore,
        audit: Arc<gunnar_core::account_audit::AccountAuditLog>,
    ) -> Self {
        self.enterprise = Some((names, audit));
        self
    }

    /// Write one event into the account `namespace` belongs to, if a person
    /// owns it. Best effort: one WARN when the log cannot be written.
    fn account_event(
        &self,
        principal: &Principal,
        namespace: &str,
        kind: &str,
        target: &str,
        detail: &str,
        refused: bool,
    ) {
        let Some((_, log)) = &self.enterprise else {
            return;
        };
        let Ok(Some(account)) = self.auth.account_of_namespace(namespace) else {
            return;
        };
        let role = self
            .auth
            .role_in_account(&principal.id, &account)
            .unwrap_or("unknown");
        let written = log.append(
            &account,
            gunnar_core::account_audit::AccountEvent {
                kind,
                actor: &principal.id,
                actor_role: role,
                target,
                detail,
                outcome: if refused {
                    gunnar_core::account_audit::Outcome::Refused
                } else {
                    gunnar_core::account_audit::Outcome::Ok
                },
            },
            gunnar_core::now_ms(),
        );
        if let Err(err) = written {
            tracing::warn!(target: "gunnar::audit", event = "account_audit.not_written", account = %account, kind, error = %err, "an event could not be written to its account's audit log");
        }
    }

    /// **An event about a person's own encryption key** (RA4-3): written into
    /// the account the person belongs to, the same kinds the gRPC door writes
    /// (`encryption-key.created`, `.method-added`, `.method-removed`).
    pub fn encryption_key_event(&self, principal: &Principal, kind: &str, detail: &str) {
        let Some((_, log)) = &self.enterprise else {
            return;
        };
        let Ok(Some(account)) = self.auth.account_of_principal(&principal.id) else {
            return;
        };
        let role = self
            .auth
            .role_in_account(&principal.id, &account)
            .unwrap_or("unknown");
        let written = log.append(
            &account,
            gunnar_core::account_audit::AccountEvent {
                kind,
                actor: &principal.id,
                actor_role: role,
                target: &principal.id,
                detail,
                outcome: gunnar_core::account_audit::Outcome::Ok,
            },
            gunnar_core::now_ms(),
        );
        if let Err(err) = written {
            tracing::warn!(target: "gunnar::audit", event = "account_audit.not_written", account = %account, kind, error = %err, "an event could not be written to its account's audit log");
        }
    }

    /// **The name filter at the push door**: the sentence a first push to
    /// `path` is refused with, or `None`. A blocked name is noted for the
    /// account admin; a borderline one is held for them (and refused until
    /// they allow it). Reserved names are a person's and an account's, never
    /// a repository's.
    pub fn push_create_name_refusal(
        &self,
        principal: &Principal,
        path: &RepoRef,
    ) -> Option<String> {
        use gunnar_core::name_filter::{NameKind, Verdict};
        let (names, _) = self.enterprise.as_ref()?;
        let account = self
            .auth
            .account_of_namespace(path.namespace())
            .ok()
            .flatten();
        let verdict = names
            .check(
                NameKind::Repo,
                path.name(),
                account.as_deref().unwrap_or(""),
            )
            .ok()?;
        if verdict == Verdict::Clean || matches!(verdict, Verdict::Reserved { .. }) {
            return None;
        }
        let hold = matches!(verdict, Verdict::Borderline { .. }) && account.is_some();
        if let Some(account) = &account {
            let now = gunnar_core::now_ms();
            let queued = if hold {
                names.hold(
                    NameKind::Repo,
                    path.name(),
                    account,
                    &principal.id,
                    &verdict,
                    now,
                )
            } else {
                names.note(
                    NameKind::Repo,
                    path.name(),
                    account,
                    &principal.id,
                    &verdict,
                    now,
                )
            };
            if let Err(err) = queued {
                tracing::warn!(target: "gunnar::audit", event = "name_filter.not_queued", account = %account, error = %err, "a name the filter stopped could not be put in its account's queue");
            }
        }
        self.account_event(
            principal,
            path.namespace(),
            if hold { "name.held" } else { "name.refused" },
            path.name(),
            &format!(
                "a repo name, {} by the name filter, at a first push",
                verdict.as_str()
            ),
            true,
        );
        Some(verdict.sentence(NameKind::Repo, path.name()))
    }

    /// **Share the standby write-gate in.** The `Arc` the server's replication
    /// slot writes on a role change; reading it is how the front door learns,
    /// per push, that this box has become a read-only standby.
    #[must_use]
    pub fn with_write_gate(mut self, writes_frozen: Arc<AtomicBool>) -> Self {
        self.writes_frozen = writes_frozen;
        self
    }

    /// Share the browse flag in — `ServerStores::browse()`, the same `Arc`
    /// the replication slot flips.
    #[must_use]
    pub fn with_browse_flag(mut self, browse: Arc<AtomicBool>) -> Self {
        self.browse = browse;
        self
    }

    /// Set the anonymous policy. `Refuse` is overridden by the browse flag:
    /// see [`Control::refuses_anonymous`].
    #[must_use]
    pub fn with_anonymous_policy(mut self, policy: AnonymousPolicy) -> Self {
        self.anonymous = policy;
        self
    }

    /// Share the twin flag in — `ServerStores::twin_flag()`, the same `Arc`
    /// `Bootstrap.BecomeTwin` flips while the server runs.
    #[must_use]
    pub fn with_twin_flag(mut self, twin: Arc<AtomicBool>) -> Self {
        self.twin = twin;
        self
    }

    /// Is this box a browse replica right now?
    pub fn is_browse(&self) -> bool {
        self.browse.load(Ordering::SeqCst)
    }

    /// **Is this box a twin right now?** A live read: the answer changes under
    /// a running server the moment `Bootstrap.BecomeTwin` returns.
    pub fn is_twin(&self) -> bool {
        self.twin.load(Ordering::SeqCst)
    }

    /// **Is an anonymous request refused here?** `Refuse` on a box that is
    /// not a browse replica. A browse replica serves anonymous reads whatever
    /// the knob says — it is the box the knob points AT, and a deployment that
    /// flipped a primary into a browse box by `SetRole` must not carry the
    /// primary's refusal along.
    pub fn refuses_anonymous(&self) -> bool {
        matches!(self.anonymous, AnonymousPolicy::Refuse { .. }) && !self.is_browse()
    }

    /// The browse host the refusal names, when the operator gave one.
    pub fn browse_host(&self) -> Option<&str> {
        match &self.anonymous {
            AnonymousPolicy::Refuse { browse_host } => browse_host.as_deref(),
            AnonymousPolicy::Allow => None,
        }
    }

    /// **Append a gate to the chain.** Order is the call order: the server
    /// hands the disk-headroom gate in first, then (`metered`) the entitlement
    /// gate, so a full box is told it is full before it is told about its
    /// plan. From here on every push asks each link in
    /// `Dispatch::authorise_service`, and the first `Refuse` is the answer.
    #[must_use]
    pub fn with_plan_gate(mut self, gate: Arc<dyn crate::plan::PlanGate>) -> Self {
        self.plan.push(gate);
        self
    }

    /// The chain, in the order it is asked. Empty when nothing was wired.
    pub fn plan_gates(&self) -> &[Arc<dyn crate::plan::PlanGate>] {
        &self.plan
    }

    /// Open both over the server's control store (the certificate-identity plan).
    pub fn open(store: Arc<dyn ControlStore>) -> Result<Self, gunnar_core::Error> {
        Ok(Control {
            auth: AuthTable::new(Arc::clone(&store))?,
            encryption_keys: Some(
                gunnar_core::encryption_key::EncryptionKeys::new(Arc::clone(&store)).map_err(
                    |e| match e {
                        gunnar_core::encryption_key::KeyRefusal::Store(e) => e,
                        other => gunnar_core::Error::ControlStore(other.to_string()),
                    },
                )?,
            ),
            catalog: RepoCatalog::new(store)?,
            writes_frozen: Arc::new(AtomicBool::new(false)),
            browse: Arc::new(AtomicBool::new(false)),
            twin: Arc::new(AtomicBool::new(false)),
            anonymous: AnonymousPolicy::Allow,
            plan: Vec::new(),
            enterprise: None,
        })
    }

    /// **Would a client write be refused right now?** `true` on a standby.
    ///
    /// The replication push from a primary is a [`PrincipalKind::Machine`]
    /// identity and is NOT a client write — a standby exists to receive it —
    /// so the freeze is scoped to non-`Machine` principals by
    /// [`Control::write_frozen_for`], which is what the dispatch actually calls.
    pub fn writes_frozen(&self) -> bool {
        self.writes_frozen.load(Ordering::SeqCst)
    }

    /// **Must this principal's write be refused because this box is a standby?**
    ///
    /// `true` only when the box is frozen AND the writer is not the replication
    /// source. A `Machine` identity is how a primary authenticates its
    /// replication push, and a standby is a box that exists precisely to accept
    /// that — freezing it would stop the very copy the standby is for. A human
    /// (`User`) or a `Deploy` key pushing directly to a standby is the write
    /// that must be refused: it would land here and never reach the primary.
    pub fn write_frozen_for(&self, principal: &Principal) -> bool {
        self.writes_frozen() && principal.kind != PrincipalKind::Machine
    }

    /// The identity table.
    pub fn auth(&self) -> &AuthTable {
        &self.auth
    }

    /// The repository catalog.
    pub fn catalog(&self) -> &RepoCatalog {
        &self.catalog
    }

    /// **Who is this?** The principal that owns `fingerprint`, or `None` to
    /// refuse the authentication.
    ///
    /// `Err` is a storage fault and is *not* the same as `Ok(None)`. The caller
    /// must refuse the key either way, but it must say which happened, because
    /// "nobody holds this key" and "the identity table could not be read" call
    /// for entirely different operator action.
    pub fn principal_for_key(
        &self,
        fingerprint: &Fingerprint,
    ) -> Result<Option<Principal>, gunnar_core::Error> {
        self.auth.principal_by_fingerprint(fingerprint)
    }

    /// **Does it exist?** The repository at `path`, or `None`.
    ///
    /// `None` here is only ever "no such repository". It must not be conflated
    /// with "no permission" at this layer: the server merges the two into one
    /// client-visible message on purpose, so a client cannot use the difference
    /// to enumerate private repositories, but they arrive separately so the
    /// server log can tell them apart.
    pub fn resolve(&self, path: &RepoRef) -> Result<Option<Repo>, gunnar_core::Error> {
        self.catalog.get(path)
    }

    /// **May they?** The highest level `principal` holds on `repo`, or `None`
    /// for no access at all.
    ///
    /// Derived by asking [`AuthTable::may`] from the top down, so this is the
    /// ACL's answer and not a second reading of the grant table. A namespace
    /// grant matches its namespace **exactly**: `team` does not reach
    /// `team/sub`. That is what closes the escalation the permissive path
    /// sandbox would otherwise carry, and it is pinned by
    /// `gunnar-core`'s `tests/acl.rs`.
    pub fn access_for(
        &self,
        principal: &Principal,
        repo: &Repo,
    ) -> Result<Option<Level>, gunnar_core::Error> {
        for level in [Level::Admin, Level::Write, Level::Read] {
            if self.auth.may(&principal.id, repo, level)? {
                return Ok(Some(level));
            }
        }
        Ok(None)
    }

    /// **The first-repository door, the checks**: may `principal` create
    /// `path` by pushing to it? `Some(spec)` is what would be created —
    /// nothing has been written; `None` the dispatch MUST render as its
    /// ordinary not-found, so the door is invisible to everyone it does not
    /// admit.
    ///
    /// # Why this exists (the appliance brick, 2026-08-25)
    ///
    /// On an installed appliance the bootstrap admin from `--admin-key`
    /// authenticates and can then do nothing: the exec surface is three git
    /// verbs, all of which want a repository; the catalog is empty; gRPC —
    /// the only surface with `Repos.Create` — is off by design;
    /// and there is no shell by construction. Nobody could ever create the
    /// first namespace. `a8ba0665` fixed "authorizes nobody"; this fixes
    /// "can do nothing".
    ///
    /// # It asks gRPC's questions, verbatim (one implementation)
    ///
    /// This is deliberately NOT a new permission vocabulary. The two
    /// predicates are exactly the ones the gRPC control plane gates creation
    /// on, so SSH and gRPC can never disagree about who may create:
    ///
    /// * a namespace that does not exist needs what `Repos.CreateNamespace`
    ///   needs: [`AuthTable::may_server`] at [`Level::Admin`] — creating a
    ///   tenant is a server-wide act;
    /// * the repository needs what `Repos.Create` needs:
    ///   [`AuthTable::may_namespace`] at [`Level::Admin`] — namespace-scoped,
    ///   because the repository does not exist yet, so there is nothing
    ///   narrower to hold a grant on.
    ///
    /// So the door admits exactly who the RPC would admit: the gunnar-admin
    /// anywhere, an account owner inside their own (existing) account
    /// namespace, a namespace-admin inside that namespace exactly (no
    /// inheritance into `team/sub`). The anonymous principal is refused by
    /// construction inside both predicates.
    ///
    /// # Why the catalog's "never conjure a namespace" rule bends here
    ///
    /// [`RepoCatalog::create`] refuses a repository whose namespace does not
    /// exist, "because then a typo in a push URL would silently become a
    /// tenant". That rule is about *unprivileged* writers. Here the namespace
    /// is only ever conjured by a principal holding `Server`/`Admin` — whose
    /// grant already covers every namespace on the box, so no tenant boundary
    /// is crossed; the blast radius of the admin's typo is a private, empty
    /// repository on the admin's own server.
    ///
    /// # What it would create
    ///
    /// [`NewRepo::new`] and nothing else: sha256, private, default
    /// branch. The one writer of defaults — a door that chose its own
    /// defaults would be the second answer one implementation forbids.
    ///
    /// # Why the write is NOT here (the ghost repository, 2026-08-30)
    ///
    /// This used to create the catalog row at exec time, before the client
    /// had seen the advertisement. A sha1 working tree pushing to the sha256
    /// default reads `object-format=sha256` in that advertisement, prints
    /// *"the receiving end does not support this repository's hash
    /// algorithm"* and hangs up **having sent no command** — and the empty
    /// repository it never pushed to was there afterwards: `repo get`
    /// succeeded, `ls-remote` answered (the site test, two FAIL rows).
    /// The server cannot know the client's hash before advertising, so the
    /// only order that leaves nothing behind on a refusal is: advertise the
    /// *prospect*, read the commands, check the hash, **then** write. The
    /// write is [`Self::admit_push_create`], called by the transport once
    /// there is a push to admit; the catalog is untouched until then.
    pub fn push_create_spec(
        &self,
        principal: &Principal,
        path: &RepoRef,
    ) -> Result<Option<NewRepo>, gunnar_core::Error> {
        let namespace = path.namespace();
        let namespace_exists = self.catalog.namespace(namespace)?.is_some();
        if !namespace_exists && !self.auth.may_server(&principal.id, Level::Admin)? {
            return Ok(None);
        }
        // The account admin's door — and a CLAN's member's (clans 2026-10-05,
        // the rule `Repos.Create` keeps): the member creates PRIVATE
        // repositories in the clan with the Write its membership gives, and
        // the prospect below is private. An invited account that has not
        // accepted is no member.
        let admin = self
            .auth
            .may_namespace(&principal.id, namespace, Level::Admin)?;
        let member = !admin
            && namespace_exists
            && self.auth.clan_role(&principal.id, namespace)?
                == Some(gunnar_core::ClanRole::Member);
        if !(admin || member) {
            return Ok(None);
        }
        // **SHA-1, and only here.** `NewRepo::new` defaults to sha256, which is
        // the right default for a repository somebody ASKS for: `repo create`
        // takes a hash kind, and a caller that wants sha256 gets it. A
        // first-push prospect is not asked for — it is inferred, before the
        // client has said one word — and git's v1 receive-pack advertises the
        // repository's object format BEFORE the client sends anything, so there
        // is nothing to infer it from. The client then reads the capability and
        // decides on its own.
        //
        // MEASURED on gunnar.rs, 2026-09-20, seeding a freshly re-imaged forge:
        //
        //     xtask: REFUSED [seed-push] nordisk/gunnar: protocol error:
        //     local repository is Sha1 but the remote is Sha256;
        //     git cannot push between object formats
        //
        // The push never reached the hash check this door's whole ordering was
        // built around ("advertise the prospect, read the commands, check the
        // hash, THEN write") — the client refused the advertisement. So the
        // first-push door, as shipped, admitted only sha256 clients, and this
        // estate has none: all four declared repositories are sha1, as is every
        // clone of them and every repository git creates without being told
        // otherwise.
        //
        // sha1 is what an un-negotiated client assumes, so it is the only
        // prospect that can be right for a client that has not spoken yet. The
        // hash check after the commands is unchanged and still has the final
        // word; a sha256 pusher is still served by `repo create`, where the
        // hash is a decision somebody made rather than one nobody could.
        Ok(Some(NewRepo::new(path.clone()).hash_kind(HashKind::Sha1)))
    }

    /// **The first-repository door, the write**: the namespace, conjured if
    /// it does not exist, and the catalog row — for a push that has sent its
    /// commands and passed the hash check. Re-asks
    /// [`Self::push_create_spec`]'s questions, so a grant revoked between the
    /// advertisement and the pack is still a refusal; `Ok(None)` is that
    /// refusal.
    ///
    /// The on-disk stores are NOT created here (this crate has no
    /// filesystem): the transport creates them through `ServerStores`, and it
    /// is the transport that makes the pair transactional — a store that
    /// cannot be made rolls this row back, see `ServerStores::create_repo`.
    ///
    /// # Races
    ///
    /// Two admins pushing the same new name race on the catalog's
    /// compare-and-swap. The loser of either CAS finds the thing it was
    /// creating already there and proceeds — the repository exists, which is
    /// what it wanted. `Err` is only ever a storage fault, and the caller
    /// must report it as one ([`crate::server::DispatchError::Store`]), never
    /// as not-found.
    ///
    /// Returns the repository and whether **this call** conjured the
    /// namespace, so a caller that fails afterwards can take the namespace
    /// back with the row and leave no trace at all.
    pub fn admit_push_create(
        &self,
        principal: &Principal,
        path: &RepoRef,
    ) -> Result<Option<(Repo, bool)>, gunnar_core::Error> {
        let Some(spec) = self.push_create_spec(principal, path)? else {
            return Ok(None);
        };
        let namespace = path.namespace();
        let mut conjured = false;
        if self.catalog.namespace(namespace)?.is_none() {
            match self.catalog.create_namespace(namespace) {
                Ok(_) => conjured = true,
                // Lost a race; it exists now, which is all that was wanted.
                Err(gunnar_core::Error::NamespaceExists(_)) => {}
                Err(err) => return Err(err),
            }
        }
        let repo = match self.catalog.create(spec) {
            Ok(repo) => repo,
            // The other pusher won; serve the repository it created.
            Err(gunnar_core::Error::RepoExists(_)) => self.catalog.open(path)?,
            Err(err) => return Err(err),
        };
        self.account_event(
            principal,
            path.namespace(),
            "repo.created",
            &path.to_string(),
            if conjured {
                "by its first push (the namespace with it)"
            } else {
                "by its first push"
            },
            false,
        );
        tracing::warn!(
            target: "gunnar::audit",
            event = "repo.created_on_first_push",
            principal = %principal.id,
            repo = %path,
            namespace_conjured = conjured,
            "repository created by its first push; the creator's grant, not a new rule, \
             is what authorised it"
        );
        Ok(Some((repo, conjured)))
    }

    /// Undo [`Self::admit_push_create`] after the store could not be made:
    /// the row goes, and the namespace too if that call conjured it. Nothing
    /// is left for `repo get` to find.
    pub fn retract_push_create(
        &self,
        path: &RepoRef,
        namespace_conjured: bool,
    ) -> Result<(), gunnar_core::Error> {
        match self.catalog.delete(path) {
            Ok(_) | Err(gunnar_core::Error::NoSuchRepo(_)) => {}
            Err(err) => return Err(err),
        }
        if namespace_conjured {
            match self.catalog.delete_namespace(path.namespace(), false) {
                Ok(_) | Err(gunnar_core::Error::NoSuchNamespace(_)) => {}
                Err(err) => return Err(err),
            }
        }
        Ok(())
    }
}

impl std::fmt::Debug for Control {
    /// By hand rather than derived, because the gates are `dyn` trait objects
    /// with no `Debug` of their own — and what an operator wants to see of them
    /// is which are wired and in what order, not their insides.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let gates: Vec<&'static str> = self.plan.iter().map(|g| g.name()).collect();
        f.debug_struct("Control")
            .field("auth", &self.auth)
            .field("catalog", &self.catalog)
            .field("writes_frozen", &self.writes_frozen())
            .field("browse", &self.is_browse())
            .field("anonymous", &self.anonymous)
            .field("gates", &gates)
            .finish()
    }
}