Skip to main content

gunnar_ssh/
server.rs

1//! The russh server: authentication, the exec dispatch, and the channel
2//! hand-off (the ACL model, the two-listener split).
3//!
4//! # The shape of a connection
5//!
6//! 0. **`auth_none`**, the anonymous read path, and the only place
7//!    gunnar reads the SSH username. See below.
8//! 1. **`auth_publickey`**: russh has already verified that the client owns the
9//!    offered key. gunnar fingerprints it (SHA-256 over the wire blob, OpenSSH's
10//!    printed form, [`crate::access::fingerprint_of_key`]) and asks
11//!    `gunnar-core`'s identity table through [`Control`]. The SSH *username* is
12//!    logged and otherwise ignored: identity comes from the key, so
13//!    `git@host:ns/repo.git` works and `git` means nothing (the ACL model).
14//! 2. **`channel_open_session`**. The channel is kept, because the exec request
15//!    arrives as a separate message that only carries a [`ChannelId`], and the
16//!    stream can only be taken from the `Channel` itself. A watchdog is armed
17//!    here so a session that authenticates and then never execs cannot sit on
18//!    the connection.
19//! 3. **`env_request`**, answered against [`ENV_ALLOWLIST`], which holds
20//!    exactly one name: `GIT_PROTOCOL`. Everything else is refused. See below.
21//! 4. **`exec_request`**: [`Dispatch::authorise`] parses, resolves and checks
22//!    the grant. Only then is the channel's stream handed to [`GitTransport`].
23//! 5. **everything else**. `shell_request`, `subsystem_request`,
24//!    `pty_request` and `x11_request` are answered with a channel failure.
25//!    **No shell, ever.**
26//!
27//! # Why the transport is spawned and not awaited
28//!
29//! russh calls handler methods from the session's own read loop (see
30//! `russh::server::Session::run`). Awaiting a whole clone inside `exec_request`
31//! would stop the loop that has to deliver the client's bytes to the channel;
32//! the transfer would deadlock on its first read. russh's own `sftp_server`
33//! example awaits inline; gunnar does not.
34//!
35//! # The exit status, and why the channel is split rather than consumed
36//!
37//! `Channel::into_stream()` wraps the channel in russh's `ChannelCloseOnDrop`,
38//! which spawns a task sending `ChannelMsg::Close` when the transport lets the
39//! stream go. That `Close` raced the `exit-status` the server sent next, and
40//! won often enough that a *successful* push reported failure. Measured
41//! 2026-08-03: a complete 244-byte advertisement delivered with `ssh` exiting
42//! **255**. See [`crate::channel`].
43//!
44//! The channel is therefore `split()` instead. The read half becomes an owned
45//! [`ChannelReader`], the write half stays with the server, and `exit-status`
46//! and `close` go out through **the same sender the data went through**, in
47//! order, after the transport has finished with it. Nothing sleeps and nothing
48//! races.
49//!
50//! # One span per connection, and what hangs off it
51//!
52//! ```text
53//!   ssh.connection { conn, peer, principal, repo }     gunnar::ssh
54//!     └── git.transfer { service, access }             gunnar::git
55//!           ├── git.upload_pack | git.receive_pack     gunnar::git
56//!           └── policy.gate { ref_name }               gunnar::policy
57//! ```
58//!
59//! `principal` and `repo` are declared empty when the connection opens and
60//! **recorded** when `auth_publickey` and `exec_request` learn them, so a line
61//! emitted at any depth carries all four fields without any call site restating
62//! them. That is the difference from what was here before: connection identity
63//! was an ad-hoc `?self.peer` copy-pasted to fourteen call sites, there was no
64//! id at all, and two lines from the same connection could not be tied
65//! together.
66//!
67//! Two hand-offs would drop the span if they were not made explicit, and each
68//! is a place where an operator would silently lose half a connection:
69//!
70//! * `tokio::spawn`: the exec watchdog and the transfer task both take
71//!   `.instrument(span)`, because a spawned task does not inherit a
72//!   thread-local span;
73//! * `spawn_blocking`: `gunnar-cli`'s `WireTransport::serve` captures
74//!   `Span::current()` and enters it on the blocking thread, which is where the
75//!   whole git conversation actually happens.
76//!
77//! # How an anonymous session arrives, and why it is `auth_none` on one
78//! # username
79//!
80//! SSH has no unauthenticated mode: a client must complete *some* userauth
81//! method before it may open a channel. Three shapes were available.
82//!
83//! | | why not |
84//! |---|---|
85//! | accept `none` for **every** username | **This is the one that would break the server.** OpenSSH's client always sends a `none` request first, to learn the method list. A server that answers it with SUCCESS authenticates *every* stock client as anonymous before it ever offers its key, so every existing push would start failing, and every existing user would silently lose their identity. Measured against the russh dispatch: `server/encrypted.rs` calls `auth_none` on that first probe regardless of the advertised `methods`. |
86//! | a **well-known published key** | The anonymous-FTP shape. It works, but a stranger has to obtain a key file before they can clone, which is worse than the problem it solves, and gunnar would then hold a key whose private half is public in the same table that authenticates real users. |
87//! | **`none`, on one reserved username** | Taken. |
88//!
89//! `ssh://anonymous@host/ns/repo.git` gets the `none` method accepted and a
90//! session whose principal is
91//! [`Principal::anonymous`](gunnar_core::Principal::anonymous). Every other
92//! username gets `none` refused, exactly as russh's default handler refused it
93//! before this existed, and the client falls through to publickey unchanged.
94//!
95//! **Why it cannot become a bypass**, in the order the objections come:
96//!
97//! * *Does it weaken private repositories?* No. The username selects the
98//!   authentication method; the **authorisation** answer still comes from
99//!   `AuthTable::may` (the ACL model), where the anonymous principal is one rule, `Read` on
100//!   `Public`, and is refused everything else including `may_namespace` and
101//!   `may_server`. There is no second ACL (one implementation).
102//! * *Can a caller claim to be someone by picking a username?* No. The
103//!   anonymous principal is the **only** identity a username can select, it is
104//!   never in the identity table, [`Principal::new`] and
105//!   `AuthTable::add_principal` refuse its id, so no key and no grant can ever
106//!   attach to it.
107//! * *Can an authenticated user be silently downgraded?* No. `none` is refused
108//!   for their username, so nothing changes for them; and a user who *does*
109//!   type `anonymous@` asked for strictly less access, which is not an
110//!   escalation.
111//! * *Does the username now reach an authorisation decision, against the ACL model?* It
112//!   reaches exactly one, in the direction of *removing* everything. The ACL model's
113//!   rule exists so that a username cannot be used to become somebody; this
114//!   uses it to become nobody.
115//!
116//! **`none` is not advertised** in [`RusshConfig::methods`], which stays
117//! publickey-only. Advertising it would tell every client that `none` is
118//! available when it is available only for one name (the failure list is
119//! supposed to be true), and russh calls `auth_none` on the client's opening
120//! probe either way, which is the only place it is ever needed.
121//!
122//! [`SshConfig::allow_anonymous`](crate::SshConfig::allow_anonymous) closes the
123//! whole path at the listener for an appliance that wants it shut.
124//!
125//! # `GIT_PROTOCOL`, and why it is an allowlist of one
126//!
127//! A git client announces protocol v2 over SSH by sending
128//! `GIT_PROTOCOL=version=2` as a channel environment request. Refusing every
129//! `env_request`, which this server used to do, meant every session fell back
130//! to v0. Stock `git` copes; `gix`'s blocking v0 client does not, because it
131//! requires `multi_ack_detailed` and `gunnar_wire::caps` correctly declines to
132//! advertise a capability it does not implement. The measured consequence was
133//! that `gunnar clone` could not clone from `gunnar serve` at all.
134//!
135//! Accepting the variable is **not** the same as accepting an environment. An
136//! SSH server that passes arbitrary client-supplied names through has handed
137//! the client `LD_PRELOAD`, `PATH` and every other lever there is. [`ENV_ALLOWLIST`]
138//! names the one variable gunnar reads; every other name is refused with a
139//! channel failure, exactly as before. The accepted value is not an input to
140//! any authorisation decision: it reaches [`ServiceRequest::git_protocol`] and
141//! nothing else, and it only ever selects the *framing* the wire answers in.
142
143use std::collections::HashMap;
144use std::net::SocketAddr;
145use std::sync::atomic::{AtomicBool, Ordering};
146use std::sync::Arc;
147use std::time::Duration;
148
149/// How long [`BoundServer::shutdown`] waits for sessions to answer the
150/// disconnect before it aborts the listener. Longer than the gRPC grace (2 s):
151/// a git transfer mid-pack deserves a moment to finish its write.
152pub const SSH_SHUTDOWN_GRACE: Duration = Duration::from_secs(10);
153
154use russh::keys::ssh_key::PublicKey;
155use russh::server::{Auth, Config as RusshConfig, Handler, Msg, Server, Session};
156use russh::{Channel, ChannelId, Disconnect, MethodKind, MethodSet};
157use tokio::net::TcpListener;
158use tokio::task::JoinHandle;
159use tracing::Instrument;
160
161use gunnar_core::observe::{RequestId, ServerMetrics, TARGET_GIT, TARGET_SSH};
162use gunnar_core::RepoRef;
163
164use crate::access::{fingerprint_of_key, Access, Fingerprint, Principal};
165use crate::admission::{Admission, AdmissionStats};
166use crate::channel::{serve_until_session_ends, session_ended_error, session_over, ChannelReader, SessionBound};
167use crate::command::GitService;
168use crate::command::{self, CommandError};
169use crate::config::{SshConfig, ANONYMOUS_SSH_USER};
170use crate::control::Control;
171use crate::governor::OpClass;
172use crate::seam::{Delivery, GitTransport, ServiceRequest};
173
174// ───────────────────────────────────────────────────────────────────────────
175// The authorisation decision, without any SSH in it
176// ───────────────────────────────────────────────────────────────────────────
177
178/// Why an exec request was not served.
179#[derive(Debug, thiserror::Error)]
180pub enum DispatchError {
181    /// The command did not parse. **The `bash` case, and every other one.**
182    #[error("{0}")]
183    Command(#[from] CommandError),
184    /// The repository does not exist.
185    #[error("no such repository: {0}")]
186    NoSuchRepo(RepoRef),
187    /// The principal does not hold enough access.
188    #[error("{principal} holds {held} on {repo}, needs {needed}")]
189    Denied {
190        /// Who asked.
191        principal: String,
192        /// What they asked for.
193        repo: RepoRef,
194        /// What the command requires.
195        needed: Access,
196        /// What they actually hold, rendered as `none` when they hold nothing.
197        held: String,
198    },
199    /// **This box is a read-only STANDBY** and the write is refused.
200    ///
201    /// A distinct arm, not a [`DispatchError::Denied`]: the writer holds the
202    /// grant — on a primary this push would land — so "you do not have access"
203    /// would be a lie that sends them to an administrator about permissions.
204    /// The honest sentence is that this box does not take client writes; push
205    /// to the primary. It is only ever reached by a principal who already
206    /// passed the grant check, so it leaks no repository the writer could not
207    /// already see.
208    #[error("{principal}: this box is a standby; writes go to the primary ({repo})")]
209    Standby {
210        /// Who asked.
211        principal: String,
212        /// The repository they tried to write.
213        repo: RepoRef,
214        /// The box is a browse replica, so the
215        /// refusal says so: a stranger who found a public repository here and
216        /// tried to push to it should learn what this box is, not merely that
217        /// it is frozen.
218        browse: bool,
219    },
220    /// **This box is a TWIN and serves no git at all** — not a read, not a
221    /// write, not to anybody (2026-09-21, lane T4).
222    ///
223    /// Its own arm and not a [`DispatchError::Standby`], because a standby is a
224    /// box that serves reads and refuses writes, and a twin is a box that
225    /// serves NEITHER: `serve.rs` says why in one line — *a read would grow the
226    /// caches the twin must not hold* — and the twin's whole value is that its
227    /// disk holds the primary's log and nothing derived.
228    ///
229    /// **Why this exists at the door rather than only at the bind.** The bind
230    /// is chosen ONCE, in `serve::start`, from `stores.is_twin()` at that
231    /// instant. A box becomes a twin over the wire, through
232    /// `Bootstrap.BecomeTwin`, long after that — and the owner's one-medium
233    /// ruling makes that the ONLY way it happens, because a twin is now
234    /// installed from the same image as a main with no per-box parameter to
235    /// say which it is. MEASURED under KVM by lane T3, 2026-09-21: a box
236    /// conferred at runtime kept `0.0.0.0:2222` bound for the rest of its life,
237    /// the `this box is a TWIN` warning never printed, and its volume ended up
238    /// holding five `objects.exploded` cache files. A socket that cannot be
239    /// moved without a restart is not a reason to serve git from it in the
240    /// meantime.
241    #[error("{principal}: this box is a TWIN; it serves no git to anyone — clone and push to the primary it follows ({repo})")]
242    Twin {
243        /// Who asked.
244        principal: String,
245        /// What they asked for.
246        repo: RepoRef,
247    },
248    /// **Anonymous reads are not served here** (`--anonymous refuse`); the
249    /// browse replica serves them, and `browse_host` names it when the
250    /// operator did. Raised before the repository is even looked up, so a
251    /// stranger learns nothing about what exists on the primary.
252    #[error("anonymous reads are refused on this box ({repo}); browse host: {browse_host:?}")]
253    AnonymousElsewhere {
254        /// The repository the stranger asked for.
255        repo: RepoRef,
256        /// Where anonymous reads are served, when the operator named it.
257        browse_host: Option<String>,
258    },
259    /// **The gate chain refuses this**: the disk is under its floor (every
260    /// build), or (`metered`) a Suspended or Retention tenant's push or LFS
261    /// upload, an anonymous read of a lapsed tenant's public repository, or a
262    /// push at the pack quota.
263    ///
264    /// Its own arm for the same reason [`DispatchError::Standby`] is one: the
265    /// caller holds the grant, so "you do not have access" would be a lie
266    /// that sends a paying customer to an administrator about permissions.
267    /// The honest sentence names the fact — free bytes and the floor, or the
268    /// state, the date and where to renew — and it is only ever reached AFTER
269    /// the grant check admitted the request, so it leaks nothing about a
270    /// repository the caller could not already see.
271    #[error("{namespace}: {line}")]
272    Plan {
273        /// The tenant.
274        namespace: String,
275        /// What the client is told, verbatim.
276        line: String,
277    },
278    /// The identity table or the catalog could not be read.
279    ///
280    /// **Deliberately not folded into [`DispatchError::Denied`].** The tables
281    /// live in storage (the Arrow-IPC ref log) and a read can fail; answering a storage
282    /// fault with "you do not have access" would turn an outage into a silent,
283    /// convincing permission denial that nobody would think to investigate.
284    /// The ACL model is explicit that `may(...)` returns `Result<bool>` for this
285    /// reason, and this variant is where that `Result` lands.
286    #[error("the control plane could not answer: {0}")]
287    Store(#[from] gunnar_core::Error),
288}
289
290impl DispatchError {
291    /// What the client is told.
292    ///
293    /// `None` means "answer with a channel failure and say nothing": the
294    /// request was not even a git command, so there is no git client on the
295    /// other end that would understand a pkt-line.
296    ///
297    /// **"No such repository" and "no permission" produce the same string on
298    /// purpose.** A client that could tell them apart could enumerate private
299    /// repositories one guess at a time. The server log keeps the distinction;
300    /// the wire does not.
301    ///
302    /// Those two arms are **one arm**, and were two identical `format!`s until
303    /// 2026-08-05. That is the twinning one implementation forbids, and this is the case it
304    /// forbids it for: the property is that the two strings are equal, so two
305    /// copies of the string put the property one careless edit away from being
306    /// false, in the direction where the wire starts leaking which private
307    /// repositories exist.
308    /// `a_missing_repository_and_a_forbidden_one_are_one_string_to_the_client`
309    /// guards it from the far side of a real session; this makes the guard hard
310    /// to need.
311    pub fn client_message(&self) -> Option<String> {
312        match self {
313            // A verb gunnar knows and refuses BY NAME gets its sentence; the
314            // rest of the parse failures are not git clients at all.
315            DispatchError::Command(CommandError::NotServed { verb, why }) => {
316                Some(format!("{verb}: {why}"))
317            }
318            DispatchError::Command(_) => None,
319            DispatchError::NoSuchRepo(repo) | DispatchError::Denied { repo, .. } => Some(format!(
320                "repository not found, or you do not have access: {repo}"
321            )),
322            // A standby refusal is told plainly — the writer is authorised, so
323            // there is no repository to hide from them, and "push to the
324            // primary" is the one thing they need to hear to succeed.
325            // A twin refusal is told plainly for the same reason, and it is told
326            // to EVERYONE including a stranger: it names no repository and
327            // reveals nothing this box holds — "this is a twin" is a fact about
328            // the box, not about its contents.
329            DispatchError::Twin { .. } => Some(
330                "this box is a TWIN: it holds a copy of the primary's transaction log and serves \
331                 no git to anyone, read or write. Clone and push to the primary it follows"
332                    .to_string(),
333            ),
334            DispatchError::Standby { browse: false, .. } => {
335                Some("this box is a standby; push to the primary".to_string())
336            }
337            DispatchError::Standby { browse: true, .. } => Some(
338                "this box is a browse replica: it serves public repositories read-only and \
339                 takes no writes; push to the primary"
340                    .to_string(),
341            ),
342            DispatchError::AnonymousElsewhere {
343                browse_host: Some(host),
344                repo,
345            } => Some(format!(
346                "anonymous reads are served by the browse replica, not here: clone \
347                 {repo} from {host} (or authenticate with a registered key)"
348            )),
349            DispatchError::AnonymousElsewhere {
350                browse_host: None, ..
351            } => Some(
352                "this server does not serve anonymous reads; authenticate with a registered \
353                 key, or use the browse replica"
354                    .to_string(),
355            ),
356            // The gate's refusal is told in full: the state, the date and the
357            // URL are the three things a lapsed tenant needs to get going again,
358            // and the free bytes and the floor are what a full box's operator
359            // needs; hiding any of them would turn a fact into a support ticket.
360            DispatchError::Plan { line, .. } => Some(line.clone()),
361            // A fault, not a decision. The client is told the server broke,
362            // because telling it "you do not have access" would be a lie that
363            // sends the user to their administrator about permissions.
364            DispatchError::Store(_) => {
365                Some("the gunnar control plane is unavailable; try again".to_string())
366            }
367        }
368    }
369}
370
371/// Parse → resolve → check, with no SSH anywhere in it.
372///
373/// Kept separate from the [`Handler`] so the security decision can be tested
374/// directly, exhaustively and without a handshake, and so there is exactly one
375/// place where it is made.
376#[derive(Clone, Debug)]
377pub struct Dispatch {
378    control: Control,
379}
380
381impl Dispatch {
382    /// Build a dispatch over the control plane.
383    pub fn new(control: Control) -> Self {
384        Dispatch { control }
385    }
386
387    /// The control plane this dispatch authenticates and authorises against.
388    pub fn control(&self) -> &Control {
389        &self.control
390    }
391
392    /// Decide whether `principal` may run `raw` on this server.
393    ///
394    /// Parse (which includes the sandbox), resolve, then check the grant. A
395    /// storage failure at either lookup comes back as
396    /// [`DispatchError::Store`] rather than as a denial. See
397    /// [`crate::control`].
398    ///
399    /// **The client's environment is not a parameter here, and must not
400    /// become one.** The returned [`ServiceRequest`] carries
401    /// `git_protocol: None`; the handler attaches the allowlisted value with
402    /// [`ServiceRequest::with_git_protocol`] *after* this decision has already
403    /// been made, so there is no path by which anything the client typed into
404    /// its environment can reach the grant check.
405    pub fn authorise(
406        &self,
407        principal: &Principal,
408        key_fingerprint: Option<&Fingerprint>,
409        raw: &[u8],
410    ) -> Result<ServiceRequest, DispatchError> {
411        let parsed = command::parse(raw)?;
412        self.authorise_service(principal, key_fingerprint, parsed.service, parsed.repo)
413    }
414
415    /// The same decision, entered from a **URL path** rather than an exec line.
416    ///
417    /// Smart HTTP does not send `git-upload-pack '<path>'`; it sends
418    /// `POST /<path>/git-upload-pack`. The verb and the path arrive already
419    /// separated, and the path has no shell quoting on it, so there is nothing
420    /// for [`command::parse`] to do — but everything **after** the parse is
421    /// identical and must stay identical: resolve, ask the ACL, and refuse in
422    /// the one way that does not tell a caller whether a repository exists.
423    ///
424    /// So this is where that decision lives and [`Dispatch::authorise`] is now
425    /// a parser in front of it. Two entry points, one answer (one implementation). A second
426    /// copy of this in `gunnar-http` is precisely the thing that ends with SSH
427    /// and HTTP disagreeing about who may push, and the disagreement being
428    /// found by whoever pushes.
429    ///
430    /// `repo` is a [`RepoRef`], which is to say it has already been through
431    /// [`gunnar_core::resolve_repo_path`]. The sandbox is not optional on this
432    /// path either: the caller cannot construct a `RepoRef` without it.
433    pub fn authorise_service(
434        &self,
435        principal: &Principal,
436        key_fingerprint: Option<&Fingerprint>,
437        service: GitService,
438        path: RepoRef,
439    ) -> Result<ServiceRequest, DispatchError> {
440        let required = service.required_access();
441        // **A TWIN serves no git**, before anything else is asked — before the
442        // anonymous policy, before the catalog, before the grant. A read on a
443        // twin grows exactly the caches the twin exists not to hold, and the
444        // box's own `browse` flag (which `become_twin` sets) would otherwise
445        // OVERRIDE `--anonymous refuse` and serve a stranger. This is a live
446        // read of the same flag `ServerStores` flips, so the refusal begins the
447        // instant `Bootstrap.BecomeTwin` returns and does not wait for a
448        // restart to move the listener.
449        if self.control.is_twin() {
450            return Err(DispatchError::Twin { principal: principal.id.clone(), repo: path });
451        }
452        // **`--anonymous refuse`** (the browse replica), before the catalog is consulted:
453        // a stranger on a primary that has a browse replica is sent there, and
454        // learns nothing about what this box holds on the way.
455        if principal.id == gunnar_core::ANONYMOUS_PRINCIPAL_ID && self.control.refuses_anonymous() {
456            return Err(DispatchError::AnonymousElsewhere {
457                repo: path,
458                browse_host: self.control.browse_host().map(str::to_owned),
459            });
460        }
461        let (repo, prospective) = match self.control.resolve(&path)? {
462            Some(repo) => (repo, false),
463            // The repository does not exist. For exactly one shape of request
464            // — a PUSH, by a principal the control plane already permits to
465            // create repositories here — absence is the first-repository door
466            // rather than a refusal: the repository is created and the push
467            // proceeds through the unchanged grant check below.
468            //
469            // # Why a door here, and not a fourth verb or the control plane
470            //
471            // The shipped appliance runs `--no-grpc` (gRPC v1 is
472            // unauthenticated, loopback-only), has no shell by construction,
473            // and its exec surface is three git verbs that all want a
474            // repository — so after `--admin-key` enrolled the bootstrap
475            // admin (a8ba0665), the admin could authenticate and then do
476            // NOTHING, forever (measured under KVM, 2026-08-25). The
477            // alternatives lost on the code, not on taste:
478            //
479            // * **a create verb** would be a wire surface no git client
480            //   speaks, would need its own client incantation, and does not
481            //   fit the dispatch's own shape: `SERVICES` carries per-service
482            //   *repo grant levels* and this decision compares levels — a
483            //   verb whose target does not exist has no repo to hold a level
484            //   on, so it would be a special case beside the table, and a
485            //   probe-discoverable one for every stranger;
486            // * **turning gRPC on** is the door the unauthenticated v1 and
487            //   `gunnar-grpc/tests/loopback_default.rs` exist to keep shut on
488            //   a mailed medium: `Bootstrap.Enroll` with no credential is a
489            //   race between the customer and whoever else finds the box.
490            //
491            // Push-to-create adds no verb, no parse production and no
492            // vocabulary: `Control::create_for_push` asks the exact two
493            // questions `Repos.CreateNamespace`/`Repos.Create` gate on, so
494            // this door and the RPC door can never disagree (one implementation) — and it
495            // sits HERE, where the two entry points (exec line, smart-HTTP
496            // URL) have already fanned in, so SSH and HTTP answer a push to a
497            // missing repository identically.
498            //
499            // Only a WRITE: ReceivePack, and (2026-09-28) an LFS upload.
500            // Reading never creates: a `git clone` of a typo, or an LFS
501            // download, stays an error.
502            //
503            // **Why the LFS upload too.** git-lfs's pre-push hook runs AFTER
504            // git has taken receive-pack's advertisement and BEFORE it sends
505            // a command, and it asks `git-lfs-transfer <repo> upload` first.
506            // MEASURED 2026-09-28 (git-lfs 3.8.0, and on a real box): into a
507            // repository that did not exist yet that exec was the fused
508            // not-found, the hook died — `batch request: exec request failed
509            // on channel 0: exit status 255` — and the push that would have
510            // created the repository never sent its commands. The upload is
511            // the same act as the push (it puts content in the repository),
512            // by the same principal, and it is asked the same two questions
513            // (`push_create_spec`), so it opens the same door and nobody else
514            // learns anything new. Nothing is written here either: the
515            // transport creates the repository when the first OBJECT arrives
516            // (`gunnar-server` `ProspectiveLfs`), so a client that only
517            // batches, or hangs up, leaves nothing.
518            //
519            // For everyone `create_for_push` answers `None` for, the refusal
520            // below is bit-identical to what it always was — the one fused
521            // "repository not found, or you do not have access" — so the
522            // door's existence is not observable to a key it does not admit.
523            //
524            // **Nothing is written here.** The door answers with the
525            // repository it WOULD create (`prospective`), the transport
526            // advertises that, and the catalog row and the store are made
527            // only once the client has sent commands that pass the hash
528            // check — see `Control::push_create_spec` for the ghost
529            // repository this order closes.
530            None if service == GitService::ReceivePack
531                || service == GitService::LfsTransfer(crate::command::Operation::Upload) =>
532            {
533                match self.control.push_create_spec(principal, &path)? {
534                    Some(spec) => {
535                        // The name filter (2026-10-05) — asked only of a
536                        // pusher the door admits, so its sentence tells
537                        // nobody else anything.
538                        if let Some(line) = self.control.push_create_name_refusal(principal, &path) {
539                            return Err(DispatchError::Plan {
540                                namespace: path.namespace().to_owned(),
541                                line,
542                            });
543                        }
544                        (spec.prospect()?, true)
545                    }
546                    None => return Err(DispatchError::NoSuchRepo(path)),
547                }
548            }
549            None => return Err(DispatchError::NoSuchRepo(path)),
550        };
551        let held = self.control.access_for(principal, &repo)?;
552        match held {
553            // **The standby freeze.** Checked here, INSIDE the authorised arm,
554            // so it is only ever reached by a writer the grant check already
555            // admitted — an unauthorised push still gets the fused not-found and
556            // learns nothing new. A write-requiring service by a non-`Machine`
557            // principal on a frozen box is refused: this box has abdicated or is
558            // a configured standby, and a client write here would be a split
559            // the primary never sees. The primary's own replication push is a
560            // `Machine` identity and sails through — that is what a standby is
561            // for.
562            Some(level)
563                if level.permits(required)
564                    && required.permits(Access::Write)
565                    && self.control.write_frozen_for(principal) =>
566            {
567                Err(DispatchError::Standby {
568                    principal: principal.id.clone(),
569                    repo: path,
570                    browse: self.control.is_browse(),
571                })
572            }
573            // **An encrypted repository takes no LFS upload** (wiring tracer B
574            // 2026-10-08). git-lfs uploads in the pre-push hook, BEFORE the
575            // pack the encrypted gate reads, and an LFS object is the
576            // plaintext of a `filter=lfs` file that never went through
577            // `filter=gunnar-crypt`: refused here, for SSH and smart HTTP
578            // alike, before a byte is stored. Asked inside the authorised arm,
579            // so a stranger still gets the fused not-found. A download is never
580            // refused.
581            Some(level)
582                if level.permits(required)
583                    && repo.encrypted
584                    && service == GitService::LfsTransfer(crate::command::Operation::Upload) =>
585            {
586                Err(DispatchError::Plan {
587                    namespace: path.namespace().to_owned(),
588                    line: format!(
589                        "{path} is encrypted, so it takes no LFS objects: LFS stores files as \
590                         they are, outside your members' encryption. Track the file with \
591                         `filter=gunnar-crypt` instead of `filter=lfs`"
592                    ),
593                })
594            }
595            Some(level) if level.permits(required) => {
596                // **The gate chain**, asked LAST: after the grant (so a
597                // refusal leaks nothing to a stranger) and after the standby
598                // freeze (so a standby stays one whatever the gates say). A
599                // write asks `Write`; an anonymous read asks `AnonymousRead`;
600                // an authenticated read asks nothing at all, because the
601                // owner's clone is allowed in every state — data is never held
602                // hostage. See `crate::plan`.
603                let PlanOutcome {
604                    notice: plan_notice,
605                    ceiling: pack_ceiling,
606                    pack_mark,
607                    allowance: lfs_allowance,
608                } = self.ask_plan(principal, service, required, &repo)?;
609                Ok(ServiceRequest {
610                    service,
611                    repo,
612                    prospective,
613                    principal: principal.clone(),
614                    access: level,
615                    // The key that authenticated this session, carried
616                    // rather than guessed back out of the identity table. A
617                    // parameter and not a builder, so it cannot be forgotten into
618                    // `None`.
619                    key_fingerprint: key_fingerprint.cloned(),
620                    git_protocol: None,
621                    delivery: Delivery::Session,
622                    // Derived from `key_fingerprint` by the transport rather
623                    // than spelled here: see `ServiceRequest::credential`.
624                    credential: None,
625                    plan_notice,
626                    pack_ceiling,
627                    pack_mark,
628                    lfs_allowance,
629                    cancel: crate::channel::Cancel::new(),
630                })
631            }
632            other => Err(DispatchError::Denied {
633                principal: principal.id.clone(),
634                repo: path,
635                needed: required,
636                held: match other {
637                    Some(level) => level.to_string(),
638                    None => "none".to_string(),
639                },
640            }),
641        }
642    }
643
644    /// **Ask the gate chain**, for a request the grant check already admitted.
645    /// `Ok(Some(line))` is an allowed request that carries a warning (Grace);
646    /// `Err(Plan)` is the refusal, with the line the client is told.
647    ///
648    /// The three shapes, and why each asks what it asks:
649    ///
650    /// * a **write** (`git-receive-pack`, LFS upload) asks
651    ///   [`PlanAction::Write`](crate::plan::PlanAction::Write), and if that is
652    ///   allowed asks [`PlanGate::pack_quota`](crate::plan::PlanGate::pack_quota)
653    ///   — both before a single pack byte is read;
654    /// * an **anonymous read** asks
655    ///   [`PlanAction::AnonymousRead`](crate::plan::PlanAction::AnonymousRead):
656    ///   a lapsed tenant's public surface is what costs the provider and earns
657    ///   nothing;
658    /// * an **authenticated read** asks nothing. The owner clones their own
659    ///   repositories in every state, and so does anyone they granted. That is
660    ///   rule 1 of `gunnar_core::entitlement`, and it is enforced here by the
661    ///   absence of a question rather than by an answer.
662    ///
663    /// **The chain.** Gates are asked in the order `Control::with_plan_gate`
664    /// added them — disk first — and the FIRST `Refuse` is the answer; a later
665    /// gate is not consulted, so a full box is told it is full and not also
666    /// that its plan lapsed. Each gate is asked BOTH its questions before the
667    /// next gate is asked anything: its ladder (`verdict`), then, for a write
668    /// that grows data, its quota (`pack_quota`). Within one gate that order is
669    /// what tells a Suspended tenant they are suspended rather than also full;
670    /// across gates it is what makes "disk first" mean disk first — the disk
671    /// gate only ever refuses from `pack_quota`, and a chain that asked every
672    /// ladder before any quota would let a plan's ladder outrank it whatever
673    /// the order (`tests/disk_headroom.rs` caught exactly that, red first).
674    /// Warnings from every allowing gate are joined.
675    ///
676    /// With no gate wired (`Control::with_plan_gate` never called) every
677    /// namespace is Free and this is a no-op. A `Machine` principal —
678    /// replication, the primary's copy arriving — is asked only of the gates
679    /// that say they apply to it ([`PlanGate::applies_to_machine`]): the disk
680    /// does, a plan does not, because refusing a lapsed tenant's replication
681    /// would make their standby diverge from the primary.
682    fn ask_plan(
683        &self,
684        principal: &Principal,
685        service: GitService,
686        required: Access,
687        repo: &gunnar_core::Repo,
688    ) -> Result<PlanOutcome, DispatchError> {
689        use crate::plan::{PlanAction, PlanVerdict};
690        let machine = principal.kind == gunnar_core::PrincipalKind::Machine;
691        let gates: Vec<_> = self
692            .control
693            .plan_gates()
694            .iter()
695            .filter(|gate| !machine || gate.applies_to_machine())
696            .collect();
697        if gates.is_empty() {
698            return Ok(PlanOutcome::default());
699        }
700        let anonymous = principal.id == gunnar_core::ANONYMOUS_PRINCIPAL_ID;
701        let action = if required.permits(Access::Write) {
702            PlanAction::Write
703        } else if anonymous {
704            PlanAction::AnonymousRead
705        } else {
706            return Ok(PlanOutcome::default());
707        };
708        let refused = |line: String| DispatchError::Plan {
709            namespace: repo.namespace.clone(),
710            line,
711        };
712        // `service` is named here so a future fourth write-shaped verb has to
713        // decide whether it grows the tenant's data.
714        let grows_data = matches!(
715            service,
716            GitService::ReceivePack | GitService::LfsTransfer(crate::command::Operation::Upload)
717        );
718        let mut notices: Vec<String> = Vec::new();
719        let mut ceiling: Option<crate::plan::PackCeiling> = None;
720        let mut allowance: Option<gunnar_lfs::Allowance> = None;
721        // **Marked BEFORE any gate walks the disk, and that ordering is the
722        // contract**, not an accident of where the line sits. A gate's
723        // `lfs_allowance` measures the namespace's bytes off the disk; an upload
724        // admitted DURING that walk may or may not have landed bytes for the
725        // walk to see. Taking the mark first puts such an admission after the
726        // mark, so it is charged — and charged twice if the walk saw it too,
727        // which refuses a shade early. Taking the mark after the walk would put
728        // it before the mark and possibly outside the walk, which admits too
729        // much. Only one of those two roundings is a quota that leaks. See
730        // `gunnar_lfs::ledger`.
731        let mark = gunnar_lfs::AllowanceLedger::shared().mark(&repo.namespace);
732        // …and the same standing in the PACK meter, taken in the same breath and
733        // for the same reason: `pack_ceiling` below measures the namespace's
734        // packfile bytes off the disk, and a push admitted during that walk has
735        // to land after this mark to be counted.
736        //
737        // **Two marks and not one, because they are two meters.**
738        // `caps["pack_bytes"]` and `caps["lfs_bytes"]` are capped separately on
739        // the signed fact and reported separately by `Usage.Read`, and the two
740        // numbers these marks are judged against are computed from different
741        // caps. One shared counter would charge an LFS upload against a push's
742        // ceiling — cutting a push mid-transfer over room its own cap never
743        // lost — which is a quota that refuses wrongly, the same defect as one
744        // that admits wrongly. They share a disk; they do not share a cap. See
745        // `gunnar_lfs::AllowanceLedger::shared_pack`.
746        let pack_mark = gunnar_lfs::AllowanceLedger::shared_pack().mark(&repo.namespace);
747        for gate in &gates {
748            match gate.verdict(&repo.namespace, action) {
749                PlanVerdict::Allow => {}
750                PlanVerdict::Warn(line) => notices.push(line),
751                PlanVerdict::Refuse(line) => return Err(refused(line)),
752            }
753            // The quota, only for a write this gate's ladder allowed.
754            if action == PlanAction::Write && grows_data {
755                if let PlanVerdict::Refuse(line) = gate.pack_quota(&repo.namespace) {
756                    return Err(refused(line));
757                }
758                // **The smallest ceiling in the chain wins**, which is the
759                // byte-wise spelling of first-`Refuse`-wins one line above: a
760                // push has to satisfy every gate, so it is held to the
761                // tightest one, and the line the client reads is that gate's
762                // own. Asked only where `pack_quota` is asked, so a read and a
763                // deletion are never given a ceiling to trip over.
764                if let Some(asked) = gate.pack_ceiling(&repo.namespace) {
765                    let tighter = match &ceiling {
766                        Some(held) => asked.bytes < held.bytes,
767                        None => true,
768                    };
769                    if tighter {
770                        ceiling = Some(asked);
771                    }
772                }
773                // …and the same tightest-wins rule for the LFS wire, for the
774                // same reason: a transfer has to satisfy every gate, so it is
775                // held to the smallest room any of them will admit, and the
776                // sentence the tenant reads is that gate's own.
777                if let Some(asked) = gate.lfs_allowance(&repo.namespace) {
778                    let tighter = match &allowance {
779                        Some(held) => asked.bytes < held.bytes,
780                        None => true,
781                    };
782                    if tighter {
783                        allowance = Some(asked);
784                    }
785                }
786            }
787        }
788        Ok(PlanOutcome {
789            notice: (!notices.is_empty()).then(|| notices.join("; ")),
790            // The tightest gate's number and the standing taken before any gate
791            // looked, carried side by side rather than folded into one type:
792            // see `ServiceRequest::pack_mark` for why the ceiling stays exactly
793            // what `PlanGate::pack_ceiling` returned.
794            pack_mark: ceiling.is_some().then_some(pack_mark),
795            ceiling,
796            // The tightest gate's number, bound to the standing taken above it.
797            // `mark` is dropped unused when no gate asked for an allowance,
798            // which costs one map read and holds nothing: there is no
799            // reservation in a mark to leak.
800            allowance: allowance
801                .map(|allowance| gunnar_lfs::NamespaceAllowance::new(mark, allowance)),
802        })
803    }
804}
805
806/// What [`Dispatch::ask_plan`] learned: everything the gate chain wants carried
807/// into the transfer, as opposed to the refusals, which are returned as errors
808/// and never reach here.
809///
810/// A struct and not a tuple because both fields are `Option` and both are
811/// about the plan: a `(Option<String>, Option<PackCeiling>)` at the call site
812/// is two anonymous maybes that the next field added would make three.
813#[derive(Default)]
814struct PlanOutcome {
815    /// See [`crate::ServiceRequest::plan_notice`].
816    notice: Option<String>,
817    /// See [`crate::ServiceRequest::pack_ceiling`].
818    ceiling: Option<crate::plan::PackCeiling>,
819    /// See [`crate::ServiceRequest::pack_mark`]. `Some` exactly when `ceiling`
820    /// is: a mark with no number to be judged against is bookkeeping nobody
821    /// reads, and a number with no mark is the bug this closed.
822    pack_mark: Option<gunnar_lfs::Mark>,
823    /// See [`crate::ServiceRequest::lfs_allowance`]. A third `Option` on this
824    /// struct, and the reason the struct exists rather than a tuple.
825    allowance: Option<gunnar_lfs::NamespaceAllowance>,
826}
827
828// ───────────────────────────────────────────────────────────────────────────
829// The server
830// ───────────────────────────────────────────────────────────────────────────
831
832struct Shared {
833    dispatch: Dispatch,
834    transport: Arc<dyn GitTransport>,
835    exec_grace: Option<Duration>,
836    /// Whether `auth_none` on [`ANONYMOUS_SSH_USER`] is accepted.
837    allow_anonymous: bool,
838    admission: Arc<Admission>,
839    metrics: Arc<ServerMetrics>,
840    /// **Are the stores open?** See [`GunnarSsh::with_readiness`].
841    ready: Arc<std::sync::atomic::AtomicBool>,
842    /// [`SshConfig::max_auth_attempts`], carried here because the CONNECTION
843    /// enforces it and russh 0.61.2 does not. See
844    /// [`Connection::auth_attempts`].
845    config_max_auth_attempts: usize,
846    /// [`SshConfig::session_end_grace`] and
847    /// [`SshConfig::session_end_grace_crowded`].
848    session_end_grace: (Duration, Duration),
849    /// The sealed box's tunnel-address door ([`TUNNEL_ADDRESS_VERB`]), or
850    /// `None` on a build with no WireGuard tunnel.
851    tunnel_door: Option<TunnelAddressDoor>,
852    /// The repository-settings door ([`REPO_VERB`]), or `None` on a server
853    /// with no control plane (`serve --no-grpc`).
854    repo_door: Option<RepoDoor>,
855}
856
857/// **The one control verb on this listener** (2026-09-21, T17):
858/// `gunnar-tunnel-address [a.b.c.d/prefix]`.
859///
860/// The sealed installer runs `serve --no-grpc`, so its SSH door is its only
861/// door, and a box's WireGuard tunnel address is per box and not on the
862/// medium. So it arrives here, held to **server-wide Admin** and answered by
863/// the closure `gunnar-server` injects — the SAME function the gRPC doors call.
864/// No argument is a read. Output is `key value` lines on stdout, exit 0; a
865/// refusal is `error: <why>` on stdout, exit 1 (stdout and not stderr, so a
866/// client reading only the data band still gets the sentence). Not a git
867/// verb, so it never enters [`Dispatch::authorise`] or the transfer path.
868pub const TUNNEL_ADDRESS_VERB: &str = "gunnar-tunnel-address";
869
870/// **The operator's fire drill** (stinkies ssh-double-exec-error, 2026-09-30): ONE exec
871/// verb, held to server-wide Admin like [`TUNNEL_ADDRESS_VERB`], that says exactly one
872/// ERROR naming itself a drill and who asked — what the estate's watch raises on purpose
873/// to prove its alert path end to end. Nothing is wrong when it is said; no other caller
874/// can raise the server's ERROR at will.
875pub const FIRE_DRILL_VERB: &str = "gunnar-fire-drill";
876
877/// **A member's SSH public keys, for an encrypted repository's key ring**
878/// (2026-10-06): `gunnar-recipient-keys '<namespace>/<repository>' '<member>'`.
879///
880/// What `gunnar crypt add-member NAME --from-server` asks when it is given no
881/// control address — on an installed estate the forge's SSH door (tcp/2222
882/// through the front) is the one door a member's key opens; the control plane
883/// is bound to the WireGuard overlay. The SAME rule as `Repos.RecipientKeys`:
884/// the caller reads the repository (otherwise it is told what a stranger is
885/// told, the one "not found, or no access" sentence), and so must the person
886/// named — gunnar hands out keys only for someone who could read it anyway,
887/// never a directory of everybody's keys. Output is one OpenSSH public key
888/// line per key on stdout, exit 0; a refusal is `error: <why>` on stdout,
889/// exit [`EXIT_DENIED`] (stdout, like [`TUNNEL_ADDRESS_VERB`]'s). An anonymous
890/// session is refused the channel. Not a git verb: it never enters
891/// [`Dispatch::authorise`] or the transfer path, and it changes nothing.
892pub const RECIPIENT_KEYS_VERB: &str = "gunnar-recipient-keys";
893
894/// **Who this key is** (wiring tracer B 2026-10-08): `gunnar-whoami` answers the principal id
895/// the session authenticated as, one line, exit 0 — what `gunnar crypt init` names its first
896/// member by (the account name `upgrade-pq --from-server` looks members up by). Reveals nothing
897/// the key's holder does not already know. An anonymous session is refused the channel. Not a
898/// git verb.
899pub const WHOAMI_VERB: &str = "gunnar-whoami";
900
901/// **A repository's settings, over the forge's door** (owner, 2026-10-14):
902/// `gunnar-repo show '<namespace>/<repository>'`,
903/// `gunnar-repo visibility '<namespace>/<repository>' public|private`,
904/// `gunnar-repo default-branch '<namespace>/<repository>' '<refs/heads/…>'`.
905///
906/// What the client `gunnar repo show|visibility|default-branch` asks. On an
907/// installed estate the forge's SSH door (tcp/2222 through the front) is the
908/// one door a person's key opens; the control plane is bound to the WireGuard
909/// overlay. The verb decides NOTHING itself: it hands the authenticated
910/// principal and the ask to [`RepoDoor`], which `gunnar-server` answers with
911/// the SAME `Repos.Get` / `Repos.SetVisibility` / `Repos.SetDefaultBranch`
912/// handlers the control plane mounts — so the account-admin rule for
913/// visibility, the repository-Admin rule for the default branch, the audit
914/// event and the browse replica's notice are the gRPC door's, not a copy.
915/// Output is `key value` lines on stdout, exit 0; a refusal is
916/// `error: <why>` on stdout, exit [`EXIT_DENIED`]. An anonymous session is
917/// refused the channel. Not a git verb: it never enters
918/// [`Dispatch::authorise`] or the transfer path.
919pub const REPO_VERB: &str = "gunnar-repo";
920
921/// What a [`REPO_VERB`] line asks.
922#[derive(Debug, Clone, PartialEq, Eq)]
923pub enum RepoAction {
924    /// `Repos.Get`.
925    Show,
926    /// `Repos.SetVisibility`: `true` is public.
927    Visibility(bool),
928    /// `Repos.SetDefaultBranch`, the branch exactly as sent (the server
929    /// validates it; the client expands a short name).
930    DefaultBranch(String),
931    /// `Repos.Create` of an empty repository in this object format (`true`
932    /// is sha256): what a client that knows its own format asks before its
933    /// first push, since a push-made repository can only be sha1.
934    Create { sha256: bool },
935}
936
937/// One [`REPO_VERB`] ask, with who asked: the principal the session's key
938/// authenticated as (re-read at the exec, never cached) and that key's
939/// fingerprint.
940#[derive(Debug, Clone, PartialEq, Eq)]
941pub struct RepoAsk {
942    pub principal: String,
943    pub fingerprint: String,
944    pub repo: RepoRef,
945    pub action: RepoAction,
946}
947
948/// What answers [`REPO_VERB`]: the ask in, the `key value` lines or the
949/// refusal's sentence out.
950pub type RepoDoor = Arc<
951    dyn Fn(RepoAsk) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, String>> + Send>>
952        + Send
953        + Sync,
954>;
955
956/// Parse a [`REPO_VERB`] exec line. `None`: not this verb. `Some(Err)`: this
957/// verb, malformed. Each argument may be single-quoted the way git quotes a
958/// path; the path goes through the one sandbox
959/// ([`gunnar_core::resolve_repo_path`]).
960pub fn parse_repo_verb(raw: &[u8]) -> Option<Result<(RepoRef, RepoAction), String>> {
961    let text = std::str::from_utf8(raw).ok()?;
962    let mut tokens = text.split(' ').filter(|t| !t.is_empty());
963    if tokens.next()? != REPO_VERB {
964        return None;
965    }
966    let usage = || {
967        format!(
968            "the line is not `{REPO_VERB} show <namespace>/<repository>`, \
969             `{REPO_VERB} visibility <namespace>/<repository> public|private` or \
970             `{REPO_VERB} default-branch <namespace>/<repository> <branch>` or \
971             `{REPO_VERB} create <namespace>/<repository> sha1|sha256`"
972        )
973    };
974    if raw.len() > crate::command::MAX_EXEC_LEN || raw.iter().any(|b| *b < 0x20 || *b == 0x7f) {
975        return Some(Err(usage()));
976    }
977    let mut args = Vec::new();
978    for token in tokens {
979        match crate::command::unquote(token) {
980            Ok(arg) => args.push(arg),
981            Err(_) => return Some(Err(usage())),
982        }
983    }
984    let (sub, path, value) = match args.as_slice() {
985        [sub, path] => (sub.as_str(), path.as_str(), None),
986        [sub, path, value] => (sub.as_str(), path.as_str(), Some(value.as_str())),
987        _ => return Some(Err(usage())),
988    };
989    let action = match (sub, value) {
990        ("show", None) => RepoAction::Show,
991        ("visibility", Some("public")) => RepoAction::Visibility(true),
992        ("visibility", Some("private")) => RepoAction::Visibility(false),
993        ("visibility", Some(other)) => {
994            return Some(Err(format!("visibility is `public` or `private`, not `{}`", other.escape_default())))
995        }
996        ("default-branch", Some(branch)) if !branch.is_empty() => RepoAction::DefaultBranch(branch.to_string()),
997        ("create", Some("sha256")) => RepoAction::Create { sha256: true },
998        ("create", Some("sha1")) => RepoAction::Create { sha256: false },
999        ("create", Some(other)) => {
1000            return Some(Err(format!("an object format is `sha1` or `sha256`, not `{}`", other.escape_default())))
1001        }
1002        _ => return Some(Err(usage())),
1003    };
1004    let repo = match gunnar_core::resolve_repo_path(path) {
1005        Ok(repo) => repo,
1006        Err(err) => return Some(Err(err.to_string())),
1007    };
1008    Some(Ok((repo, action)))
1009}
1010
1011/// Parse a [`RECIPIENT_KEYS_VERB`] exec line. `None`: not this verb.
1012/// `Some(Err)`: this verb, malformed. The path and the member may each be
1013/// single-quoted the way git quotes a path; the path goes through the one
1014/// sandbox ([`gunnar_core::resolve_repo_path`]) and the member through the
1015/// principal id rule.
1016pub fn parse_recipient_keys_verb(raw: &[u8]) -> Option<Result<(RepoRef, String), String>> {
1017    let text = std::str::from_utf8(raw).ok()?;
1018    let (verb, rest) = text.split_once(' ').unwrap_or((text, ""));
1019    if verb != RECIPIENT_KEYS_VERB {
1020        return None;
1021    }
1022    let usage = || format!("the line is not `{RECIPIENT_KEYS_VERB} <namespace>/<repository> <member>`");
1023    if raw.len() > crate::command::MAX_EXEC_LEN || raw.iter().any(|b| *b < 0x20 || *b == 0x7f) {
1024        return Some(Err(usage()));
1025    }
1026    let Some((path, member)) = rest.trim().rsplit_once(' ') else {
1027        return Some(Err(usage()));
1028    };
1029    let (Ok(path), Ok(member)) = (crate::command::unquote(path.trim()), crate::command::unquote(member.trim())) else {
1030        return Some(Err(usage()));
1031    };
1032    let repo = match gunnar_core::resolve_repo_path(&path) {
1033        Ok(repo) => repo,
1034        Err(err) => return Some(Err(err.to_string())),
1035    };
1036    if gunnar_core::Principal::new(member.clone(), member.clone(), gunnar_core::PrincipalKind::User).is_err() {
1037        return Some(Err(format!("`{}` is not an account name", member.escape_default())));
1038    }
1039    Some(Ok((repo, member)))
1040}
1041
1042/// **The answer to [`RECIPIENT_KEYS_VERB`]** for `caller`: `(exit status,
1043/// stdout)`. Reads the identity table and the catalog; changes nothing.
1044pub fn recipient_keys_answer(control: &Control, caller: &str, repo: &RepoRef, member: &str) -> (u32, String) {
1045    if control.is_twin() {
1046        let said = DispatchError::Twin { principal: caller.to_string(), repo: repo.clone() }.client_message().unwrap_or_default();
1047        return (EXIT_DENIED, format!("error: {said}\n"));
1048    }
1049    let hidden = || {
1050        let said = DispatchError::NoSuchRepo(repo.clone()).client_message().unwrap_or_default();
1051        (EXIT_DENIED, format!("error: {said}\n"))
1052    };
1053    // The member is told Gunnar could not answer; what broke is the operator's (the journal).
1054    let fault = |err: gunnar_core::Error| {
1055        tracing::error!(target: "gunnar::ssh", event = "ssh.recipient_keys.fault", %err, "gunnar-recipient-keys: the identity table or the catalog could not be read");
1056        (EXIT_DENIED, "error: Gunnar could not answer this just now; try again in a while\n".to_string())
1057    };
1058    let entry = match control.catalog().get(repo) {
1059        Ok(Some(entry)) => entry,
1060        Ok(None) => return hidden(),
1061        Err(err) => return fault(err),
1062    };
1063    match control.auth().may(caller, &entry, gunnar_core::Level::Read) {
1064        Ok(true) => {}
1065        Ok(false) => return hidden(),
1066        Err(err) => return fault(err),
1067    }
1068    match control.auth().may(member, &entry, gunnar_core::Level::Read) {
1069        Ok(true) => {}
1070        Ok(false) => {
1071            return (
1072                EXIT_DENIED,
1073                format!(
1074                    "error: {member} cannot read {}/{}: grant them access before adding them as a member of its key ring\n",
1075                    repo.namespace(),
1076                    repo.name()
1077                ),
1078            )
1079        }
1080        Err(err) => return fault(err),
1081    }
1082    match control.auth().recipient_keys_of(member) {
1083        Ok(keys) if keys.is_empty() => (EXIT_DENIED, format!("error: gunnar holds no SSH key for {member}\n")),
1084        Ok(keys) => (0, keys.iter().map(|k| format!("{}\n", k.openssh.trim())).collect()),
1085        Err(err) => fault(err),
1086    }
1087}
1088
1089/// What answers [`TUNNEL_ADDRESS_VERB`]: the address (empty = a read) in, the
1090/// answer's lines or the refusal out.
1091pub type TunnelAddressDoor = Arc<dyn Fn(&str) -> Result<String, String> + Send + Sync>;
1092
1093/// Parse a [`TUNNEL_ADDRESS_VERB`] exec line. `None`: not this verb. `Some(Err)`:
1094/// this verb, malformed. Printable ASCII only, at most one argument, short.
1095pub fn parse_tunnel_address_verb(raw: &[u8]) -> Option<Result<String, String>> {
1096    let text = std::str::from_utf8(raw).ok()?;
1097    let mut tokens = text.split(' ').filter(|t| !t.is_empty());
1098    if tokens.next()? != TUNNEL_ADDRESS_VERB {
1099        return None;
1100    }
1101    if raw.len() > 64 || !raw.iter().all(|b| (0x20..0x7f).contains(b)) {
1102        return Some(Err("the line is not `gunnar-tunnel-address [a.b.c.d/prefix]`".to_string()));
1103    }
1104    let address = tokens.next().unwrap_or("").to_string();
1105    if tokens.next().is_some() {
1106        return Some(Err("one argument at most: `gunnar-tunnel-address [a.b.c.d/prefix]`".to_string()));
1107    }
1108    Some(Ok(address))
1109}
1110
1111/// **The key exchanges gunnar's SSH door offers, post-quantum FIRST**
1112/// (2026-10-06, the PQ transport lane).
1113///
1114/// `mlkem768x25519-sha256` (draft-ietf-sshm-mlkem-hybrid-kex: ML-KEM-768 and
1115/// X25519, the session key from both) is what OpenSSH ≥ 9.9 offers first and
1116/// 10.0 makes its default — russh 0.63 runs it on RustCrypto's `ml-kem`. The
1117/// classical exchanges stay after it, so an older client still connects
1118/// (preferred, never required). `sntrup761x25519-sha512` is not here: russh
1119/// does not implement it.
1120///
1121/// **gunnar's own list, not russh's default**, though the two agree today:
1122/// a russh bump that reorders or drops its `Preferred::DEFAULT` must not
1123/// change what this door offers without a red test here. SSH lets the
1124/// CLIENT's order decide; the server's job is to offer the hybrid at all.
1125pub static KEX_ORDER: &[russh::kex::Name] = &[
1126    russh::kex::MLKEM768X25519_SHA256,
1127    russh::kex::CURVE25519,
1128    russh::kex::CURVE25519_PRE_RFC_8731,
1129    russh::kex::DH_GEX_SHA256,
1130    russh::kex::DH_G18_SHA512,
1131    russh::kex::DH_G17_SHA512,
1132    russh::kex::DH_G16_SHA512,
1133    russh::kex::DH_G15_SHA512,
1134    russh::kex::DH_G14_SHA256,
1135    russh::kex::EXTENSION_SUPPORT_AS_CLIENT,
1136    russh::kex::EXTENSION_SUPPORT_AS_SERVER,
1137    russh::kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT,
1138    russh::kex::EXTENSION_OPENSSH_STRICT_KEX_AS_SERVER,
1139];
1140
1141/// russh's algorithm preferences with gunnar's [`KEX_ORDER`]; ciphers, MACs
1142/// and host-key algorithms stay russh's.
1143pub fn preferred() -> russh::Preferred {
1144    russh::Preferred { kex: std::borrow::Cow::Borrowed(KEX_ORDER), ..russh::Preferred::DEFAULT }
1145}
1146
1147/// gunnar's SSH listener.
1148pub struct GunnarSsh {
1149    config: SshConfig,
1150    shared: Arc<Shared>,
1151    /// **A connection that arrives at a refused LOCAL address is dropped**
1152    /// before a byte is spoken (`gunnar serve --wg-only`: the utility
1153    /// network's address — every box-to-box connection runs over the tunnel).
1154    refused_local: Option<RefusedLocal>,
1155}
1156
1157/// Is a connection that arrived at this local address refused?
1158pub type RefusedLocal = Arc<dyn Fn(std::net::IpAddr) -> bool + Send + Sync>;
1159
1160impl GunnarSsh {
1161    /// Build a server over the control plane and the wire seam.
1162    pub fn new(config: SshConfig, control: Control, transport: Arc<dyn GitTransport>) -> Self {
1163        let exec_grace = config.exec_grace;
1164        let allow_anonymous = config.allow_anonymous;
1165        let config_max_auth_attempts = config.max_auth_attempts;
1166        let session_end_grace = (config.session_end_grace, config.session_end_grace_crowded);
1167        // The memory budget travels with the two concurrency numbers: one
1168        // constructor, so a configuration cannot arrive with a governed
1169        // concurrency limit and an unmetered budget. `None` falls back to the
1170        // shipped constant rather than to "unlimited" — an operator who did not
1171        // set a budget wants the default, not the removal of a gate.
1172        let budget = config
1173            .memory_budget_bytes
1174            .unwrap_or(crate::governor::DEFAULT_MEMORY_BUDGET_BYTES);
1175        let wait = config
1176            .admission_wait
1177            .unwrap_or(crate::governor::DEFAULT_ADMISSION_WAIT);
1178        let admission = Arc::new(
1179            if config.per_principal_limit_is_hard {
1180                Admission::hard_capped(
1181                    config.max_concurrent_transfers,
1182                    config.max_transfers_per_principal,
1183                )
1184                .with_budget(budget)
1185            } else {
1186                Admission::with_memory_budget(
1187                    config.max_concurrent_transfers,
1188                    config.max_transfers_per_principal,
1189                    budget,
1190                )
1191            }
1192            // The anonymous door and its width are configured together, so
1193            // an appliance that opens the door cannot forget the ceiling.
1194            .with_anonymous_ceiling(config.max_anonymous_transfers)
1195            .with_wait(wait),
1196        );
1197        // The usage share (`admit-share`): the configured policy, unseeded
1198        // until `with_flow` hands it the durable ledger to seed from.
1199        #[cfg(feature = "admit-share")]
1200        let admission = Arc::new(
1201            Arc::try_unwrap(admission)
1202                .expect("the gate is not shared yet")
1203                .with_policy(config.admit.build()),
1204        );
1205        GunnarSsh {
1206            config,
1207            refused_local: None,
1208            shared: Arc::new(Shared {
1209                dispatch: Dispatch::new(control),
1210                transport,
1211                exec_grace,
1212                allow_anonymous,
1213                admission,
1214                metrics: Arc::new(ServerMetrics::new()),
1215                // Ready unless somebody hands us a flag to watch. A listener
1216                // built by a test has nothing to wait for, and a default of
1217                // "not ready" would make every such test hang on a sentence
1218                // about a boot that never happened.
1219                ready: Arc::new(std::sync::atomic::AtomicBool::new(true)),
1220                config_max_auth_attempts,
1221                session_end_grace,
1222                tunnel_door: None,
1223                repo_door: None,
1224            }),
1225        }
1226    }
1227
1228    /// **Watch a readiness flag**, and answer git traffic with a *retryable*
1229    /// sentence until it is set.
1230    ///
1231    /// # The boot window — MEASURED 2026-08-30
1232    ///
1233    /// After a power cut the appliance answered ssh :2222 and grpc :50051
1234    /// **0.6–0.7 s before its stores had finished their post-crash open**, and
1235    /// a read in that window came back `the repository is unavailable; try
1236    /// again`. That sentence is `gunnar_wire::Error::Store`'s, it arrives with
1237    /// a *denial* exit status, and it is what a client is told when a
1238    /// repository is genuinely broken — so a harness could not tell "not yet"
1239    /// from "gone". The first power-cut door reported **64 acked pushes
1240    /// missing**; a boot one second later found every one of them, and the same
1241    /// window wedged a whole bench run's clone and fetch phases.
1242    ///
1243    /// The fix is not to bind later — PID 1 must answer quickly or an operator
1244    /// watching a console sees a dead box — but to answer **honestly**. While
1245    /// the flag is clear every git exec is refused in git's own protocol with
1246    /// [`EXIT_BUSY`] (75, EX_TEMPFAIL — the same status the admission gate
1247    /// uses, which is the status that means *retry*) and a sentence containing
1248    /// the word **starting**, which is the word a script may key on. Nothing is
1249    /// served, nothing is opened, and nothing looks like a missing repository.
1250    ///
1251    /// A flag and not a trait: `gunnar-ssh` may not read the operating system,
1252    /// and "have the stores finished opening" is `gunnar-server`'s fact to
1253    /// publish, not this crate's to discover.
1254    pub fn with_readiness(mut self, ready: Arc<std::sync::atomic::AtomicBool>) -> Self {
1255        // `Arc::get_mut` cannot be used: `bind` has not run, but `Shared` is
1256        // already behind an `Arc`. Rebuilding it is one allocation at start-up
1257        // and keeps every field in one place.
1258        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
1259            panic!(
1260                "readiness must be set before the listener is shared: {} strong refs",
1261                Arc::strong_count(&shared)
1262            )
1263        });
1264        self.shared = Arc::new(Shared { ready, ..shared });
1265        self
1266    }
1267
1268    /// **Refuse every connection that arrives at a local address `refused`
1269    /// names** — closed at accept, nothing said, logged once per peer at
1270    /// debug. `gunnar serve --wg-only` names the utility address: the public's
1271    /// git clients still reach SSH, a box over plain IP does not.
1272    pub fn with_refused_local(mut self, refused: RefusedLocal) -> Self {
1273        self.refused_local = Some(refused);
1274        self
1275    }
1276
1277    /// **Answer [`TUNNEL_ADDRESS_VERB`]** with `door` — `gunnar-server` hands in
1278    /// its `appliance::tunnel_address_answer`, the function both gRPC doors
1279    /// call. Without one the verb is answered "no WireGuard tunnel".
1280    pub fn with_tunnel_address_door(mut self, door: TunnelAddressDoor) -> Self {
1281        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
1282            panic!(
1283                "the tunnel door must be set before the listener is shared: {} strong refs",
1284                Arc::strong_count(&shared)
1285            )
1286        });
1287        self.shared = Arc::new(Shared { tunnel_door: Some(door), ..shared });
1288        self
1289    }
1290
1291    /// **Answer [`REPO_VERB`]** with `door` — `gunnar-server` hands in the
1292    /// control plane's own `Repos` handlers. Without one the verb is answered
1293    /// "this server answers repository settings only on its control plane".
1294    pub fn with_repo_door(mut self, door: RepoDoor) -> Self {
1295        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
1296            panic!(
1297                "the repository door must be set before the listener is shared: {} strong refs",
1298                Arc::strong_count(&shared)
1299            )
1300        });
1301        self.shared = Arc::new(Shared { repo_door: Some(door), ..shared });
1302        self
1303    }
1304
1305    /// **Give the admission gate the per-namespace flow ledger.**
1306    ///
1307    /// The gate is what all three front doors — this exec line, smart HTTP and
1308    /// the gRPC transfer gate — ask before a transfer, so it is the one place a
1309    /// refusal can be charged to a tenant once rather than three times. The
1310    /// exec line's own `no-grant` denial reaches the same ledger back through
1311    /// `Admission::flow`.
1312    ///
1313    /// A listener built without one refuses exactly as it always did and counts
1314    /// the refusal only in its own `AdmissionStats`, which is what a test that
1315    /// drives `gunnar-ssh` alone wants — and what every build before the ledger
1316    /// existed did.
1317    ///
1318    /// The gate is rebuilt rather than mutated for the reason
1319    /// [`Self::with_readiness`] rebuilds `Shared`: `bind` has not run, but the
1320    /// `Arc` is already there.
1321    pub fn with_flow(mut self, flow: Arc<gunnar_core::FlowLedger>) -> Self {
1322        let shared =
1323            Arc::get_mut(&mut self.shared).expect("the shared state is not published until `bind`");
1324        let gate = Arc::try_unwrap(std::mem::replace(
1325            &mut shared.admission,
1326            Arc::new(Admission::unlimited()),
1327        ))
1328        .unwrap_or_else(|gate| {
1329            panic!(
1330                "the flow ledger must be set before the gate is shared: {} strong refs",
1331                Arc::strong_count(&gate)
1332            )
1333        });
1334        // admit-share: the policy is rebuilt SEEDED from this ledger — its
1335        // table is then a cache of the ledger, filled once per namespace.
1336        #[cfg(feature = "admit-share")]
1337        let gate = gate.with_policy(
1338            self.config
1339                .admit
1340                .build_seeded(Some(Admission::flow_seed(Arc::clone(&flow)))),
1341        );
1342        shared.admission = Arc::new(gate.with_flow(flow));
1343        self
1344    }
1345
1346    /// Share the server-wide counters with the rest of the process.
1347    ///
1348    /// **Not a second set of counters.** `gunnar serve` mints one
1349    /// [`ServerMetrics`] and hands the same `Arc` to this listener and to the
1350    /// git transport, so "connections refused" and "pushes rejected by gate"
1351    /// are numbers about one server rather than two crates' private tallies.
1352    /// A listener built without this one still counts, into a set nobody else
1353    /// can read, which is what a test that only drives `gunnar-ssh` wants.
1354    pub fn with_metrics(mut self, metrics: Arc<ServerMetrics>) -> Self {
1355        let shared =
1356            Arc::get_mut(&mut self.shared).expect("the shared state is not published until `bind`");
1357        shared.metrics = metrics;
1358        self
1359    }
1360
1361    /// The counters this listener writes to.
1362    pub fn metrics(&self) -> &Arc<ServerMetrics> {
1363        &self.shared.metrics
1364    }
1365
1366    /// The russh configuration this server runs with.
1367    ///
1368    /// `methods` is publickey **only**, and stays that way even with the
1369    /// anonymous path open: `none` is available for exactly one username,
1370    /// so advertising it to everybody would make the failure list a lie and
1371    /// invite every client to try something that will not work. russh calls
1372    /// [`Handler::auth_none`] on the client's opening probe regardless of what
1373    /// is advertised, which is the only place it is needed.
1374    ///
1375    /// `password` is not offered and there is no handler for it, so the
1376    /// rejecting default is never the last line of defence.
1377    fn russh_config(&self, host_key: russh::keys::PrivateKey) -> RusshConfig {
1378        RusshConfig {
1379            server_id: russh::SshId::Standard(self.config.server_id.clone().into()),
1380            preferred: preferred(),
1381            methods: MethodSet::from(&[MethodKind::PublicKey][..]),
1382            auth_rejection_time: self.config.auth_rejection_time,
1383            auth_rejection_time_initial: Some(Duration::ZERO),
1384            keys: vec![host_key],
1385            max_auth_attempts: self.config.max_auth_attempts,
1386            inactivity_timeout: self.config.inactivity_timeout,
1387            keepalive_interval: self.config.keepalive_interval,
1388            nodelay: true,
1389            ..RusshConfig::default()
1390        }
1391    }
1392
1393    /// Bind the listener and start accepting.
1394    ///
1395    /// Returns once the socket is bound, so a caller (or a test) knows the
1396    /// real address before the first client connects. That matters for
1397    /// [`SshConfig::ephemeral_loopback`], where the port is chosen by the
1398    /// kernel.
1399    pub async fn bind(self, host_key: russh::keys::PrivateKey) -> std::io::Result<BoundServer> {
1400        let listener = TcpListener::bind(self.config.bind).await?;
1401        let local_addr = listener.local_addr()?;
1402        let russh_config = Arc::new(self.russh_config(host_key));
1403        let shared = self.shared.clone();
1404        let admission = Arc::clone(&self.shared.admission);
1405        let metrics = Arc::clone(&self.shared.metrics);
1406        let (handle, _) = tokio::sync::broadcast::channel::<String>(1);
1407
1408        let join = tokio::spawn(accept_loop(
1409            listener,
1410            russh_config,
1411            Factory { shared },
1412            handle.clone(),
1413            self.refused_local.clone(),
1414        ));
1415
1416        tracing::info!(
1417            target: TARGET_SSH,
1418            %local_addr,
1419            max_concurrent_transfers = ?admission.max_concurrent(),
1420            max_transfers_per_principal = ?admission.max_per_principal(),
1421            per_principal_is_hard = admission.per_principal_is_hard(),
1422            "gunnar ssh listener bound"
1423        );
1424        Ok(BoundServer {
1425            local_addr,
1426            handle,
1427            join,
1428            admission,
1429            metrics,
1430        })
1431    }
1432}
1433
1434/// How long the accept loop waits after the kernel refused it a descriptor,
1435/// before it asks again. A session ending in that window frees what the next
1436/// `accept` needs; nothing shorter gains anything, nothing longer is felt.
1437pub const ACCEPT_BACKOFF: Duration = Duration::from_millis(100);
1438
1439/// Is this `accept` failure the kind the loop must outlive?
1440///
1441/// The resource ones — no descriptor left in the process (`EMFILE`) or in the
1442/// system (`ENFILE`), no buffer or memory for the socket — and the one the
1443/// client causes, `ECONNABORTED` (it hung up between SYN and `accept`).
1444/// Everything else (`EBADF`, `EINVAL`, a closed listener) is the listener
1445/// itself being gone, and the loop must return that.
1446pub fn accept_error_is_transient(err: &std::io::Error) -> bool {
1447    // Linux errno values, spelled here rather than through a `libc` crate this
1448    // crate does not otherwise carry: ENOMEM 12, ENFILE 23, EMFILE 24,
1449    // ECONNABORTED 103, ENOBUFS 105.
1450    matches!(err.raw_os_error(), Some(12 | 23 | 24 | 103 | 105))
1451        || matches!(
1452            err.kind(),
1453            std::io::ErrorKind::ConnectionAborted | std::io::ErrorKind::OutOfMemory
1454        )
1455}
1456
1457/// **The accept loop, owned here so that it cannot die of `EMFILE`.**
1458///
1459/// russh's `run_on_socket` is the same loop with one difference that cost a
1460/// whole appliance: its `Err(e) => return Err(e)` on `accept`. MEASURED
1461/// 2026-08-29 on the rig: the 202nd repository touched since boot ran the
1462/// process out of descriptors, the next `accept` failed with `EMFILE`, the
1463/// loop returned, and every later client got `kex_exchange_identification:
1464/// read: Connection reset by peer` until reboot — the kernel kept completing
1465/// handshakes into a backlog nobody drained — while gRPC on its own listener
1466/// answered as if nothing were wrong. A server that has no descriptor for the
1467/// next client must say so and try again in a moment, not resign.
1468///
1469/// So on a transient failure ([`accept_error_is_transient`]) it logs a WARN
1470/// on `gunnar::ssh` — a row on the appliance dashboard — sleeps
1471/// [`ACCEPT_BACKOFF`] and continues. Everything else is what russh did: one
1472/// task per connection, `set_nodelay`, `run_stream`, the session raced against
1473/// the shutdown broadcast so `BoundServer::shutdown` still disconnects every
1474/// live session by application, and session errors go through
1475/// [`note_session_end`] exactly as `Server::handle_session_error` did.
1476async fn accept_loop(
1477    listener: TcpListener,
1478    config: Arc<RusshConfig>,
1479    mut factory: Factory,
1480    shutdown: tokio::sync::broadcast::Sender<String>,
1481    refused_local: Option<RefusedLocal>,
1482) -> std::io::Result<()> {
1483    let mut shutdown_rx = shutdown.subscribe();
1484    let mut refused_in_a_row: u32 = 0;
1485    loop {
1486        tokio::select! {
1487            _ = shutdown_rx.recv() => {
1488                tracing::debug!(target: TARGET_SSH, "ssh listener shutdown requested");
1489                return Ok(());
1490            }
1491            accepted = listener.accept() => {
1492                let (socket, peer) = match accepted {
1493                    Ok(pair) => pair,
1494                    Err(err) if accept_error_is_transient(&err) => {
1495                        refused_in_a_row += 1;
1496                        tracing::warn!(
1497                            target: TARGET_SSH,
1498                            event = "ssh.accept.refused",
1499                            error = %err,
1500                            refused_in_a_row,
1501                            "ssh accept failed: {err}; the listener stays up and retries in {} ms",
1502                            ACCEPT_BACKOFF.as_millis()
1503                        );
1504                        tokio::time::sleep(ACCEPT_BACKOFF).await;
1505                        continue;
1506                    }
1507                    Err(err) => {
1508                        tracing::error!(
1509                            target: TARGET_SSH,
1510                            event = "ssh.accept.failed",
1511                            error = %err,
1512                            "ssh accept failed with an error the listener cannot outlive"
1513                        );
1514                        return Err(err);
1515                    }
1516                };
1517                if let (Some(refused), Ok(local)) = (&refused_local, socket.local_addr()) {
1518                    if refused(local.ip()) {
1519                        tracing::debug!(target: TARGET_SSH, event = "ssh.accept.plain_ip_refused", %peer, %local, "a connection at the utility address is refused (--wg-only): boxes speak over the tunnel");
1520                        drop(socket);
1521                        continue;
1522                    }
1523                }
1524                if refused_in_a_row > 0 {
1525                    tracing::info!(
1526                        target: TARGET_SSH,
1527                        event = "ssh.accept.recovered",
1528                        refused_in_a_row,
1529                        "ssh accept works again"
1530                    );
1531                    refused_in_a_row = 0;
1532                }
1533                let handler = factory.new_client(Some(peer));
1534                let config = Arc::clone(&config);
1535                let mut session_shutdown = shutdown.subscribe();
1536                tokio::spawn(async move {
1537                    if config.nodelay {
1538                        if let Err(err) = socket.set_nodelay(true) {
1539                            tracing::debug!(target: TARGET_SSH, %err, "set_nodelay failed");
1540                        }
1541                    }
1542                    let session = match russh::server::run_stream(config, socket, handler).await {
1543                        Ok(session) => session,
1544                        Err(err) => {
1545                            note_session_end(&err);
1546                            return;
1547                        }
1548                    };
1549                    let handle = session.handle();
1550                    tokio::select! {
1551                        reason = session_shutdown.recv() => {
1552                            let reason = reason.unwrap_or_default();
1553                            let _ = handle
1554                                .disconnect(russh::Disconnect::ByApplication, reason, String::new())
1555                                .await;
1556                        }
1557                        ended = session => {
1558                            if let Err(err) = ended {
1559                                note_session_end(&err);
1560                            }
1561                        }
1562                    }
1563                });
1564            }
1565        }
1566    }
1567}
1568
1569/// A bound, running listener.
1570pub struct BoundServer {
1571    local_addr: SocketAddr,
1572    /// The shutdown broadcast every live session and the accept loop listen on.
1573    handle: tokio::sync::broadcast::Sender<String>,
1574    join: JoinHandle<std::io::Result<()>>,
1575    admission: Arc<Admission>,
1576    metrics: Arc<ServerMetrics>,
1577}
1578
1579impl BoundServer {
1580    /// The address actually bound, with the kernel's choice of port filled in.
1581    pub fn local_addr(&self) -> SocketAddr {
1582        self.local_addr
1583    }
1584
1585    /// What the admission gate has done since this listener started.
1586    ///
1587    /// This is the visibility half of the fix. A limit whose saturation cannot
1588    /// be read is the same undiagnosable "git is slow" the accidental 512-thread
1589    /// ceiling produced; `admitted`, `refused_*` and `peak_in_flight` are what
1590    /// an operator, or a test, looks at instead.
1591    pub fn admission(&self) -> AdmissionStats {
1592        self.admission.stats()
1593    }
1594
1595    /// The admission gate itself, so a **second transport can route through the
1596    /// SAME one** (one implementation).
1597    ///
1598    /// `gunnar-http` is a second front door onto the same server, and a git
1599    /// transfer over HTTP has to count against the same server-wide and
1600    /// per-principal ceilings — and the same memory budget — as one over SSH,
1601    /// or an appliance's real concurrency is silently twice the number written
1602    /// on the box. Handing out the `Arc` (rather than rebuilding a gate from
1603    /// the same numbers) is what makes it one gate with one set of counters,
1604    /// not two that happen to agree: on 2026-08-31 the HTTP front was found with no gate
1605    /// at all, and two gates would have been the next-worst answer.
1606    pub fn admission_gate(&self) -> Arc<Admission> {
1607        Arc::clone(&self.admission)
1608    }
1609
1610    /// The server-wide counters.
1611    pub fn metrics(&self) -> gunnar_core::observe::MetricsSnapshot {
1612        self.metrics.snapshot()
1613    }
1614
1615    /// The counter block itself, for a caller that wants to keep reading it
1616    /// after this listener is consumed by [`BoundServer::shutdown`].
1617    pub fn metrics_handle(&self) -> &Arc<ServerMetrics> {
1618        &self.metrics
1619    }
1620
1621    /// Ask every session to disconnect, then wait for the accept loop to stop
1622    /// — for at most [`SSH_SHUTDOWN_GRACE`].
1623    ///
1624    /// **Bounded, the way the gRPC half already is** (`GRPC_SHUTDOWN_GRACE` in
1625    /// gunnar-server): the accept loop only ends when its sessions do, and a
1626    /// session whose peer never answers the disconnect — MEASURED 2026-09-28 on
1627    /// loki, a primary's replication session into a standby being stopped, 1 run
1628    /// in 17 under a parallel test binary — held `shutdown` for an hour. A
1629    /// shutdown that can hang forever is a restart, a drain or a growth that can
1630    /// hang forever. Past the grace the listener task is aborted and the reason
1631    /// is logged; the sessions go with the process.
1632    pub async fn shutdown(self, reason: &str) -> std::io::Result<()> {
1633        self.shutdown_within(reason, SSH_SHUTDOWN_GRACE).await
1634    }
1635
1636    /// [`Self::shutdown`] with an explicit grace (tests use a short one).
1637    pub async fn shutdown_within(self, reason: &str, grace: Duration) -> std::io::Result<()> {
1638        let _ = self.handle.send(reason.to_string());
1639        let abort = self.join.abort_handle();
1640        match tokio::time::timeout(grace, self.join).await {
1641            Ok(Ok(result)) => result,
1642            Ok(Err(err)) if err.is_cancelled() => Ok(()),
1643            Ok(Err(err)) => Err(std::io::Error::other(err)),
1644            Err(_elapsed) => {
1645                tracing::warn!(
1646                    target: TARGET_SSH,
1647                    ?grace,
1648                    "the SSH listener did not stop within its grace: a session did not \
1649                     answer the disconnect. Aborting the listener so the shutdown completes."
1650                );
1651                abort.abort();
1652                Ok(())
1653            }
1654        }
1655    }
1656}
1657
1658struct Factory {
1659    shared: Arc<Shared>,
1660}
1661
1662impl Server for Factory {
1663    type Handler = Connection;
1664
1665    fn new_client(&mut self, peer_addr: Option<SocketAddr>) -> Connection {
1666        // ── ONE span per connection, and one id ────────────────────────────
1667        //
1668        // `principal` and `repo` are declared `Empty` and filled in by
1669        // `auth_publickey` and `exec_request` as they become known, so every
1670        // line emitted under this connection carries all three WITHOUT the
1671        // call site repeating them. Before this, connection identity in the
1672        // SSH logs was an ad-hoc `?self.peer` copy-pasted to fourteen call
1673        // sites, and there was no way at all to tie two lines together.
1674        let conn = RequestId::next();
1675        let span = tracing::info_span!(
1676            target: TARGET_SSH,
1677            "ssh.connection",
1678            %conn,
1679            peer = tracing::field::Empty,
1680            principal = tracing::field::Empty,
1681            repo = tracing::field::Empty,
1682        );
1683        if let Some(peer) = peer_addr {
1684            span.record("peer", tracing::field::display(peer));
1685        }
1686        self.shared.metrics.add_connections_opened(1);
1687        span.in_scope(|| tracing::debug!(target: TARGET_SSH, "ssh connection accepted"));
1688        Connection {
1689            shared: self.shared.clone(),
1690            span,
1691            principal: None,
1692            key_fingerprint: None,
1693            channels: HashMap::new(),
1694            exec_seen: Arc::new(AtomicBool::new(false)),
1695            misuse_said: false,
1696            auth_attempts: 0,
1697            session_over: tokio::sync::watch::channel(false).0,
1698        }
1699    }
1700
1701    /// The session loop stopped. Usually because the client hung up, which is
1702    /// how every git client ends a transfer and is not news.
1703    ///
1704    /// A free function does the deciding, for the same reason
1705    /// [`note_teardown`] is one: the rule can then be asserted without standing
1706    /// up a russh session, and what is asserted is the line an operator reads.
1707    fn handle_session_error(&mut self, error: anyhow::Error) {
1708        note_session_end(&error);
1709    }
1710}
1711
1712/// One open session channel, before its exec request arrives.
1713struct OpenChannel {
1714    /// The channel itself. The exec request carries only a [`ChannelId`], and
1715    /// the byte stream can only be taken from the `Channel`.
1716    channel: Channel<Msg>,
1717    /// The value of `GIT_PROTOCOL`, if the client sent it and
1718    /// [`ENV_ALLOWLIST`] accepted the name. Kept beside the channel it was sent
1719    /// on so that it cannot outlive it or be read for a different one.
1720    git_protocol: Option<String>,
1721}
1722
1723/// One client connection.
1724struct Connection {
1725    shared: Arc<Shared>,
1726    /// The `ssh.connection` span every line from this connection is emitted
1727    /// under. Cloned (cheap: it is an id, not the data) rather than borrowed
1728    /// at each entry point, because entering it borrows `self` immutably while
1729    /// every handler needs `&mut self`.
1730    span: tracing::Span,
1731    /// Set by `auth_publickey`, and the only thing that authorises anything.
1732    principal: Option<Principal>,
1733    /// The key `auth_publickey` authenticated this session with.
1734    ///
1735    /// Beside `principal` rather than inside it, because a principal is a
1736    /// server-side identity that may hold several keys and this is the one that
1737    /// made *this* connection. Stays `None` for the anonymous path, which
1738    /// authenticated with no key at all.
1739    key_fingerprint: Option<Fingerprint>,
1740    channels: HashMap<ChannelId, OpenChannel>,
1741    exec_seen: Arc<AtomicBool>,
1742    /// A request for a channel this connection does not hold open (a client's second
1743    /// `exec` on one channel) was said already: WARN once per connection, then quiet —
1744    /// a client's protocol misuse is never the server's ERROR (stinkies ssh-double-exec-error).
1745    misuse_said: bool,
1746    /// **How many publickey offers this connection has had refused.**
1747    ///
1748    /// Counted here because russh does not count it. `SshConfig::    /// max_auth_attempts` is handed to russh 0.61.2's `Config`, which declares
1749    /// the field, defaults it to 10 and `Debug`-prints it — and never compares
1750    /// it to anything. VERIFIED by reading the whole of that crate's source:
1751    /// the only three occurrences are the declaration, the default and the
1752    /// `Debug` impl. (russh 0.63 does compare it, and carries tests named for
1753    /// exactly this; gunnar is pinned to 0.61.2 through an exact `ssh-key`
1754    /// dependency, and a limit that works is not worth an SSH-stack bump to
1755    /// obtain.)
1756    ///
1757    /// So until 2026-09-11 a connection could offer keys without bound: a
1758    /// dictionary of public keys against the identity table, at
1759    /// `auth_rejection_time` apiece, on one TCP connection that never had to be
1760    /// re-established.
1761    auth_attempts: usize,
1762    /// **Set when this session ends** — by `Drop`, because russh drops the
1763    /// handler on every way a session ends (hang-up, unanswered keepalives,
1764    /// inactivity, protocol error, shutdown). Every transfer's writer watches
1765    /// it ([`SessionBound`]): russh's own writer parks on the SSH window and
1766    /// nothing wakes it when the session dies, which left a fetch to a dead
1767    /// client holding its fan worker and admission slot forever (cov-13
1768    /// finding 4, 2026-10-06).
1769    session_over: tokio::sync::watch::Sender<bool>,
1770}
1771
1772impl Drop for Connection {
1773    fn drop(&mut self) {
1774        self.session_over.send_replace(true);
1775    }
1776}
1777
1778/// Why a session that authenticated may no longer act.
1779///
1780/// Two variants and not a `bool`, for [`Control::principal_for_key`]'s reason:
1781/// "nobody holds this key any more" and "the identity table could not be read"
1782/// call for entirely different operator action, and both refuse.
1783#[derive(Debug)]
1784enum Stale {
1785    /// The key that opened this session no longer resolves to anyone, or
1786    /// resolves to somebody else. Revoked, expired, or re-registered.
1787    Revoked,
1788    /// The identity table could not be read. Refused, because failing open is
1789    /// not on the table — the same answer `auth_publickey` gives.
1790    Unreadable(gunnar_core::Error),
1791}
1792
1793impl Connection {
1794    /// **Who this session may act as right now — asked again, per exec.**
1795    ///
1796    /// `auth_publickey` caches the principal, and until 2026-09-11 every exec
1797    /// on the connection was authorised out of that cache. Revocation
1798    /// therefore took effect immediately for a NEW connection and never for an
1799    /// open one: `DocRef::read` caches nothing, so the identity table was
1800    /// current the whole time and nothing asked it. A long-lived session — and
1801    /// git over SSH holds one open for as long as a client wants to keep
1802    /// pushing — outlived the credential that opened it, for as long as it
1803    /// liked. An operator who revokes a leaked key and sees the revocation take
1804    /// effect on their own next connection has every reason to believe it took
1805    /// effect everywhere.
1806    ///
1807    /// Cheap enough to do per exec: it is the same single
1808    /// `principal_by_fingerprint` lookup the connection already pays once at
1809    /// auth, and an exec is about to fork a git process and move a pack.
1810    ///
1811    /// A key that now belongs to a DIFFERENT identity is refused rather than
1812    /// adopted. This session authenticated as somebody; if the key has since
1813    /// been registered to somebody else, this is not that session, and
1814    /// continuing as the new holder would make re-registration a way to inherit
1815    /// an open connection.
1816    ///
1817    /// The anonymous path has no fingerprint — it authenticated with no key at
1818    /// all — so there is nothing to re-read, and `Principal::anonymous()` is not
1819    /// in the identity table to be revoked from. It keeps the cached value, and
1820    /// what anonymity may DO is still `Dispatch::authorise`'s answer.
1821    fn live_principal(&self) -> Result<Principal, Stale> {
1822        let Some(cached) = self.principal.clone() else {
1823            return Err(Stale::Revoked);
1824        };
1825        let Some(fingerprint) = self.key_fingerprint.as_ref() else {
1826            return Ok(cached);
1827        };
1828        match self
1829            .shared
1830            .dispatch
1831            .control()
1832            .principal_for_key(fingerprint)
1833        {
1834            Ok(Some(now)) if now.id == cached.id => Ok(now),
1835            Ok(_) => Err(Stale::Revoked),
1836            Err(err) => Err(Stale::Unreadable(err)),
1837        }
1838    }
1839
1840    /// **Count a refused publickey offer, and say whether this connection has
1841    /// had its last one.**
1842    ///
1843    /// One place, called from every refusing arm of `auth_publickey`, so a new
1844    /// refusal reason cannot arrive uncounted — which is how a limit stops
1845    /// applying to exactly the case somebody added last.
1846    ///
1847    /// `0` is no limit, which is russh's own reading of the same field and the
1848    /// reading an operator who writes `0` expects.
1849    ///
1850    /// The `none` probe is deliberately NOT counted. Every OpenSSH client opens
1851    /// with one to learn the method list, so counting it would spend one of the
1852    /// six on the ordinary first packet of an ordinary session — and it carries
1853    /// no key, so it cannot be the dictionary attempt the limit is for.
1854    fn refused_auth(&mut self) -> bool {
1855        self.auth_attempts += 1;
1856        let max = self.shared.config_max_auth_attempts;
1857        max > 0 && self.auth_attempts >= max
1858    }
1859
1860    /// The error that ends a connection which has offered too many keys.
1861    ///
1862    /// An `Err` out of a handler, because russh 0.61.2's `auth_publickey` is
1863    /// handed no session and no handle, so there is nothing to call
1864    /// `disconnect` on — and the run loop treating a handler error as the end
1865    /// of the session is exactly the outcome wanted. It is a `warn!`, not a
1866    /// `debug!`: one of these is a client with a full agent, and a hundred is
1867    /// somebody reading the identity table one key at a time.
1868    fn too_many_auth_attempts(&self) -> anyhow::Error {
1869        tracing::warn!(
1870            target: TARGET_SSH,
1871            event = "ssh.auth.too_many",
1872            attempts = self.auth_attempts,
1873            max = self.shared.config_max_auth_attempts,
1874            "disconnecting: too many refused publickey offers on one connection"
1875        );
1876        self.shared.metrics.add_connections_refused(1);
1877        anyhow::anyhow!(
1878            "too many authentication failures ({} refused publickey offers)",
1879            self.auth_attempts
1880        )
1881    }
1882}
1883
1884impl Handler for Connection {
1885    type Error = anyhow::Error;
1886
1887    /// **The anonymous read path.** Accept the `none` method for exactly
1888    /// one username, and refuse it for every other.
1889    ///
1890    /// The refusing branch is byte-for-byte what russh's default handler did
1891    /// before this method existed: `Auth::reject()`, which makes russh drop
1892    /// `MethodKind::None` from the advertised set and let the client proceed to
1893    /// publickey. So a client that does not ask for anonymity is on exactly the
1894    /// path it was on yesterday.
1895    ///
1896    /// The accepting branch does **not** consult the catalog, the repository or
1897    /// the ACL: it establishes *who* is asking (nobody), and *what nobody may
1898    /// do* is `Dispatch::authorise`'s answer, through
1899    /// [`AuthTable::may`](gunnar_core::AuthTable::may), like everyone else.
1900    /// Deciding here would be a second ACL.
1901    async fn auth_none(&mut self, user: &str) -> Result<Auth, Self::Error> {
1902        let span = self.span.clone();
1903        let _entered = span.enter();
1904        if !self.shared.allow_anonymous || user != ANONYMOUS_SSH_USER {
1905            // Not a refusal worth a `warn!`: EVERY OpenSSH client opens with a
1906            // `none` probe to learn the method list, so this is the ordinary
1907            // first packet of an ordinary authenticated session.
1908            tracing::debug!(
1909                target: TARGET_SSH,
1910                event = "ssh.auth.none.refused",
1911                reason = if self.shared.allow_anonymous {
1912                    "not-the-anonymous-user"
1913                } else {
1914                    "anonymous-access-disabled"
1915                },
1916                decorative_ssh_user = %user,
1917                "the none method is refused; publickey is the only way in"
1918            );
1919            return Ok(Auth::reject());
1920        }
1921
1922        let anonymous = Principal::anonymous();
1923        span.record("principal", tracing::field::display(&anonymous.id));
1924        self.shared.metrics.add_connections_anonymous(1);
1925        tracing::info!(
1926            target: TARGET_SSH,
1927            event = "ssh.auth.anonymous",
1928            // Named so nobody later reads this line as "a key was accepted".
1929            // It is on `gunnar::ssh` at `info` because "who is reading my
1930            // public repositories" must be answerable from a production log.
1931            authenticated = false,
1932            "anonymous session accepted: no key, public repositories only"
1933        );
1934        self.principal = Some(anonymous);
1935        Ok(Auth::Accept)
1936    }
1937
1938    async fn auth_publickey(
1939        &mut self,
1940        user: &str,
1941        public_key: &PublicKey,
1942    ) -> Result<Auth, Self::Error> {
1943        let span = self.span.clone();
1944        let _entered = span.enter();
1945        // Over the key's WIRE BLOB, not over its type: russh's `ssh-key` is
1946        // 0.7.0-rc.10 and `gunnar-core`'s is 0.6.7, so this is the only value
1947        // that can reach the identity table. It is byte-for-byte the
1948        // string `ssh-keygen -lf` prints for the same key.
1949        //
1950        // `None` means the key would not re-encode. That used to be an
1951        // `expect()`, a panic in the auth handler, over client-supplied bytes.
1952        // It is a refusal now; `fingerprint_of_key` has already
1953        // logged why.
1954        let Some(fingerprint) = fingerprint_of_key(public_key) else {
1955            let refusal = refuse_unencodable_key(&self.shared.metrics, user);
1956            if self.refused_auth() {
1957                return Err(self.too_many_auth_attempts());
1958            }
1959            return Ok(refusal);
1960        };
1961        let started = std::time::Instant::now();
1962        // `peer` is a field of the enclosing span from here down, so it is not
1963        // repeated on any line below (one implementation, and the whole point of the span).
1964        match self
1965            .shared
1966            .dispatch
1967            .control()
1968            .principal_for_key(&fingerprint)
1969        {
1970            Ok(Some(principal)) => {
1971                span.record("principal", tracing::field::display(&principal.id));
1972                self.shared.metrics.add_connections_accepted(1);
1973                tracing::info!(
1974                    target: TARGET_SSH,
1975                    event = "ssh.auth.accepted",
1976                    %fingerprint,
1977                    // Logged so an operator can see what was sent, and named so
1978                    // nobody later mistakes it for an input to the decision.
1979                    decorative_ssh_user = %user,
1980                    elapsed_ms = started.elapsed().as_millis() as u64,
1981                    "publickey accepted"
1982                );
1983                self.principal = Some(principal);
1984                // Kept, not recomputed and not dropped. This is the
1985                // value the identity table was probed with one line above, so
1986                // it is the key that authenticated, by construction.
1987                self.key_fingerprint = Some(fingerprint);
1988                Ok(Auth::Accept)
1989            }
1990            Ok(None) => {
1991                self.shared.metrics.add_connections_refused(1);
1992                tracing::warn!(
1993                    target: TARGET_SSH,
1994                    event = "ssh.auth.refused",
1995                    reason = "unknown-key",
1996                    %fingerprint,
1997                    decorative_ssh_user = %user,
1998                    elapsed_ms = started.elapsed().as_millis() as u64,
1999                    "publickey refused: no principal holds this key"
2000                );
2001                if self.refused_auth() {
2002                    return Err(self.too_many_auth_attempts());
2003                }
2004                Ok(Auth::reject())
2005            }
2006            // The key is still refused, since failing open is not on the table, but
2007            // this is a FAULT, not "unknown key", and the log has to say so.
2008            // An identity table that has become unreadable looks exactly like
2009            // an empty one from the client's side (the ACL model).
2010            Err(err) => {
2011                self.shared.metrics.add_connections_refused(1);
2012                self.shared.metrics.add_auth_faults(1);
2013                tracing::error!(
2014                    target: TARGET_SSH,
2015                    event = "ssh.auth.refused",
2016                    reason = "identity-table-unreadable",
2017                    %fingerprint,
2018                    decorative_ssh_user = %user,
2019                    elapsed_ms = started.elapsed().as_millis() as u64,
2020                    %err,
2021                    "publickey refused: the identity table could not be read"
2022                );
2023                if self.refused_auth() {
2024                    return Err(self.too_many_auth_attempts());
2025                }
2026                Ok(Auth::reject())
2027            }
2028        }
2029    }
2030
2031    async fn channel_open_session(
2032        &mut self,
2033        channel: Channel<Msg>,
2034        // russh 0.63 (2026-10-03): the open is answered through this handle, not
2035        // a returned bool; dropping it unanswered would refuse the channel.
2036        reply: russh::server::ChannelOpenHandle,
2037        session: &mut Session,
2038    ) -> Result<(), Self::Error> {
2039        // The exec request arrives later, carrying only a `ChannelId`, and the
2040        // byte stream can only be taken from the `Channel` itself, so it is
2041        // kept here. russh's own sftp example does the same.
2042        self.channels.insert(
2043            channel.id(),
2044            OpenChannel {
2045                channel,
2046                git_protocol: None,
2047            },
2048        );
2049
2050        // A session that authenticates, opens a channel and then does nothing is
2051        // not idle, because it answers keepalives, so the inactivity timeout never
2052        // fires and the connection would be held indefinitely. Arm a watchdog.
2053        if let Some(grace) = self.shared.exec_grace {
2054            let handle = session.handle();
2055            let exec_seen = self.exec_seen.clone();
2056            // `.instrument`, not a captured `peer`: a spawned task does NOT
2057            // inherit the thread-local span, so the watchdog's line would
2058            // otherwise be the one line in the whole connection that could not
2059            // be tied back to it.
2060            let span = self.span.clone();
2061            tokio::spawn(
2062                async move {
2063                    tokio::time::sleep(grace).await;
2064                    if !exec_seen.load(Ordering::SeqCst) {
2065                        tracing::info!(
2066                            target: TARGET_SSH,
2067                            event = "ssh.watchdog.disconnect",
2068                            ?grace,
2069                            "disconnecting: authenticated but never asked for a git command"
2070                        );
2071                        note_teardown(
2072                            "watchdog-disconnect",
2073                            handle
2074                                .disconnect(
2075                                    Disconnect::ByApplication,
2076                                    "no git command requested".to_string(),
2077                                    String::new(),
2078                                )
2079                                .await,
2080                        );
2081                    }
2082                }
2083                .instrument(span),
2084            );
2085        }
2086        reply.accept().await;
2087        Ok(())
2088    }
2089
2090    async fn exec_request(
2091        &mut self,
2092        channel: ChannelId,
2093        data: &[u8],
2094        session: &mut Session,
2095    ) -> Result<(), Self::Error> {
2096        self.exec_seen.store(true, Ordering::SeqCst);
2097        let span = self.span.clone();
2098        let _entered = span.enter();
2099
2100        // **The credential is re-read here, not remembered.** See
2101        // [`Connection::live_principal`]: russh will not dispatch an exec
2102        // before authentication succeeds, so an unauthenticated session is
2103        // unreachable rather than checked — but a session whose key was
2104        // REVOKED while it was open is entirely reachable, and until
2105        // 2026-09-11 it was authorised out of a cache filled at auth time.
2106        let principal = match self.live_principal() {
2107            Ok(principal) => principal,
2108            Err(stale) => {
2109                self.shared.metrics.add_exec_refused(1);
2110                let (reason, detail) = match &stale {
2111                    Stale::Revoked => ("key-revoked", String::new()),
2112                    Stale::Unreadable(err) => ("identity-table-unreadable", format!("{err}")),
2113                };
2114                tracing::warn!(
2115                    target: TARGET_SSH,
2116                    event = "ssh.exec.denied",
2117                    reason,
2118                    fingerprint = self.key_fingerprint.as_ref().map(tracing::field::display),
2119                    detail = %detail,
2120                    "exec denied: the key that opened this session no longer authorises it"
2121                );
2122                // Answered in git's own protocol, like every other denial on
2123                // this path: an `ERR ` pkt-line is what `git` prints verbatim
2124                // after `remote: `. A bare `channel_failure` here would put the
2125                // client back on "Could not read from remote repository", which
2126                // is the failure `report_error` was found unused for.
2127                session.channel_success(channel)?;
2128                session.data(channel, err_packet(REVOKED_MESSAGE)?)?;
2129                session.exit_status_request(channel, EXIT_DENIED)?;
2130                session.close(channel)?;
2131                // And the session goes with it. Refusing the one exec would
2132                // leave the connection open for the next one, which is the
2133                // finding restated a request later: a revoked key must not keep
2134                // a channel it can keep asking on.
2135                note_teardown(
2136                    "revoked-key-disconnect",
2137                    session
2138                        .handle()
2139                        .disconnect(
2140                            Disconnect::ByApplication,
2141                            "the key that opened this session no longer authorises it".to_string(),
2142                            String::new(),
2143                        )
2144                        .await,
2145                );
2146                return Ok(());
2147            }
2148        };
2149
2150        // ── the operator's fire drill: ONE ERROR on purpose ─────────────
2151        //
2152        // What the estate's watch raises to prove its alert path end to end (estate-box
2153        // 8A). Held to server-wide Admin before it says anything; an anonymous session
2154        // is refused the channel like the tunnel verb below.
2155        if std::str::from_utf8(data).map(str::trim) == Ok(FIRE_DRILL_VERB) {
2156            if self.key_fingerprint.is_none() {
2157                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {FIRE_DRILL_VERB} needs a key");
2158                session.channel_failure(channel)?;
2159                return Ok(());
2160            }
2161            let (exit, out) = match self.shared.dispatch.control().auth().may_server(&principal.id, gunnar_core::Level::Admin) {
2162                Ok(true) => {
2163                    tracing::error!(
2164                        target: TARGET_SSH,
2165                        event = "ssh.fire_drill",
2166                        principal = %principal.id,
2167                        "fire drill: {} asked for this ERROR to prove the alert path end to end — nothing is wrong",
2168                        principal.id
2169                    );
2170                    (0, "fire drill: one ERROR said on this server's log (nothing is wrong)\n".to_string())
2171                }
2172                Ok(false) => {
2173                    tracing::warn!(target: TARGET_SSH, event = "ssh.exec.denied", reason = "not-server-admin", principal = %principal.id, "{FIRE_DRILL_VERB} refused");
2174                    (EXIT_DENIED, format!("error: {FIRE_DRILL_VERB} needs server-wide Admin, which {} does not hold\n", principal.id))
2175                }
2176                Err(err) => (EXIT_DENIED, format!("error: the identity table could not be read: {err}\n")),
2177            };
2178            session.channel_success(channel)?;
2179            session.data(channel, out.into_bytes())?;
2180            session.exit_status_request(channel, exit)?;
2181            session.close(channel)?;
2182            return Ok(());
2183        }
2184
2185        // ── who this key is (wiring tracer B 2026-10-08) ──
2186        if std::str::from_utf8(data).map(str::trim) == Ok(WHOAMI_VERB) {
2187            if self.key_fingerprint.is_none() {
2188                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {WHOAMI_VERB} needs a key");
2189                session.channel_failure(channel)?;
2190                return Ok(());
2191            }
2192            session.channel_success(channel)?;
2193            session.data(channel, format!("{}\n", principal.id).into_bytes())?;
2194            session.exit_status_request(channel, 0)?;
2195            session.close(channel)?;
2196            return Ok(());
2197        }
2198
2199        // ── a repository's settings (owner, 2026-10-14) ──
2200        //
2201        // Decided by the control plane's own `Repos` handlers through
2202        // [`RepoDoor`]: this listener only proves who asks (the key, re-read
2203        // above) and carries the answer. An anonymous session is refused the
2204        // channel, before the pause and the plan gates (not a transfer).
2205        if let Some(parsed) = parse_repo_verb(data) {
2206            let Some(fingerprint) = self.key_fingerprint.as_ref().map(|f| f.to_string()) else {
2207                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {REPO_VERB} needs a key");
2208                session.channel_failure(channel)?;
2209                return Ok(());
2210            };
2211            let (exit, out) = match parsed {
2212                Err(why) => (1, format!("error: {why}\n")),
2213                Ok((repo, action)) => match self.shared.repo_door.clone() {
2214                    None => (
2215                        1,
2216                        "error: this server answers repository settings only on its control plane\n".to_string(),
2217                    ),
2218                    Some(door) => {
2219                        let ask = RepoAsk { principal: principal.id.clone(), fingerprint, repo: repo.clone(), action: action.clone() };
2220                        let answer = door(ask).await;
2221                        tracing::info!(
2222                            target: TARGET_SSH,
2223                            event = "ssh.control",
2224                            verb = REPO_VERB,
2225                            principal = %principal.id,
2226                            repo = %repo,
2227                            action = ?action,
2228                            decision = if answer.is_ok() { "allow" } else { "deny" },
2229                            "repository settings asked over the forge's door"
2230                        );
2231                        match answer {
2232                            Ok(said) => (0, if said.ends_with('\n') { said } else { format!("{said}\n") }),
2233                            Err(why) => (EXIT_DENIED, format!("error: {why}\n")),
2234                        }
2235                    }
2236                },
2237            };
2238            session.channel_success(channel)?;
2239            session.data(channel, out.into_bytes())?;
2240            session.exit_status_request(channel, exit)?;
2241            session.close(channel)?;
2242            return Ok(());
2243        }
2244
2245        // ── a member's keys for an encrypted repository's ring (2026-10-06) ──
2246        //
2247        // A read of the identity table, held to the same rule as
2248        // `Repos.RecipientKeys` (see [`recipient_keys_answer`]). Before the
2249        // pause and the plan gates for the tunnel verb's reason: not a
2250        // transfer. An anonymous session is refused the channel.
2251        if let Some(parsed) = parse_recipient_keys_verb(data) {
2252            if self.key_fingerprint.is_none() {
2253                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {RECIPIENT_KEYS_VERB} needs a key");
2254                session.channel_failure(channel)?;
2255                return Ok(());
2256            }
2257            let (exit, out) = match parsed {
2258                Err(why) => (1, format!("error: {why}\n")),
2259                Ok((repo, member)) => {
2260                    let (exit, out) = recipient_keys_answer(self.shared.dispatch.control(), &principal.id, &repo, &member);
2261                    tracing::info!(
2262                        target: TARGET_SSH,
2263                        event = "ssh.control",
2264                        verb = RECIPIENT_KEYS_VERB,
2265                        principal = %principal.id,
2266                        repo = %repo,
2267                        member = %member,
2268                        decision = if exit == 0 { "allow" } else { "deny" },
2269                        "recipient keys asked over the forge's door"
2270                    );
2271                    (exit, out)
2272                }
2273            };
2274            session.channel_success(channel)?;
2275            session.data(channel, out.into_bytes())?;
2276            session.exit_status_request(channel, exit)?;
2277            session.close(channel)?;
2278            return Ok(());
2279        }
2280
2281        // ── the ONE control verb: the tunnel address (T17) ─────────────
2282        //
2283        // Before the pause and the plan gates: those are about transfers, and
2284        // this is not one. Never through `Dispatch::authorise`, which knows
2285        // only git verbs. An anonymous session gets the channel refused like a
2286        // shell would; everyone else is held to server-wide Admin BEFORE the
2287        // door runs.
2288        if let Some(parsed) = parse_tunnel_address_verb(data) {
2289            if self.key_fingerprint.is_none() {
2290                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {TUNNEL_ADDRESS_VERB} needs a key");
2291                session.channel_failure(channel)?;
2292                return Ok(());
2293            }
2294            let (exit, out) = match self
2295                .shared
2296                .dispatch
2297                .control()
2298                .auth()
2299                .may_server(&principal.id, gunnar_core::Level::Admin)
2300            {
2301                Ok(false) => {
2302                    tracing::warn!(target: TARGET_SSH, event = "ssh.exec.denied", reason = "not-server-admin", principal = %principal.id, "{TUNNEL_ADDRESS_VERB} refused");
2303                    (EXIT_DENIED, format!("error: {TUNNEL_ADDRESS_VERB} needs server-wide Admin, which {} does not hold\n", principal.id))
2304                }
2305                Err(err) => (EXIT_DENIED, format!("error: the identity table could not be read: {err}\n")),
2306                Ok(true) => match parsed {
2307                    Err(why) => (1, format!("error: {why}\n")),
2308                    Ok(address) => match &self.shared.tunnel_door {
2309                        None => (1, "error: this box has no WireGuard tunnel (built without wg-client)\n".to_string()),
2310                        Some(door) => match door(&address) {
2311                            Ok(answer) => {
2312                                tracing::warn!(target: TARGET_SSH, event = "ssh.control", verb = TUNNEL_ADDRESS_VERB, principal = %principal.id, address = %address, decision = "allow", "tunnel address door answered");
2313                                (0, if answer.ends_with('\n') { answer } else { format!("{answer}\n") })
2314                            }
2315                            Err(why) => {
2316                                tracing::warn!(target: TARGET_SSH, event = "ssh.control", verb = TUNNEL_ADDRESS_VERB, principal = %principal.id, address = %address, decision = "deny", reason = %why, "tunnel address door refused");
2317                                (1, format!("error: {why}\n"))
2318                            }
2319                        },
2320                    },
2321                },
2322            };
2323            session.channel_success(channel)?;
2324            session.data(channel, out.into_bytes())?;
2325            session.exit_status_request(channel, exit)?;
2326            session.close(channel)?;
2327            return Ok(());
2328        }
2329
2330        // ── the PAUSED door, before the plan gates ──────────────────────
2331        //
2332        // MEASURED under KVM (bench R1, 2026-09-15): with the gate paused for
2333        // a growth and `/data` already unmounted, a push knocked, the plan
2334        // gates ran FIRST, and the disk-floor check statvfs'd the read-only
2335        // guard tmpfs under the mountpoint — the client was told "/data has
2336        // 64.0 KiB free … no push until space is reclaimed (`gunnar control
2337        // repo delete --purge`)". The wrong sentence, with advice to delete
2338        // repositories, on a box that is growing. The pause is the box's own
2339        // state and says nothing about any repository, so it is answered
2340        // before the command is even resolved: the same words and the same
2341        // EX_TEMPFAIL for every principal and every path.
2342        if let Some(pause) = self.shared.admission.paused() {
2343            self.shared.admission.note_refused_paused();
2344            tracing::info!(
2345                target: TARGET_GIT,
2346                event = "git.transfer.refused",
2347                reason = pause.reason(),
2348                command = %String::from_utf8_lossy(data),
2349                "transfer REFUSED: the admission gate is paused"
2350            );
2351            session.channel_success(channel)?;
2352            session.data(channel, err_packet(pause.words())?)?;
2353            session.exit_status_request(channel, EXIT_BUSY)?;
2354            session.close(channel)?;
2355            return Ok(());
2356        }
2357
2358        let request =
2359            match self
2360                .shared
2361                .dispatch
2362                .authorise(&principal, self.key_fingerprint.as_ref(), data)
2363            {
2364                Ok(request) => request,
2365                Err(err) => {
2366                    self.shared.metrics.add_exec_refused(1);
2367                    // **THE counting point for a tenant's `auth_failures`**: a
2368                    // principal that authenticated and then held no grant on
2369                    // the repository it named.
2370                    //
2371                    // Only `Denied`, and deliberately only `Denied`. The other
2372                    // arms are a different sentence: `NoSuchRepo` hides a
2373                    // typo behind the same wire message on purpose and would
2374                    // charge a namespace that may not exist; `Standby` and
2375                    // `Plan` are the box and the plan refusing, not the
2376                    // credential; `Command` never named a repository at all.
2377                    // And the OTHER half of "authentication failed" — a key
2378                    // that resolves to nobody — happens before any exec line
2379                    // and belongs to no namespace, so it stays where it is, on
2380                    // the box's own `connections_refused`.
2381                    if let (DispatchError::Denied { repo, .. }, Some(flow)) =
2382                        (&err, self.shared.admission.flow())
2383                    {
2384                        flow.auth_failure(repo.namespace(), gunnar_core::now_ms());
2385                    }
2386                    let raw = String::from_utf8_lossy(data);
2387                    match err.client_message() {
2388                        // Not a git command at all: `bash`, `scp`, an empty
2389                        // string. Refuse the channel request itself; the client is
2390                        // not speaking a protocol that could read a reply.
2391                        // git-lfs asks SSH for its credentials (`git-lfs-authenticate`,
2392                        // `git-lfs-transfer`) before every transfer and goes to its
2393                        // https endpoint when refused: a client's known first step,
2394                        // said at INFO (RUN A 2026-10-08: seven a minute during one LFS
2395                        // push failed the log gate as WARN spam). Anything else stays WARN.
2396                        None if is_lfs_over_ssh(&raw) => {
2397                            tracing::info!(
2398                                target: TARGET_SSH,
2399                                event = "ssh.exec.refused",
2400                                reason = "lfs-over-ssh",
2401                                command = %raw,
2402                                "exec refused: git-lfs asks SSH first and uses its https endpoint"
2403                            );
2404                            session.channel_failure(channel)?;
2405                        }
2406                        None => {
2407                            tracing::warn!(
2408                                target: TARGET_SSH,
2409                                event = "ssh.exec.refused",
2410                                reason = "not-a-git-command",
2411                                command = %raw,
2412                                %err,
2413                                "exec refused"
2414                            );
2415                            session.channel_failure(channel)?;
2416                        }
2417                        // A git client asked for something it may not have. Accept
2418                        // the channel and answer in its own protocol, so it prints
2419                        // the reason instead of "Could not read from remote
2420                        // repository".
2421                        Some(message) => {
2422                            tracing::warn!(
2423                                target: TARGET_SSH,
2424                                event = "ssh.exec.denied",
2425                                // The distinction the *wire* deliberately hides:
2426                                // "no such repository" and "no permission" are one
2427                                // string to the client so a caller cannot enumerate
2428                                // private repositories. The log keeps it.
2429                                reason = match &err {
2430                                    DispatchError::NoSuchRepo(_) => "no-such-repository",
2431                                    DispatchError::Denied { .. } => "no-grant",
2432                                    DispatchError::Standby { .. } => "standby-read-only",
2433                                    DispatchError::Twin { .. } => "twin-serves-no-git",
2434                                    DispatchError::AnonymousElsewhere { .. } => {
2435                                        "anonymous-elsewhere"
2436                                    }
2437                                    DispatchError::Plan { .. } => "gate-refused",
2438                                    DispatchError::Store(_) => "control-plane-unavailable",
2439                                    DispatchError::Command(_) => "not-a-git-command",
2440                                },
2441                                command = %raw,
2442                                %err,
2443                                "exec denied"
2444                            );
2445                            session.channel_success(channel)?;
2446                            session.data(channel, err_packet(&message)?)?;
2447                            session.exit_status_request(channel, EXIT_DENIED)?;
2448                            session.close(channel)?;
2449                        }
2450                    }
2451                    return Ok(());
2452                }
2453            };
2454
2455        let Some(open) = self.channels.remove(&channel) else {
2456            // The CLIENT's misuse (a second exec on one channel, or a channel never
2457            // opened): refused, said at WARN once per connection — any caller that may
2458            // run one git command could otherwise raise the server's ERROR at will.
2459            if std::mem::replace(&mut self.misuse_said, true) {
2460                tracing::debug!(target: TARGET_SSH, ?channel, "exec request for a channel that is not open (said already on this connection)");
2461            } else {
2462                tracing::warn!(
2463                    target: TARGET_SSH,
2464                    event = "ssh.exec.misuse",
2465                    ?channel,
2466                    "exec request for a channel that is not open (a second exec on one channel, or one never opened) — refused; said once per connection"
2467                );
2468            }
2469            session.channel_failure(channel)?;
2470            return Ok(());
2471        };
2472
2473        // The repository is known from here on, so it goes on the CONNECTION
2474        // span: every line the transfer emits underneath inherits it, and the
2475        // "which repository was that connection touching" question is answered
2476        // even for a line that has nothing to do with git.
2477        span.record("repo", tracing::field::display(&request.repo));
2478
2479        // Attached *after* the decision above, never before it: an environment
2480        // variable selects the framing of the answer and nothing else.
2481        let request = request.with_git_protocol(open.git_protocol);
2482
2483        // ── the explicit ceiling ───────────────────────────────────────────
2484        //
2485        // After authorisation, deliberately. "The server is busy" is a fact
2486        // about the server, and telling it to a client that may not touch this
2487        // repository at all would leak that the repository exists. The order
2488        // also means the gate meters *transfers gunnar was going to serve*,
2489        // which is what the counter is supposed to mean.
2490        //
2491        // It refuses rather than queues; `crate::admission` sets out why at
2492        // length. In one line: a git client retries, and a fast refusal with a
2493        // sentence in it beats a socket that goes quiet while still holding the
2494        // session, the channel and a slot.
2495        // The class the governor meters this operation as. `git-upload-pack` is
2496        // clone, fetch AND ls-remote on the wire and cannot be told apart before
2497        // the stream is open, which is *after* this decision by design; see
2498        // `OpClass::for_ssh_service` for what that costs and why the middle
2499        // class is the right guess.
2500        // ── the BOOT WINDOW door, before admission ─────────────────────────
2501        //
2502        // After the repository is known and authorised, so a stranger cannot
2503        // learn from this whether a repository exists, and before a slot is
2504        // taken, so a boot does not spend its gate on requests it cannot serve.
2505        // See `GunnarSsh::with_readiness` for what was measured.
2506        if !self.shared.ready.load(std::sync::atomic::Ordering::Relaxed) {
2507            tracing::info!(
2508                target: TARGET_GIT,
2509                event = "git.transfer.refused",
2510                reason = "starting",
2511                service = %request.service,
2512                "transfer REFUSED: the stores are still opening after a restart"
2513            );
2514            session.channel_success(channel)?;
2515            session.data(channel, err_packet(STARTING_MESSAGE)?)?;
2516            session.exit_status_request(channel, EXIT_BUSY)?;
2517            session.close(channel)?;
2518            return Ok(());
2519        }
2520
2521        let class = OpClass::for_ssh_service(request.service);
2522        // `admit_within`, not `try_admit`: over the limit the transfer parks for
2523        // a bounded moment instead of being refused on the spot. A refused
2524        // client reconnects, and a reconnect is a whole SSH handshake — more
2525        // than the wait costs. `crate::governor::DEFAULT_ADMISSION_WAIT` has the
2526        // measurement; `--admission-wait-ms 0` is the old behaviour exactly.
2527        let admitted = match self
2528            .shared
2529            .admission
2530            .admit_within(&request.principal.id, &request.repo.namespace, class)
2531            .await
2532        {
2533            Ok((guard, waited)) => {
2534                // AGGREGATE, at most once a second. A row per waiting client is
2535                // the appliance's whole twenty-row console under load (the
2536                // measured 5411 in one 256-client run); a row per second
2537                // carrying the totals is the same fact, readable.
2538                if !waited.is_zero() && self.shared.admission.wait_row_due() {
2539                    let stats = self.shared.admission.stats();
2540                    tracing::info!(
2541                        target: TARGET_GIT,
2542                        event = "git.transfer.waited",
2543                        class = class.name(),
2544                        waited_ms = waited.as_millis() as u64,
2545                        waited_total = stats.admitted_after_wait,
2546                        mean_wait_ms = stats.mean_wait_ms(),
2547                        peak_wait_ms = stats.peak_wait_ms,
2548                        in_flight = stats.in_flight,
2549                        limit = stats.limit,
2550                        "transfers are WAITING for a slot and are then served: {} so far, mean \
2551                         {} ms, worst {} ms against a bound of {} ms",
2552                        stats.admitted_after_wait,
2553                        stats.mean_wait_ms(),
2554                        stats.peak_wait_ms,
2555                        self.shared.admission.wait().as_millis() as u64
2556                    );
2557                }
2558                guard
2559            }
2560            Err(refusal) => {
2561                let stats = self.shared.admission.stats();
2562                tracing::warn!(
2563                    target: TARGET_GIT,
2564                    event = "git.transfer.refused",
2565                    // Which gate, not just "a gate": the anonymous ceiling
2566                    // fires with the box nearly empty, and a line that
2567                    // still said `admission-gate-full` there would send an
2568                    // operator to raise a limit that was never reached.
2569                    reason = refusal.reason(),
2570                    service = %request.service,
2571                    class = class.name(),
2572                    %refusal,
2573                    in_flight = stats.in_flight,
2574                    peak_in_flight = stats.peak_in_flight,
2575                    // The governed limit, so a log line says what was actually
2576                    // in force rather than what was configured. Without it the
2577                    // shed is invisible again.
2578                    limit = stats.limit,
2579                    reserved_bytes = stats.reserved_bytes,
2580                    refused_total = stats.refused(),
2581                    // The bound the transfer waited out before this. Without it
2582                    // a reader cannot tell an instant refusal from one the
2583                    // server spent two seconds trying to avoid.
2584                    waited_ms = self.shared.admission.wait().as_millis() as u64,
2585                    peak_wait_ms = stats.peak_wait_ms,
2586                    "transfer REFUSED by the admission gate"
2587                );
2588                // Answered in the client's own protocol, like a denial, so git
2589                // prints the reason instead of "Could not read from remote
2590                // repository". Saturation the operator can see and the user
2591                // cannot is only half a fix.
2592                session.channel_success(channel)?;
2593                session.data(channel, err_packet(&refusal.client_message())?)?;
2594                session.exit_status_request(channel, EXIT_BUSY)?;
2595                session.close(channel)?;
2596                return Ok(());
2597            }
2598        };
2599
2600        // ── the span the whole transfer runs under ─────────────────────────
2601        //
2602        // A CHILD of `ssh.connection`, so it inherits `conn`, `peer`,
2603        // `principal` and `repo` and states none of them again. `service` is
2604        // the one thing that is new. Everything `gunnar-cli`'s transport and
2605        // `gunnar-policy`'s gate emit lands underneath it, in the same
2606        // process, without either of them being handed an id to repeat.
2607        let transfer = tracing::info_span!(
2608            target: TARGET_GIT,
2609            "git.transfer",
2610            service = %request.service,
2611            access = %request.access,
2612        );
2613        transfer.in_scope(|| {
2614            tracing::info!(
2615                target: TARGET_GIT,
2616                event = "git.transfer.started",
2617                git_protocol = ?request.git_protocol,
2618                "serving"
2619            )
2620        });
2621        let in_flight = self.shared.metrics.transfer_started();
2622        let metrics = Arc::clone(&self.shared.metrics);
2623        let started = std::time::Instant::now();
2624
2625        session.channel_success(channel)?;
2626
2627        let transport = self.shared.transport.clone();
2628        // Split, NOT `into_stream()`. `into_stream` wraps the channel in
2629        // russh's `ChannelCloseOnDrop`, which spawns a `Close` the instant the
2630        // transport lets the stream go, and that `Close` raced (and beat) the
2631        // `exit-status` sent next, which is how a successful push came to
2632        // report failure. Neither half closes the channel on drop, so the two
2633        // messages below are ordered by the sender's own queue.
2634        let (read_half, write_half) = open.channel.split();
2635        // METERED, so the memory ledger learns this transfer's real size as it
2636        // happens rather than being charged a flat per-class constant at
2637        // admission. `crate::admission::Metered` is the only place a byte count
2638        // reaches the governor, and it wraps the duplex the transport was
2639        // already going to be handed — no second copy, no second truth.
2640        let admitted = Arc::new(admitted);
2641        let stream = crate::admission::Metered::new(
2642            tokio::io::join(
2643                ChannelReader::new(read_half),
2644                // Bound to the session: a write parked on the SSH window ends
2645                // with an error when the session does (see `SessionBound`).
2646                SessionBound::new(write_half.make_writer(), self.session_over.subscribe()),
2647            ),
2648            Arc::clone(&admitted),
2649        );
2650        // Spawned, not awaited: see the module docs. Awaiting here would stop
2651        // the session loop that has to feed this very stream.
2652        // ── THE STANDING FAN, not a fresh thread ────────────────────────
2653        //
2654        // This used to be `tokio::spawn` alone, and the task itself was never
2655        // the cost — tasks are cheap. The cost is behind it: `SyncIoBridge`
2656        // hands the transfer to tokio's BLOCKING pool, which spawns an OS
2657        // thread on demand, per operation, up to 512 of them. That is
2658        // per-request initialisation, and it is what the fans replace.
2659        //
2660        // The future is driven by `Handle::block_on` on a worker that was
2661        // already alive and parked before this client connected. A full queue
2662        // refuses here, which is the only place a refusal is free.
2663        let handle = tokio::runtime::Handle::current();
2664        let pipelines = crate::pipeline::fans();
2665        let over = self.session_over.subscribe();
2666        let (calm_grace, crowded_grace) = self.shared.session_end_grace;
2667        let body = async move {
2668            // ── the CoDel dequeue ──────────────────────────────────────
2669            //
2670            // The FIRST line of the task, deliberately. Everything between
2671            // `try_admit` above and this point is the transfer WAITING:
2672            // tokio's ready queue, and behind it the blocking-pool handoff
2673            // that `SyncIoBridge` needs before a single byte moves. That
2674            // wait is the queue the record called invisible, and this call is
2675            // what makes the governor able to see it. Measured here rather
2676            // than at the end because sojourn is wait time; service time
2677            // filed as sojourn would make every large clone look like a
2678            // standing queue.
2679            admitted.entered_service();
2680            // ★ The WHOLE transfer is bound to the session, not only its writes
2681            // (`SessionBound`): a client that goes silent while main is still
2682            // building the pack has no write parked on the window to wake, and
2683            // the build can take minutes (the UpCloud mockup's LD1/LD2,
2684            // 2026-10-09: 2 min 10 s on TCG). When the session ends — the
2685            // keepalives unanswered, a hang-up — the transfer ends with it and
2686            // gives back its fan worker, admission slot and store handle.
2687            let cancel = request.cancel.clone();
2688            let served = serve_until_session_ends(transport.serve(request, Box::new(stream)), over.clone(), &cancel, || {
2689                // Stinky row 19 (2026-10-10): a hung-up client's worker is one
2690                // a queued client waits for, so the grace is short when any do.
2691                crate::channel::grace_for(calm_grace, crowded_grace, pipelines.queued(class))
2692            }, || {
2693                crate::seam::TransportError::Io(session_ended_error())
2694            })
2695            .await;
2696            let status = match served {
2697                Ok(()) => 0,
2698                Err(err) => {
2699                    metrics.add_transfers_failed(1);
2700                    // ── the refusal the client used to never hear ───────
2701                    //
2702                    // Until 2026-08-05 this arm logged the whole reason and
2703                    // sent the client an exit status and nothing else, so
2704                    // every post-authorisation refusal — the repository's
2705                    // store lock held by a `gunnar gc`, a storage fault on
2706                    // open, an unwired service — reached the user as
2707                    // `fatal: the remote end hung up unexpectedly`. The two
2708                    // gates ABOVE this one, in this same function, both
2709                    // answer in the client's own protocol; this one did not,
2710                    // and the asymmetry is what made it invisible.
2711                    //
2712                    // `client_message()` is `None` whenever bytes may
2713                    // already have gone out, so this cannot append a
2714                    // pkt-line to a half-written pack — see
2715                    // `TransportError::client_message`.
2716                    let told = err.client_message();
2717                    tracing::warn!(
2718                        target: TARGET_GIT,
2719                        event = "git.transfer.failed",
2720                        %err,
2721                        // Whether the user was given a reason, said in the
2722                        // log, because "the client was told" and "the client
2723                        // saw a dead socket" must not be the same line.
2724                        client_told = told.is_some(),
2725                        "transport ended with an error"
2726                    );
2727                    if let Some(message) = told {
2728                        match err_packet(&message) {
2729                            Ok(packet) => {
2730                                // Raced against the session's end: `data_bytes`
2731                                // waits for SSH window like the stream does.
2732                                let sent = tokio::select! {
2733                                    sent = write_half.data_bytes(packet) => sent,
2734                                    () = session_over(over.clone()) => Err(russh::Error::IO(session_ended_error())),
2735                                };
2736                                note_teardown("refusal-packet", sent)
2737                            }
2738                            Err(err) => tracing::error!(
2739                                target: TARGET_GIT,
2740                                %err,
2741                                "the refusal could not be encoded, so the client is \
2742                                 getting only an exit status"
2743                            ),
2744                        }
2745                    }
2746                    EXIT_TRANSPORT_FAILED
2747                }
2748            };
2749            // The channel is still open here, nothing has been dropped that
2750            // could have closed it, so the status arrives before the close,
2751            // through the same sender the data went through.
2752            if write_half.exit_status(status).await.is_err() {
2753                // WARN only when the client missed a verdict it needed. A
2754                // client that took everything it asked for and hung up
2755                // before the `exit-status` has lost nothing — it is how
2756                // `ls-remote` and `fetch` end — and warning about it made
2757                // `grep -i error` useless on a completely healthy server:
2758                // MEASURED 2026-08-06, 9 successful transfers, 9 of these.
2759                // A client that missed a NON-ZERO status did lose
2760                // something: it believes the transfer worked.
2761                if status == 0 {
2762                    tracing::debug!(
2763                        target: TARGET_GIT,
2764                        event = "git.exit_status.undelivered",
2765                        ?channel,
2766                        status,
2767                        "the client hung up before the exit status; it had already \
2768                         been served"
2769                    );
2770                } else {
2771                    tracing::warn!(
2772                        target: TARGET_GIT,
2773                        event = "git.exit_status.undelivered",
2774                        ?channel,
2775                        status,
2776                        "the client was gone before a NON-ZERO exit status, so it \
2777                         does not know this transfer failed"
2778                    );
2779                }
2780            }
2781            note_teardown("channel-close", write_half.close().await);
2782            // The duration an operator asked "why was that clone slow"
2783            // about: the WHOLE transfer, from admission to the last byte
2784            // acknowledged, not the state machine's own view of itself.
2785            tracing::info!(
2786                target: TARGET_GIT,
2787                event = "git.transfer.finished",
2788                status,
2789                elapsed_ms = started.elapsed().as_millis() as u64,
2790                "transfer finished"
2791            );
2792            // Moved into this task and dropped here, so the slot is held for the
2793            // whole transfer and released by `Drop` rather than by a call that
2794            // an early return could skip. The in-flight gauge is released the
2795            // same way, and for the same reason.
2796            drop(admitted);
2797            drop(in_flight);
2798        }
2799        .instrument(transfer);
2800
2801        // ── WHAT THE FAN IS ACTUALLY CARRYING ───────────────────────────────
2802        //
2803        // `Pipelines` has counted `served`, `queued` and `workers` since it was
2804        // written and NOTHING reported them, so the only evidence the fans were
2805        // doing anything was the absence of refusals — which is not evidence,
2806        // it is silence. Asked "is the queue filling", the honest answer was an
2807        // inference from the workload's shape.
2808        //
2809        // `submit` already returns the queue depth after the push, so this
2810        // costs one integer. Emitted only when an operation ACTUALLY QUEUED
2811        // (depth > 1, i.e. it did not go straight to a free worker), because a
2812        // line per transfer on a 102-client arm is noise that hides the signal.
2813        // `served` and `workers` ride along so one line answers "how deep, out
2814        // of how many, having done how much".
2815        match pipelines.submit(
2816            class,
2817            Box::new(move || {
2818                handle.block_on(body);
2819            }),
2820        ) {
2821            Ok(depth) if depth > 1 => tracing::info!(
2822                target: TARGET_GIT,
2823                event = "git.pipeline.queued",
2824                class = ?class,
2825                depth,
2826                workers = pipelines.workers(class),
2827                served = pipelines.served(class),
2828                "an operation waited for a worker"
2829            ),
2830            Ok(_) => {}
2831            Err(refused) => {
2832                // The fan is saturated. Say so in the client's own protocol
2833                // rather than dropping the channel — a dead socket is how a
2834                // refusal becomes "the remote end hung up unexpectedly".
2835                tracing::warn!(
2836                    target: TARGET_GIT,
2837                    event = "git.transfer.refused",
2838                    class = ?class,
2839                    depth = pipelines.queued(class),
2840                    workers = pipelines.workers(class),
2841                    served = pipelines.served(class),
2842                    %refused,
2843                    "the standing fan for this class refused the operation"
2844                );
2845                return Err(anyhow::anyhow!("{refused}"));
2846            }
2847        }
2848        Ok(())
2849    }
2850
2851    // ── everything below is the "no shell, ever" half ──────────────────────
2852    //
2853    // russh's defaults for these return `Ok(())` **without** answering the
2854    // channel request, which leaves a client waiting rather than failing. Each
2855    // one is answered explicitly.
2856
2857    async fn shell_request(
2858        &mut self,
2859        channel: ChannelId,
2860        session: &mut Session,
2861    ) -> Result<(), Self::Error> {
2862        self.refuse(channel, session, "shell").await
2863    }
2864
2865    async fn subsystem_request(
2866        &mut self,
2867        channel: ChannelId,
2868        name: &str,
2869        session: &mut Session,
2870    ) -> Result<(), Self::Error> {
2871        // Includes `sftp`. gunnar serves git and nothing else.
2872        self.span.in_scope(|| {
2873            tracing::warn!(
2874                target: TARGET_SSH,
2875                event = "ssh.request.refused",
2876                subsystem = %name,
2877                "subsystem refused"
2878            )
2879        });
2880        self.refuse(channel, session, "subsystem").await
2881    }
2882
2883    async fn pty_request(
2884        &mut self,
2885        channel: ChannelId,
2886        _term: &str,
2887        _col_width: u32,
2888        _row_height: u32,
2889        _pix_width: u32,
2890        _pix_height: u32,
2891        _modes: &[(russh::Pty, u32)],
2892        session: &mut Session,
2893    ) -> Result<(), Self::Error> {
2894        self.refuse(channel, session, "pty").await
2895    }
2896
2897    async fn x11_request(
2898        &mut self,
2899        channel: ChannelId,
2900        _single_connection: bool,
2901        _x11_auth_protocol: &str,
2902        _x11_auth_cookie: &str,
2903        _x11_screen_number: u32,
2904        session: &mut Session,
2905    ) -> Result<(), Self::Error> {
2906        self.refuse(channel, session, "x11").await
2907    }
2908
2909    async fn env_request(
2910        &mut self,
2911        channel: ChannelId,
2912        variable_name: &str,
2913        variable_value: &str,
2914        session: &mut Session,
2915    ) -> Result<(), Self::Error> {
2916        // An ALLOWLIST, not a filter: everything that is not named here is
2917        // refused, so adding a variable is a deliberate edit to one array and
2918        // the default for anything new is "no". See the module docs for why an
2919        // SSH server that passes an environment through is an attack surface.
2920        let span = self.span.clone();
2921        let _entered = span.enter();
2922        if !ENV_ALLOWLIST.contains(&variable_name) {
2923            // Same refusal, same event, same fields — two levels. See
2924            // [`ENV_ROUTINE`] for the 406-of-406 measurement that split them.
2925            if env_refusal_is_routine(variable_name) {
2926                tracing::debug!(
2927                    target: TARGET_SSH,
2928                    event = "ssh.env.refused",
2929                    reason = "not-allowlisted",
2930                    routine = true,
2931                    variable = %variable_name,
2932                    "env refused: gunnar reads exactly one environment variable"
2933                );
2934            } else {
2935                tracing::warn!(
2936                    target: TARGET_SSH,
2937                    event = "ssh.env.refused",
2938                    reason = "not-allowlisted",
2939                    routine = false,
2940                    variable = %variable_name,
2941                    "env refused: gunnar reads exactly one environment variable"
2942                );
2943            }
2944            session.channel_failure(channel)?;
2945            return Ok(());
2946        }
2947
2948        // A value gunnar will not read is not worth holding. `version=2` is
2949        // nine bytes; the cap is generous and exists so a client cannot make
2950        // the server keep an unbounded string per channel.
2951        if variable_value.len() > MAX_ENV_VALUE {
2952            tracing::warn!(
2953                target: TARGET_SSH,
2954                event = "ssh.env.refused",
2955                reason = "value-too-long",
2956                variable = %variable_name,
2957                length = variable_value.len(),
2958                "env refused: the value is implausibly long"
2959            );
2960            session.channel_failure(channel)?;
2961            return Ok(());
2962        }
2963
2964        let Some(open) = self.channels.get_mut(&channel) else {
2965            tracing::warn!(
2966                target: TARGET_SSH,
2967                ?channel,
2968                "env request for a channel that was never opened"
2969            );
2970            session.channel_failure(channel)?;
2971            return Ok(());
2972        };
2973        if variable_name != ENV_GIT_PROTOCOL {
2974            // `ENV_ALLOWLIST` grew a name and nothing here was taught to carry
2975            // it. Fail closed and say so, rather than quietly filing an unknown
2976            // variable in the `GIT_PROTOCOL` slot.
2977            tracing::error!(
2978                target: TARGET_SSH,
2979                variable = %variable_name,
2980                "env allowlisted but not carried anywhere; refusing"
2981            );
2982            session.channel_failure(channel)?;
2983            return Ok(());
2984        }
2985        tracing::debug!(
2986            target: TARGET_SSH,
2987            variable = %variable_name,
2988            value = %variable_value,
2989            "env accepted"
2990        );
2991        open.git_protocol = Some(variable_value.to_string());
2992        session.channel_success(channel)?;
2993        Ok(())
2994    }
2995
2996    async fn channel_close(
2997        &mut self,
2998        channel: ChannelId,
2999        _session: &mut Session,
3000    ) -> Result<(), Self::Error> {
3001        self.channels.remove(&channel);
3002        Ok(())
3003    }
3004}
3005
3006impl Connection {
3007    async fn refuse(
3008        &mut self,
3009        channel: ChannelId,
3010        session: &mut Session,
3011        what: &'static str,
3012    ) -> Result<(), anyhow::Error> {
3013        self.span.in_scope(|| {
3014            tracing::warn!(
3015                target: TARGET_SSH,
3016                event = "ssh.request.refused",
3017                request = what,
3018                "refused: gunnar has no shell"
3019            )
3020        });
3021        session.channel_failure(channel)?;
3022        Ok(())
3023    }
3024}
3025
3026/// The session loop ended: say so in words an operator's `grep -i error` will
3027/// not find, unless it really was an error.
3028///
3029/// See [`peer_hung_up`]. `error=early eof` fired on **every healthy fetch**
3030/// before this existed.
3031fn note_session_end(error: &anyhow::Error) {
3032    if peer_hung_up(error.as_ref()) {
3033        tracing::debug!(
3034            target: TARGET_SSH,
3035            event = "ssh.session.closed",
3036            reason = "peer-hung-up",
3037            "the client closed the connection"
3038        );
3039        return;
3040    }
3041    tracing::debug!(target: TARGET_SSH, %error, "ssh session ended with an error");
3042}
3043
3044/// Note that a best-effort teardown step did not work.
3045///
3046/// # Best-effort stays best-effort. What changes is that it stops being silent
3047///
3048/// Closing a channel and disconnecting a watchdogged session are both allowed
3049/// to fail: the peer may already be gone, which is the ordinary case and not
3050/// worth an operator's attention. Neither ever became an error, and neither
3051/// should. But a **systemic** failure to tear sessions down, from descriptor
3052/// exhaustion, the failure mode the unbounded handle cache used to
3053/// produce, was previously indistinguishable from a quiet server, because
3054/// `let _ =` leaves nothing behind at any level.
3055///
3056/// `debug!` rather than `warn!`: one of these is normal and a thousand is not,
3057/// and the thing that tells them apart is a count in a journal, not a severity
3058/// on a line that fires every time a client hangs up first.
3059/// The one encoder for a client-visible refusal: `ERR <message>` as a pkt-line.
3060///
3061/// Three call sites now answer a refusal in the client's own protocol — the
3062/// dispatch denial, the admission gate and a transfer refused before its first
3063/// byte — and a fourth is a matter of time. One implementation: the framing is written once,
3064/// so a refusal cannot acquire a subtly different shape depending on which gate
3065/// produced it. `ERR` is git's own out-of-band error line, which every git
3066/// version since 1.6 prints verbatim instead of "Could not read from remote
3067/// repository".
3068fn err_packet(message: &str) -> anyhow::Result<Vec<u8>> {
3069    let mut packet = Vec::new();
3070    gunnar_pktline::write_text(&mut packet, format!("ERR {message}").as_bytes())
3071        .map_err(|e| anyhow::anyhow!("encoding ERR packet: {e}"))?;
3072    Ok(packet)
3073}
3074
3075///
3076/// # …and "the peer is already gone" is not a failure at all
3077///
3078/// Closing a channel whose peer has already hung up returns
3079/// `russh::Error::SendError`, and until 2026-08-06 that was logged as
3080/// `ssh.teardown.failed err=Channel send error` on **every healthy transfer**
3081/// (MEASURED that day: 9 successful transfers, 6 of these). It is the same
3082/// fact `handle_session_error` and the undelivered exit status already report,
3083/// counted a third time in the third set of words with "err" in it. It is
3084/// separated here by [`peer_hung_up`] so the systemic case this function exists
3085/// for — descriptor exhaustion, a wedged sender — is still the only thing that
3086/// says `failed`.
3087fn note_teardown<E: std::error::Error + 'static>(step: &'static str, result: Result<(), E>) {
3088    if let Err(err) = result {
3089        if peer_hung_up(&err) {
3090            tracing::debug!(
3091                target: TARGET_SSH,
3092                event = "ssh.teardown.peer_gone",
3093                step,
3094                "the peer had already gone, so this teardown step had nothing to say \
3095                 it to"
3096            );
3097            return;
3098        }
3099        tracing::debug!(
3100            target: TARGET_SSH,
3101            event = "ssh.teardown.failed",
3102            step,
3103            %err,
3104            "a session teardown step did not complete"
3105        );
3106    }
3107}
3108
3109/// **Is this failure just "the peer hung up first"?**
3110///
3111/// The one classifier for that question (one implementation). Three places used to answer it
3112/// independently, and all three answered *"no"*, which is why a completely
3113/// healthy `gunnar ls-remote` left this in the journal:
3114///
3115/// ```text
3116///   DEBUG ssh session ended with an error   error=early eof
3117///   WARN  the client was gone before the exit status  channel=ChannelId(2) status=0
3118///   DEBUG ssh.teardown.failed  step="channel-close"  err=Channel send error
3119///   INFO  transfer finished  status=0  elapsed_ms=106
3120/// ```
3121///
3122/// MEASURED on the reference desktop 2026-08-06 from the live server's own journal, at
3123/// `10:55:10`, for a transfer that succeeded. Nothing was damaged — `status=0`,
3124/// `transfer finished` — but three lines carrying `error`, `err` and `failed`
3125/// per healthy call means `grep -i error` finds only noise, and that is the
3126/// first command an operator runs.
3127///
3128/// # Classified on the cause, never on the message text
3129///
3130/// A `contains("early eof")` would be a guard that silently stops matching the
3131/// day tokio rewords its `read_exact`, and the failure would be invisible
3132/// because it fails in the *quiet* direction — back to noise, not to a crash.
3133/// So this walks the source chain and matches concrete things:
3134/// `io::ErrorKind`, and russh's own named variants. `early eof` is
3135/// `ErrorKind::UnexpectedEof` (tokio's `read_exact`) and `Channel send error`
3136/// is `russh::Error::SendError`; neither string appears below.
3137///
3138/// # What it deliberately does NOT swallow
3139///
3140/// Only the socket going away. A key exchange failure, a MAC failure, a
3141/// protocol desync, a decryption failure — every one of those is still an error
3142/// and still logged as one. `tests/teardown_noise.rs` asserts both directions,
3143/// because a classifier that answered `true` for everything would produce a
3144/// beautifully quiet journal for a server that was on fire.
3145fn peer_hung_up(err: &(dyn std::error::Error + 'static)) -> bool {
3146    use std::io::ErrorKind;
3147
3148    let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(err);
3149    while let Some(e) = cause {
3150        if let Some(io) = e.downcast_ref::<std::io::Error>() {
3151            if matches!(
3152                io.kind(),
3153                ErrorKind::UnexpectedEof
3154                    | ErrorKind::BrokenPipe
3155                    | ErrorKind::ConnectionReset
3156                    | ErrorKind::ConnectionAborted
3157                    | ErrorKind::NotConnected
3158            ) {
3159                return true;
3160            }
3161        }
3162        if let Some(r) = e.downcast_ref::<russh::Error>() {
3163            if matches!(
3164                r,
3165                // The channel's sender is gone: nothing left to send to.
3166                russh::Error::SendError
3167                    // russh's own names for a hung-up peer.
3168                    | russh::Error::HUP
3169                    | russh::Error::Disconnect
3170            ) {
3171                return true;
3172            }
3173        }
3174        cause = e.source();
3175    }
3176    false
3177}
3178
3179/// Refuse a key that would not re-encode.
3180///
3181/// A free function taking only the counters, so the decision this makes (a
3182/// **rejection**, and a counted one) can be asserted without standing up a
3183/// russh session. The alternative, an `expect()`, aborted the auth handler.
3184///
3185/// `auth_faults` is deliberately **not** moved: that counter means "the
3186/// identity table could not be read", and this key never reached it. The `warn`
3187/// [`crate::access::fingerprint_of_key`] emits carries
3188/// `reason = "key-does-not-re-encode"`, which is what tells the two apart.
3189fn refuse_unencodable_key(metrics: &gunnar_core::observe::ServerMetrics, user: &str) -> Auth {
3190    metrics.add_connections_refused(1);
3191    tracing::warn!(
3192        target: TARGET_SSH,
3193        event = "ssh.auth.refused",
3194        reason = "key-does-not-re-encode",
3195        decorative_ssh_user = %user,
3196        "publickey refused: the offered key did not re-encode"
3197    );
3198    Auth::reject()
3199}
3200
3201/// The one environment variable a git client may send: how it asks for
3202/// protocol v2 over SSH.
3203pub const ENV_GIT_PROTOCOL: &str = "GIT_PROTOCOL";
3204
3205/// **Every** environment variable this server accepts from a client.
3206///
3207/// An allowlist, checked by `env_request`. `LD_PRELOAD`, `PATH`, `GIT_DIR` and
3208/// everything else a client might try is refused with a channel failure, the
3209/// same answer this server gave to *all* environment requests before
3210/// [`ENV_GIT_PROTOCOL`] was needed. Nothing here is ever executed, exported to
3211/// a process or consulted by the grant check; gunnar has no shell and spawns
3212/// nothing, so the only reason an accepted name exists is that the *wire* reads
3213/// it.
3214pub const ENV_ALLOWLIST: [&str; 1] = [ENV_GIT_PROTOCOL];
3215
3216/// The refused names that mean **nothing at all**, because OpenSSH sends them
3217/// on its own without being asked.
3218///
3219/// Still refused — [`ENV_ALLOWLIST`] is unchanged and this list grants nothing.
3220/// What it changes is the *level* the refusal is reported at, and the reason it
3221/// has to exist is a measurement.
3222///
3223/// MEASURED on the reference desktop 2026-08-05, an 18-repository ingest over real SSH with stock
3224/// git 2.53.0 as the client, 203 connections, `RUST_LOG=info`:
3225///
3226/// | WARN lines | of which `ssh.env.refused` for `LANG`/`COLORTERM` |
3227/// |---|---|
3228/// | 406 | **406 (100%)** |
3229///
3230/// Zero warnings of any other kind were emitted in the whole run. OpenSSH's
3231/// shipped `ssh_config` carries `SendEnv LANG LC_*`, and desktop terminals add
3232/// `COLORTERM`, so every `git` over `ssh` produces two of these before it has
3233/// asked for anything. An operator who greps for `WARN` on a live gunnar
3234/// therefore reads noise at a 100% false-positive rate and will not see the
3235/// first real one — which is the same disease as a check that cannot go red,
3236/// wearing the other colour.
3237///
3238/// A client naming something NOT on this list is a different event and keeps
3239/// its `warn`: `LD_PRELOAD`, `PATH`, `GIT_DIR`, `GIT_ALTERNATE_OBJECT_DIRECTORIES`
3240/// are what an attacker tries, and nothing sends them by accident.
3241pub const ENV_ROUTINE: [&str; 4] = ["LANG", "LANGUAGE", "COLORTERM", "TERM"];
3242
3243/// Whether `name` is a routine refusal ([`ENV_ROUTINE`], or any `LC_*`, which
3244/// OpenSSH sends as a glob).
3245pub fn env_refusal_is_routine(name: &str) -> bool {
3246    ENV_ROUTINE.contains(&name) || name.starts_with("LC_")
3247}
3248
3249/// The longest environment value the server will hold.
3250///
3251/// `version=2` is nine bytes. This is a bound on what one channel can make the
3252/// server remember, not a syntax check; the grammar is `gunnar-wire`'s.
3253pub const MAX_ENV_VALUE: usize = 256;
3254
3255/// Exit status reported when access was refused. Matches what `git` itself
3256/// reports for a failed remote helper.
3257pub const EXIT_DENIED: u32 = 128;
3258
3259/// **What a client is told when the key that opened its session stopped
3260/// authorising it.**
3261///
3262/// One string, named here rather than spelled at the refusal site, so the test
3263/// that asserts what reaches the client reads the same constant the server
3264/// sends. It says what happened and what to do, because "denied" on a
3265/// connection that worked a second ago is otherwise indistinguishable from the
3266/// server having broken.
3267///
3268/// It does NOT distinguish a revoked key from an unreadable identity table, for
3269/// the reason the whole of this surface merges "no such repository" with "no
3270/// permission": the difference is operator information, and the log keeps it.
3271pub const REVOKED_MESSAGE: &str =
3272    "the key that opened this session no longer authorises it; reconnect";
3273
3274/// Exit status reported when the wire failed mid-transfer.
3275pub const EXIT_TRANSPORT_FAILED: u32 = 1;
3276
3277/// Exit status reported when the admission gate refused a transfer.
3278///
3279/// `EX_TEMPFAIL` from `sysexits.h`, and deliberately **not** [`EXIT_DENIED`]:
3280/// "try again" and "you may not" are different answers, and a retry script that
3281/// cannot tell them apart either hammers a permission error forever or gives up
3282/// on a transient one.
3283pub const EXIT_BUSY: u32 = 75;
3284
3285/// What a client is told while the server's stores are still opening.
3286///
3287/// **The word `starting` is load-bearing** and is asserted by
3288/// `tests/readiness.rs`: it is what separates this from
3289/// `gunnar_wire::Error::Store`'s `the repository is unavailable; try again`,
3290/// which is what a client hears when a repository is genuinely broken. The
3291/// power-cut door read the second sentence during the first condition and
3292/// reported 64 acked pushes missing that were all present a second later. A
3293/// harness, a CI script and a person all have to be able to tell "not yet" from
3294/// "gone", and the only place that distinction can be made is here, in the
3295/// words.
3296///
3297/// It carries [`EXIT_BUSY`], not [`EXIT_DENIED`], for the same reason: 75 means
3298/// retry.
3299pub const STARTING_MESSAGE: &str =
3300    "the server is starting: its stores are still opening after a restart, and it will serve \
3301     this in a moment; retry";
3302
3303#[cfg(test)]
3304mod tests {
3305    use super::*;
3306
3307    /// `gunnar-recipient-keys`: the path through the one sandbox (git's
3308    /// quoting taken off), the member through the principal id rule, and
3309    /// nothing else on the line; another verb is not this one.
3310    #[test]
3311    fn the_recipient_keys_line_takes_a_sandboxed_path_and_an_account_name() {
3312        let ok = |line: &str| parse_recipient_keys_verb(line.as_bytes()).expect("this verb").expect("well formed");
3313        let (repo, member) = ok("gunnar-recipient-keys 'team/vault.git' 'bob'");
3314        assert_eq!((repo.namespace(), repo.name(), member.as_str()), ("team", "vault", "bob"));
3315        let (repo, member) = ok("gunnar-recipient-keys team/vault bob");
3316        assert_eq!((repo.namespace(), repo.name(), member.as_str()), ("team", "vault", "bob"));
3317        let (repo, _) = ok("gunnar-recipient-keys '/team/vault.git' bob");
3318        assert_eq!(repo.name(), "vault");
3319        assert!(parse_recipient_keys_verb(b"git-upload-pack 'team/vault.git'").is_none());
3320        assert!(parse_recipient_keys_verb(b"gunnar-recipient-keysx a/b c").is_none());
3321        for bad in [
3322            "gunnar-recipient-keys",
3323            "gunnar-recipient-keys 'team/vault.git'",
3324            "gunnar-recipient-keys '../etc/passwd' bob",
3325            "gunnar-recipient-keys team/vault 'bo b'",
3326            "gunnar-recipient-keys team/vault anonymous",
3327            "gunnar-recipient-keys team/vault 'bob",
3328            "gunnar-recipient-keys team/vault bob\u{1}",
3329        ] {
3330            assert!(
3331                matches!(parse_recipient_keys_verb(bad.as_bytes()), Some(Err(_))),
3332                "{bad:?} is refused"
3333            );
3334        }
3335        let long = format!("gunnar-recipient-keys team/vault {}", "b".repeat(600));
3336        assert!(matches!(parse_recipient_keys_verb(long.as_bytes()), Some(Err(_))));
3337    }
3338
3339    /// **The door offers the post-quantum hybrid first**, from gunnar's own
3340    /// list: `mlkem768x25519-sha256`, then the classical fallbacks, and the
3341    /// strict-kex markers (Terrapin) kept.
3342    #[test]
3343    fn the_door_offers_mlkem768x25519_first_from_gunnars_own_list() {
3344        let p = preferred();
3345        assert!(std::ptr::eq(p.kex.as_ref().as_ptr(), KEX_ORDER.as_ptr()), "the server's list is gunnar's KEX_ORDER, not whatever russh defaults to");
3346        let names: Vec<&str> = p.kex.iter().map(|n| n.as_ref()).collect();
3347        assert_eq!(names[..3], ["mlkem768x25519-sha256", "curve25519-sha256", "curve25519-sha256@libssh.org"], "{names:?}");
3348        assert!(names.contains(&"kex-strict-s-v00@openssh.com"), "strict kex stays: {names:?}");
3349        assert_eq!(p.cipher, russh::Preferred::DEFAULT.cipher, "only the key exchange is gunnar's own");
3350    }
3351
3352    /// The accept loop outlives a descriptor shortage and a client that hung
3353    /// up mid-handshake; it returns on a listener that is gone.
3354    #[test]
3355    fn transient_accept_errors_are_named() {
3356        for errno in [24, 23, 105, 12, 103] {
3357            assert!(
3358                accept_error_is_transient(&std::io::Error::from_raw_os_error(errno)),
3359                "errno {errno}"
3360            );
3361        }
3362        for errno in [9, 22, 88] {
3363            assert!(
3364                !accept_error_is_transient(&std::io::Error::from_raw_os_error(errno)),
3365                "errno {errno}"
3366            );
3367        }
3368        assert!(!accept_error_is_transient(&std::io::Error::other("closed")));
3369    }
3370
3371    /// A teardown that fails is **reported**, and one that works is
3372    /// silent.
3373    ///
3374    /// Driven through the helper rather than through a live channel: making
3375    /// `Channel::close()` fail on demand means a peer that has vanished at an
3376    /// exact moment, which is a race to arrange and a race to assert on. What
3377    /// the helper decides is the whole of what changed, `let _ =` against a
3378    /// line in the journal, and both directions are asserted, because a
3379    /// `note_teardown` that logged unconditionally would put a line in an
3380    /// operator's journal on every clean disconnect and make the signal
3381    /// useless.
3382    /// Capture every `tracing` line `emit` produces, rendered as
3383    /// `<target> <field>=<value> …`.
3384    ///
3385    /// One capture for every test in this module that reads the journal (one
3386    /// implementation): there were two copies of this layer forty lines apart, and a second
3387    /// copy of a capture is a second chance to filter out the line under test.
3388    fn captured_lines(emit: impl FnOnce()) -> Vec<String> {
3389        use std::sync::{Arc, Mutex};
3390        use tracing_subscriber::layer::SubscriberExt;
3391        use tracing_subscriber::Layer as _;
3392
3393        #[derive(Clone, Default)]
3394        struct Capture(Arc<Mutex<Vec<String>>>);
3395        struct CaptureLayer(Capture);
3396        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CaptureLayer {
3397            fn on_event(
3398                &self,
3399                event: &tracing::Event<'_>,
3400                _: tracing_subscriber::layer::Context<'_, S>,
3401            ) {
3402                struct Render(String);
3403                impl tracing::field::Visit for Render {
3404                    fn record_debug(
3405                        &mut self,
3406                        field: &tracing::field::Field,
3407                        value: &dyn std::fmt::Debug,
3408                    ) {
3409                        self.0.push_str(&format!("{}={value:?} ", field.name()));
3410                    }
3411                }
3412                let mut render = Render(String::new());
3413                event.record(&mut render);
3414                self.0 .0.lock().expect("capture").push(format!(
3415                    "{} {}",
3416                    event.metadata().target(),
3417                    render.0
3418                ));
3419            }
3420        }
3421
3422        let capture = Capture::default();
3423        let subscriber = tracing_subscriber::registry().with(
3424            CaptureLayer(capture.clone())
3425                .with_filter(tracing_subscriber::filter::LevelFilter::TRACE),
3426        );
3427        tracing::subscriber::with_default(subscriber, emit);
3428        let lines = capture.0.lock().expect("capture").clone();
3429        lines
3430    }
3431
3432    #[test]
3433    fn a_failed_teardown_step_is_reported_and_a_clean_one_is_not() {
3434        let lines = captured_lines(|| {
3435            note_teardown::<std::io::Error>("channel-close", Ok(()));
3436            note_teardown(
3437                "watchdog-disconnect",
3438                Err(std::io::Error::other("the peer was already gone")),
3439            );
3440        });
3441
3442        assert_eq!(
3443            lines.len(),
3444            1,
3445            "expected exactly one line — the failure — got {lines:?}"
3446        );
3447        assert!(
3448            lines[0].starts_with(gunnar_core::observe::TARGET_SSH),
3449            "the line is not on the ssh target: {}",
3450            lines[0]
3451        );
3452        assert!(
3453            lines[0].contains("watchdog-disconnect") && lines[0].contains("already gone"),
3454            "the line names neither the step nor the error: {}",
3455            lines[0]
3456        );
3457    }
3458
3459    /// **RED-WHEN-BROKEN: a real failure is still reported as one.**
3460    ///
3461    /// The dangerous direction of [`peer_hung_up`] is not missing a hung-up
3462    /// peer — that only restores the noise. It is answering `true` for
3463    /// everything, which gives a beautifully quiet journal for a server that is
3464    /// on fire, and does it in a way no other test would notice.
3465    ///
3466    /// So every case is asserted in both directions, against **rendered log
3467    /// output** rather than against the predicate, because the predicate is
3468    /// private and the log line is what an operator has.
3469    #[test]
3470    fn a_real_failure_is_still_reported_as_one() {
3471        // Benign: the peer went away. Each of these is a shape MEASURED in the
3472        // live journal on 2026-08-06 for a transfer that succeeded.
3473        let benign: Vec<(&str, Box<dyn std::error::Error + 'static>)> = vec![
3474            // `err=Channel send error` — the channel's sender is gone.
3475            ("russh SendError", Box::new(russh::Error::SendError)),
3476            ("russh HUP", Box::new(russh::Error::HUP)),
3477            ("russh Disconnect", Box::new(russh::Error::Disconnect)),
3478            // `error=early eof` — tokio's `read_exact` on a closed socket.
3479            (
3480                "early eof",
3481                Box::new(std::io::Error::new(
3482                    std::io::ErrorKind::UnexpectedEof,
3483                    "early eof",
3484                )),
3485            ),
3486            (
3487                "broken pipe",
3488                Box::new(std::io::Error::from(std::io::ErrorKind::BrokenPipe)),
3489            ),
3490            (
3491                "connection reset",
3492                Box::new(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
3493            ),
3494        ];
3495        for (what, err) in benign {
3496            let lines = captured_lines(|| note_teardown("channel-close", Err(BoxedError(err))));
3497            assert_eq!(lines.len(), 1, "{what}: expected one line, got {lines:?}");
3498            let line = lines[0].to_ascii_lowercase();
3499            assert!(
3500                line.contains("peer_gone"),
3501                "{what} was not classified as a hung-up peer: {}",
3502                lines[0]
3503            );
3504            assert!(
3505                !line.contains("error") && !line.contains("err=") && !line.contains("fail"),
3506                "{what} is benign but its line is still found by `grep -i error`: {}",
3507                lines[0]
3508            );
3509        }
3510
3511        // NOT benign: the protocol itself went wrong. These must keep saying so
3512        // — this is the half that stops the classifier becoming a blanket yes.
3513        let real: Vec<(&str, Box<dyn std::error::Error + 'static>)> = vec![
3514            ("key exchange", Box::new(russh::Error::Kex)),
3515            ("packet auth", Box::new(russh::Error::PacketAuth)),
3516            ("decryption", Box::new(russh::Error::DecryptionError)),
3517            ("inconsistent state", Box::new(russh::Error::Inconsistent)),
3518            ("no common algo", Box::new(russh::Error::WrongChannel)),
3519            (
3520                "permission denied on disk",
3521                Box::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
3522            ),
3523            (
3524                "out of descriptors",
3525                Box::new(std::io::Error::from(std::io::ErrorKind::Other)),
3526            ),
3527        ];
3528        for (what, err) in real {
3529            let lines = captured_lines(|| note_teardown("channel-close", Err(BoxedError(err))));
3530            assert_eq!(lines.len(), 1, "{what}: expected one line, got {lines:?}");
3531            assert!(
3532                lines[0].contains("ssh.teardown.failed"),
3533                "{what} was swallowed as benign, so the classifier is a blanket yes: {}",
3534                lines[0]
3535            );
3536            assert!(
3537                lines[0].contains("err="),
3538                "{what} lost the error itself, so the line names no cause: {}",
3539                lines[0]
3540            );
3541        }
3542    }
3543
3544    /// A boxed error that is still `std::error::Error`, so the table above can
3545    /// hold two concrete types in one `Vec` and still reach the classifier's
3546    /// `downcast_ref` through `source()`.
3547    #[derive(Debug)]
3548    struct BoxedError(Box<dyn std::error::Error + 'static>);
3549
3550    impl std::fmt::Display for BoxedError {
3551        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
3552            self.0.fmt(f)
3553        }
3554    }
3555
3556    impl std::error::Error for BoxedError {
3557        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
3558            Some(self.0.as_ref())
3559        }
3560    }
3561
3562    /// The session-level half of the same rule: an ended session says nothing
3563    /// an operator greps for when the client simply hung up, and still says
3564    /// `error` when it did not.
3565    ///
3566    /// Asserted through the real `Handler` method rather than the predicate,
3567    /// because the method is what russh calls.
3568    #[test]
3569    fn a_hung_up_session_is_not_logged_as_an_error() {
3570        let quiet = captured_lines(|| {
3571            note_session_end(&anyhow::Error::new(std::io::Error::new(
3572                std::io::ErrorKind::UnexpectedEof,
3573                "early eof",
3574            )));
3575        });
3576        assert_eq!(quiet.len(), 1, "expected one line, got {quiet:?}");
3577        let line = quiet[0].to_ascii_lowercase();
3578        assert!(
3579            line.contains("peer-hung-up") && !line.contains("error"),
3580            "a client hanging up is still logged as an error: {}",
3581            quiet[0]
3582        );
3583
3584        let loud = captured_lines(|| {
3585            note_session_end(&anyhow::Error::new(russh::Error::Kex));
3586        });
3587        assert_eq!(loud.len(), 1, "expected one line, got {loud:?}");
3588        assert!(
3589            loud[0].to_ascii_lowercase().contains("error"),
3590            "a key-exchange failure was swallowed as a hung-up peer: {}",
3591            loud[0]
3592        );
3593    }
3594
3595    /// An unencodable key is **refused and counted**, never accepted.
3596    ///
3597    /// The assertion is on the applied output, the `Auth` value that goes back
3598    /// to russh and the counter an operator reads, not on "it did not panic".
3599    /// A body that returned `Auth::Accept` would let a key nobody could
3600    /// fingerprint onto the server, which is the one outcome worse than the
3601    /// panic this replaced.
3602    #[test]
3603    fn a_key_that_does_not_re_encode_is_rejected_and_counted() {
3604        let metrics = gunnar_core::observe::ServerMetrics::default();
3605        let auth = refuse_unencodable_key(&metrics, "decorative");
3606
3607        assert!(
3608            matches!(auth, Auth::Reject { .. }),
3609            "an unencodable key must be rejected, got {auth:?}"
3610        );
3611        let snapshot = metrics.snapshot();
3612        assert_eq!(
3613            snapshot.connections_refused, 1,
3614            "the refusal must be visible to an operator"
3615        );
3616        // Not an identity-table fault, and the counters must not blur the two.
3617        assert_eq!(snapshot.auth_faults, 0);
3618        assert_eq!(snapshot.connections_accepted, 0);
3619    }
3620}
3621
3622/// Pure: whether an exec line is git-lfs asking SSH (`git-lfs-authenticate …`,
3623/// `git-lfs-transfer …`) — refused, and said at INFO.
3624pub fn is_lfs_over_ssh(command: &str) -> bool {
3625    let first = command.split_whitespace().next().unwrap_or("");
3626    // The transfer verb is named through the dispatch table's own LFS row, never spelled here
3627    // (`command_parsing::the_dispatch_table_is_the_only_place_a_verb_is_named`).
3628    // `git-lfs-authenticate` is no verb gunnar dispatches: it is git-lfs's own probe.
3629    first == "git-lfs-authenticate"
3630        || crate::command::SERVICES
3631            .iter()
3632            .any(|spec| spec.verb == first && matches!(spec.service, crate::command::GitService::LfsTransfer(_)))
3633}
3634
3635#[cfg(test)]
3636mod lfs_over_ssh_tests {
3637    /// ★ git-lfs's SSH probe is a client's known step, not a defect (RUN A 2026-10-08).
3638    #[test]
3639    fn git_lfs_asking_ssh_is_known_and_anything_else_is_not() {
3640        assert!(super::is_lfs_over_ssh("git-lfs-authenticate /cbc/vault upload"));
3641        assert!(super::is_lfs_over_ssh("git-lfs-transfer 'acme/site' download"));
3642        for other in ["bash", "scp -t /etc", "", "git-lfs-authenticatex /a upload", "echo git-lfs-authenticate"] {
3643            assert!(!super::is_lfs_over_ssh(other), "{other:?}");
3644        }
3645    }
3646}