crates/gunnar-ssh/src/seam.rs
//! **The one remaining seam: temporary, and marked so.**
//!
//! # What used to be here, and why it is gone
//!
//! This module carried three traits, because `gunnar-core` (the ACL model) and
//! `gunnar-wire` (the git wire protocol) were being written in parallel with this crate and
//! blocking on either would have meant writing no server at all.
//!
//! **`gunnar-core` has landed, so two of them are deleted.** `AuthDirectory` and
//! `RepoResolver` are now [`crate::control::Control`] over the real
//! [`AuthTable`](gunnar_core::AuthTable) and [`RepoCatalog`](gunnar_core::RepoCatalog),
//! and their test doubles went with them. They were not thin adapters worth
//! keeping: both returned `Option`, and `gunnar-core` deliberately returns
//! `Result` because the table lives in storage and a read can fail. Adapting one
//! to the other means choosing an answer for the failure case, and the
//! safe-looking choice (`None`) turns a storage fault into a silent permission
//! denial. That is precisely what the ACL model says not to do, and it is what the
//! seam was doing.
//!
//! **`gunnar-wire` has not landed**, so [`GitTransport`] stays exactly as it
//! was. It is the seam the protocol state machines plug into, and wiring it is
//! a separate job.
//!
//! | trait | delegates to | status |
//! |---|---|---|
//! | ~~`AuthDirectory`~~ | `gunnar_core::AuthTable` | **deleted** |
//! | ~~`RepoResolver`~~ | `gunnar_core::RepoCatalog` | **deleted** |
//! | [`GitTransport`] | `gunnar-wire`'s `upload_pack` / `receive_pack` | still a seam |
//!
//! # The test double
//!
//! [`BannerTransport`] is a **test double, not a backend**. It exists so this
//! crate's tests can prove the SSH machinery end to end without `gunnar-wire`.
//! Nothing outside a test should construct it.
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use gunnar_core::{Fingerprint, Level, Principal, Repo};
use tokio::io::{AsyncRead, AsyncWrite};
use crate::command::GitService;
/// A bidirectional byte stream: what an SSH channel is once the exec has been
/// accepted.
///
/// Blanket-implemented, so `Box<dyn DuplexStream>` accepts russh's
/// `ChannelStream` and, in a test, a `tokio::io::DuplexStream`.
pub trait DuplexStream: AsyncRead + AsyncWrite + Unpin + Send {}
impl<T: AsyncRead + AsyncWrite + Unpin + Send> DuplexStream for T {}
/// Which part of a git conversation one call to the seam carries.
///
/// # Why an SSH crate names an HTTP shape
///
/// This module is **the seam**, not the SSH transport: it is the join every
/// front end plugs the protocol state machines into, and `gunnar-http` plugs
/// into this one rather than growing a second copy of it. That is the whole
/// reason the crate docs above call `GitTransport` "the seam" and not "the SSH
/// bridge".
///
/// The alternative was a second trait in `gunnar-http` with its own
/// `ServiceRequest`, its own authorisation call and its own join to
/// `gunnar-wire` — which is exactly the shape that produces two answers to
/// *"may this principal push"*. Sharing one seam costs an enum with three
/// variants and no dependencies; `gunnar-http` linking `gunnar-ssh` adds
/// nothing to any build, because a gunnar server that speaks HTTP also speaks
/// SSH and `gunnar-ssh` was already there.
///
/// If a build ever wants HTTP **without** SSH, this module and
/// [`crate::server::Dispatch`] move to their own crate and nothing else
/// changes. That build does not exist today, and inventing the crate for it
/// now would be inventing a boundary before there is anything on both sides of
/// it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Delivery {
/// One connection carries the advertisement, the negotiation and the pack.
/// This is SSH, and it is what `gunnar-wire`'s `Session::Connected` means.
#[default]
Session,
/// Smart HTTP `GET /<path>/info/refs?service=…`: write the advertisement
/// and stop. Nothing is read from the client.
Advertisement,
/// Smart HTTP `POST /<path>/<service>`: one stateless-rpc round.
StatelessRpc,
}
/// Everything the wire needs to know about the session it is serving.
#[derive(Debug, Clone)]
pub struct ServiceRequest {
/// Which service the client asked for.
pub service: GitService,
/// The repository, as the catalog holds it, carrying the hash kind the wire
/// has to echo in `object-format` (the git wire protocol) and the default branch.
///
/// When [`Self::prospective`] is set this is the repository the
/// first-repository door **would** create — the same value
/// `RepoCatalog::create` would write — and it is not in the catalog yet.
pub repo: Repo,
/// **The first-repository door is open for this push, and nothing has
/// been written.** Only ever `true` for [`GitService::ReceivePack`] or an
/// LFS upload (`GitService::LfsTransfer(Upload)`, 2026-09-28) to a
/// repository that does not exist, by a principal who may create it. An
/// LFS upload's repository is made when its first object arrives. The
/// transport advertises `repo` as an empty repository, reads the client's
/// commands, checks the hash, and only then asks
/// `Control::admit_push_create` to write the row and makes the store. A
/// client that hangs up at the advertisement leaves no trace.
pub prospective: bool,
/// Who is asking. Carried into the push attestation and the
/// event stream.
pub principal: Principal,
/// The level the principal was found to hold, which the wire may need for
/// finer decisions than "may push at all".
pub access: Level,
/// **The key this session actually authenticated with**.
///
/// `None` means no key did, the anonymous read-only path, which
/// is a real state and not a missing value, so it is an `Option` rather
/// than an empty string.
///
/// The server computed this in `auth_publickey` and looked the principal up
/// by it, then used to drop it. Downstream, `gunnar-cli`'s push attestation
/// and its `RefUpdated` / `RefDeleted` events had to *guess* it back out of
/// the identity table, and could only do so when the principal held exactly
/// one key, so a real deploy account with two keys published an empty
/// fingerprint, and the audit trail could not answer "which key pushed
/// this".
///
/// It is a **parameter of [`crate::Dispatch::authorise`]**, not something
/// attached afterwards like [`ServiceRequest::with_git_protocol`], and the
/// difference is deliberate: an attach-after-the-fact field can be
/// forgotten, and a forgotten one is silently `None`, which is exactly the
/// shape of the bug this closes. The compiler asks for it instead.
/// `git_protocol` is the other way round for the opposite reason: it comes
/// from the *client*, and must not be able to reach the grant check.
pub key_fingerprint: Option<Fingerprint>,
/// The raw value of the client's `GIT_PROTOCOL` environment variable, if it
/// sent one and the server's allowlist accepted it (see
/// [`crate::server::ENV_ALLOWLIST`]).
///
/// **Raw, not parsed**, and deliberately: the grammar is
/// `gunnar-wire`'s (`ProtocolVersion::from_git_protocol`), and this crate
/// does not depend on `gunnar-wire`. Parsing it here would be a second
/// implementation of a version negotiation that already exists, and the two
/// could disagree about `version=1` (one implementation).
///
/// `None` means the client sent nothing, which is v0. It is an `Option`
/// rather than a defaulted `ProtocolVersion` so that "the client asked for
/// v0" and "the client never got to ask" stay distinguishable in a log.
pub git_protocol: Option<String>,
/// Which part of the conversation this call carries.
///
/// [`Delivery::Session`] for SSH, always. Attached afterwards like
/// [`ServiceRequest::git_protocol`] rather than taken as a parameter of
/// [`crate::server::Dispatch::authorise`], and unlike the key fingerprint
/// that is safe here for a specific reason: **forgetting it fails loudly.**
/// An HTTP `POST` served as a `Session` writes a ref advertisement in front
/// of the negotiation, and `git` rejects that on the first packet. The
/// fingerprint's failure mode is the opposite — a forgotten one is silently
/// `None` and shows up months later as an empty field in an audit trail —
/// which is why that one is a parameter and this one is not.
pub delivery: Delivery,
/// **Which credential authenticated this request**, as the audit trail
/// should name it: `key:SHA256:…` or `token:<hash prefix>`.
///
/// `None` means a caller that presented no credential at all — the
/// anonymous read path — and it is the ONLY thing `None` means here.
///
/// It exists because [`Self::key_fingerprint`] cannot mean that (GUN-014).
/// HTTP authenticates with a **token**, so there is no key to put in that
/// field and it is `None` on every HTTP request, `git-receive-pack`
/// included — while the code that consumes it states that an empty
/// fingerprint has "one reason only — the anonymous, read-only path… That
/// path cannot push". A push therefore reached `RefUpdated`, `RefDeleted`
/// and the attestation with an empty fingerprint and no other field to say
/// who had made it.
///
/// A builder and not a parameter, unlike the fingerprint, and the reason
/// is the opposite of that one's: the SSH front door does not set it at
/// all. `key_fingerprint` is already the answer there, and deriving
/// `key:<fp>` in one place (`transport::session_credential`) is one
/// implementation rather than two front doors spelling the same prefix.
pub credential: Option<String>,
/// **The gate chain's warning line**: a Grace tenant's push is allowed and
/// told `plan expired <date>; renew at <url>`. The transport puts it on the
/// sideband so `git push` prints `remote: …`. `None` for every other state,
/// for every read, and on a community build (whose one gate never warns).
pub plan_notice: Option<String>,
/// **The gate chain's byte ceiling for this push**, the smallest any gate
/// asked for. The transport hands it to `gunnar-wire`, which counts the
/// pack off the wire and cuts the transfer past it — see
/// [`crate::PackCeiling`] and [`crate::PlanGate::pack_ceiling`].
///
/// `None` for every read, for every write on a namespace with no byte cap,
/// and on a community build (whose one gate asks for no ceiling), and
/// `None` then means what it has always meant: read the pack to its end.
///
/// Decided in `authorise_service` rather than looked up in the transport
/// for the same reason [`ServiceRequest::plan_notice`] is: the gate chain
/// lives on the front door's `Control`, and the transport has never been
/// able to see it.
///
/// # What reads it today, said plainly
///
/// **The receive path, and only the receive path.** The chain asks for a
/// ceiling wherever it asks `pack_quota`, which includes an LFS upload — and
/// the LFS transfer ignores this field, deliberately. What LFS needed was
/// its own cut against its own meter (`lfs_bytes`, not `pack_bytes`), and
/// since 2026-09-07 it has one: [`ServiceRequest::lfs_allowance`]. Half a
/// ceiling carried here would have been the twin of a limit that fires on
/// one transport and not the other. A `Machine` principal is never asked at
/// all ([`crate::PlanGate::applies_to_machine`]), so a primary's replication
/// push is never cut by a tenant's plan.
pub pack_ceiling: Option<crate::plan::PackCeiling>,
/// **Where this push stands in the namespace's ledger of pack bytes** — the
/// value the `pack_bytes` counter had when
/// [`crate::server::Dispatch::ask_plan`] measured, taken BEFORE any gate
/// walked the disk. `Some` exactly when [`Self::pack_ceiling`] is.
///
/// `gunnar-server` binds the two together and hands the result to
/// `gunnar-wire`, whose receive loop charges the ledger for every chunk
/// before it takes it — so the ceiling is the room the whole namespace has
/// left, not the room each connection is separately told it has.
///
/// # Why this is a field beside the ceiling and not a field INSIDE it
///
/// The LFS wire folded its mark into the number
/// ([`gunnar_lfs::NamespaceAllowance`]) because there the two are used
/// together by one callee. Here they are not: the number goes into
/// `gunnar_wire::PackCeiling`, in a crate that cannot name a
/// `gunnar_lfs::Mark` and must not learn to, and the mark goes into the
/// same struct as a `dyn gunnar_wire::PackLedger`. They are joined at
/// `gunnar_server::transport`, the seam that already exists for exactly
/// this conversion, so folding them together here would be folding them for
/// the journey and nothing else.
///
/// The other half of the reason is whose fact it is. [`crate::PackCeiling`]
/// is what a [`crate::PlanGate`] ANSWERED, and `PlanGate` is a public trait
/// anyone may implement; the mark is the SESSION's standing in a counter
/// this crate owns, taken before any gate is consulted and existing whether
/// or not one answers. Putting it in the gate's return type would have
/// asked every implementor to produce something only the dispatch can, and
/// changed the type of a field the server's tests read.
pub pack_mark: Option<gunnar_lfs::Mark>,
/// **The gate chain's LFS room for this transfer**, the smallest any gate
/// admitted — the twin of [`ServiceRequest::pack_ceiling`] one wire over.
/// The transport hands it to `gunnar_lfs::Session` (SSH) and to
/// [`LfsAccess`] (smart HTTP), and an upload past it is refused 507 with the
/// plan's own sentence, before the object's bytes.
///
/// `None` for every read, for every namespace with no `lfs_bytes` cap, and
/// on a community build — and `None` means what the pack ceiling's `None`
/// means: take what arrives.
///
/// The pack ceiling would not have done: they are different meters on the
/// signed fact, they are reported separately by `Usage.Read`, and one is
/// counted in a receive loop while the other is declared up front. See
/// [`crate::PlanGate::lfs_allowance`].
///
/// # Why a `NamespaceAllowance` and not the plain `gunnar_lfs::Allowance`
///
/// Because a plain one is a number measured for ONE transfer, and two
/// transfers that started together measured the same "before". This one
/// carries, beside the number, where the transfer stands in the namespace's
/// ledger of bytes admitted since — so the door counts what the other
/// connections are already doing. `gunnar-server` still decides the number
/// (`PlanGate::lfs_allowance` returns the plain `Allowance` and is
/// unchanged); [`crate::server::Dispatch::ask_plan`] binds it to the
/// ledger, which is also the one place that can guarantee the mark is taken
/// BEFORE the disk is walked. See `gunnar_lfs::ledger`.
pub lfs_allowance: Option<gunnar_lfs::NamespaceAllowance>,
/// **Set when the client is gone and the transfer must stop** at its next
/// safe point (`crate::channel::serve_until_session_ends`). A transport
/// whose git work runs off the async task checks it there, so the work
/// ends before the fan worker serving it is given back.
pub cancel: crate::channel::Cancel,
}
impl ServiceRequest {
/// Attach the client's `GIT_PROTOCOL` value.
///
/// Separate from the authorisation decision on purpose: what a client may
/// do is decided from its key and the catalog, and an environment variable
/// must never be able to reach that decision. This carries a *hint about
/// framing* to the wire, nothing more.
pub fn with_git_protocol(mut self, value: Option<String>) -> Self {
self.git_protocol = value;
self
}
/// Say which part of the conversation this call carries. See
/// [`Delivery`].
pub fn with_delivery(mut self, delivery: Delivery) -> Self {
self.delivery = delivery;
self
}
/// Name the credential this request authenticated with. See
/// [`ServiceRequest::credential`].
pub fn with_credential(mut self, credential: Option<String>) -> Self {
self.credential = credential;
self
}
}
/// What went wrong while serving.
#[derive(Debug, thiserror::Error)]
pub enum TransportError {
/// The protocol state machine refused the request.
#[error("protocol error: {0}")]
Protocol(String),
/// The stream failed.
#[error("i/o error: {0}")]
Io(#[from] std::io::Error),
/// Not wired yet. **This is the honest answer until `gunnar-wire` lands**,
/// and it must never quietly become a successful empty response: a clone
/// that returns no objects and exits zero looks exactly like a working
/// server serving an empty repository (measured, not asserted).
#[error("not implemented: {0}; the wire seam is not connected to gunnar-wire yet")]
Unimplemented(&'static str),
/// The transfer was refused **before a single byte reached the client**, and
/// the string is the sentence the client is owed.
///
/// This variant exists because the other three could not carry one. A
/// transfer that failed after the advertisement went out cannot be answered
/// with a fresh pkt-line — the client is mid-pack and the framing belongs to
/// `gunnar-wire`'s sideband, which is what `UploadPack::report_error` is
/// for. A transfer that failed *before* the first write can be answered, and
/// until this variant existed it was not: the session logged the whole
/// reason and sent the client an exit status and no words, which git renders
/// as `the remote end hung up unexpectedly`.
///
/// So the variant is the *fact* "nothing has been written yet", not a
/// severity. Only a construction site that knows that may use it, which is
/// why it is a distinct variant rather than a flag on
/// [`TransportError::Protocol`].
#[error("{0}")]
Refused(String),
}
impl TransportError {
/// What the client is told, or `None` when the client cannot be told
/// anything.
///
/// Named after [`crate::server::DispatchError::client_message`] and
/// `gunnar_wire::Error::client_message` because it is the same contract in a
/// third place (one implementation): a refusal the operator can see and the user cannot
/// is half a fix.
///
/// `None` for [`TransportError::Protocol`] and [`TransportError::Io`]
/// **on purpose**, and it is the conservative answer rather than the
/// convenient one: those two are raised from inside a transfer that may
/// already have written an advertisement, a `NAK` or half a pack, and a raw
/// pkt-line appended to that stream is corruption, not an error message.
pub fn client_message(&self) -> Option<String> {
match self {
TransportError::Protocol(_) | TransportError::Io(_) => None,
// Nothing has been served, by definition of the variant, so the
// client can be told — and must be, because the alternative is a
// clone that exits non-zero with no reason at all.
TransportError::Unimplemented(_) => Some(self.to_string()),
TransportError::Refused(message) => Some(message.clone()),
}
}
}
/// An authorised repository's LFS object store, and what it will accept.
///
/// Returned by [`LfsTransport::open_lfs`], and deliberately not a session: LFS
/// over HTTP is four independent requests (`batch`, `GET`, `PUT`, `verify`),
/// each authorised on its own, so there is nothing for a session to hold
/// between them. What the two surfaces share is the *store* and the *limits*,
/// which is exactly what this carries.
pub struct LfsAccess {
/// Where the objects are. `Arc` because the caller runs the actual I/O on a
/// blocking thread and has to own it there.
pub store: std::sync::Arc<dyn gunnar_lfs::LfsStore>,
/// The quota and the batch cap in force, from the **same**
/// `WireConfig::lfs` the SSH session is built with. A second home for these
/// is how 8 GiB comes to be enforced on one transport and not the other.
pub limits: gunnar_lfs::Limits,
/// **The tenant's remaining LFS room**, decided by the same
/// [`crate::PlanGate`] chain that decided
/// [`crate::ServiceRequest::pack_ceiling`] for a push, and `None` wherever a
/// push would get no ceiling: a community build, an uncapped namespace, a
/// download.
///
/// Beside [`Self::limits`] rather than inside it because they are different
/// kinds of number. `limits` is the operator's, one set for the whole box,
/// `Copy`, and comes off the command line; this is one namespace's plan,
/// re-measured per transfer, and carries a sentence built from a signed
/// entitlement fact. A tenant who fills it is told to buy more; a client
/// that trips `limits` is told the box does not take objects that big.
pub allowance: Option<gunnar_lfs::NamespaceAllowance>,
}
impl std::fmt::Debug for LfsAccess {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LfsAccess")
.field("limits", &self.limits)
.finish_non_exhaustive()
}
}
/// Open the LFS object store of a repository a caller has **already been
/// authorised for**.
///
/// # Why this is a seam and not a store handed to `gunnar-http` directly
///
/// The same reason [`GitTransport`] is. `gunnar-http` knows HTTP and the LFS
/// batch API; it must not know that a repository is a directory, that the
/// objects live under `<git_dir>/lfs/`, or that opening one can take a lock.
/// Those are `gunnar-cli`'s, which owns `ServerStores` and the one
/// `WireConfig` both transports are configured from.
///
/// # `request` is the output of the ACL, not an input to it
///
/// The parameter is a [`ServiceRequest`], which can only be got out of
/// [`crate::server::Dispatch::authorise_service`]. So an implementation cannot
/// be called without the grant check having already happened, and there is no
/// second authorisation path — the same property `GitTransport` has, held the
/// same way. `request.service` is `GitService::LfsTransfer(operation)` and the
/// operation in it is the one that demanded Read or Write.
///
/// # Blocking
///
/// Implementations do filesystem work. The caller runs them on a blocking
/// thread; nothing here is async, for the same reason `gunnar_lfs::Session` is
/// not.
pub trait LfsTransport: Send + Sync + 'static {
/// The object store for `request`'s repository.
fn open_lfs(&self, request: &ServiceRequest) -> Result<LfsAccess, TransportError>;
}
/// Serve `git-upload-pack` / `git-receive-pack` over an accepted channel.
///
/// This is `gunnar-wire`'s entry point seen from the SSH side. The returned
/// future owns the stream for the whole of the transfer; the server spawns it
/// rather than awaiting it inline, because russh dispatches handler calls from
/// the session's own read loop and awaiting there would stop the loop that has
/// to deliver the client's bytes.
pub trait GitTransport: Send + Sync + 'static {
/// Serve one request to completion.
fn serve<'a>(
&'a self,
request: ServiceRequest,
stream: Box<dyn DuplexStream>,
) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>>;
}
/// A [`GitTransport`] that writes one line describing the request it was given,
/// then reads the client out. **A test double.**
///
/// It exists to prove a specific thing that a mock recording a call cannot: that
/// the *byte stream* reached the seam, in both directions, after a real SSH
/// handshake. The line it writes names the service and the repository, so a test
/// asserting on it is asserting on applied output rather than on a flag someone
/// set (measured, not asserted).
#[derive(Debug, Default)]
pub struct BannerTransport {
served: Mutex<Vec<String>>,
git_protocols: Mutex<Vec<Option<String>>>,
settle_after_release: bool,
fail_at_the_end: bool,
/// Set by [`BannerTransport::refusing`]: the transport writes **nothing** and
/// returns [`TransportError::Refused`], which is the shape of every
/// post-authorisation refusal — a held store lock, a repository that will not
/// open, an unwired service.
refuse_before_writing: Option<String>,
/// Set by [`BannerTransport::holding`]: the transport writes its banner and
/// then does not return until the paired [`TransportHold`] is released.
hold: Option<tokio::sync::watch::Receiver<bool>>,
}
/// The release lever of [`BannerTransport::holding`].
///
/// Dropping it releases every held transfer, so a test that panics mid-way
/// cannot wedge a server task forever.
#[derive(Debug)]
pub struct TransportHold {
tx: tokio::sync::watch::Sender<bool>,
}
impl TransportHold {
/// Let every held transfer finish.
pub fn release(&self) {
let _ = self.tx.send(true);
}
}
impl Drop for TransportHold {
fn drop(&mut self) {
self.release();
}
}
impl BannerTransport {
/// A transport that has served nothing yet.
pub fn new() -> Self {
Self::default()
}
/// The same transport, but it **drops the stream and then waits** before it
/// returns.
///
/// This is the red half of the exit-status ordering guard, and it is worth
/// spelling out why it is not a contrivance.
///
/// russh's `Channel::into_stream()` wraps the channel in
/// `ChannelCloseOnDrop`, whose `Drop` **spawns a task** that sends
/// `ChannelMsg::Close`. A server that hands the transport such a stream and
/// sends `exit-status` afterwards is in a race with that task: the ordinary
/// path wins it on a current-thread runtime (the send takes an available
/// permit without yielding) and loses it often enough on a multi-threaded
/// one to make a *successful* push report failure (measured 2026-08-03,
/// `ssh` exit 255 with a complete 244-byte advertisement).
///
/// A test on the ordinary path therefore proves nothing: it is green with
/// the bug present. This mode gives the spawned task the scheduling it needs
/// to run, which turns "usually" into "always" **in the failing direction
/// only**. A server that closes nothing on drop is unaffected by any amount
/// of waiting here, so a green under this transport is not a timing
/// accident; it is the absence of the close.
pub fn settling_after_it_releases_the_stream() -> Self {
BannerTransport {
settle_after_release: true,
..Self::default()
}
}
/// A transport whose transfers **stay in flight** until the returned
/// [`TransportHold`] is released.
///
/// This is what makes the admission gate testable at all. A limit on
/// *concurrent* transfers cannot be reached by a transport that finishes
/// instantly: a hundred sequential clones never put two in flight together,
/// and a test written against the ordinary banner would be asserting on
/// scheduling luck. Here the first N transfers are genuinely simultaneous
/// and stay that way until the test says otherwise.
///
/// The banner is written **before** the wait, so a client knows its transfer
/// was admitted rather than merely accepted.
pub fn holding() -> (Self, TransportHold) {
let (tx, rx) = tokio::sync::watch::channel(false);
(
BannerTransport {
hold: Some(rx),
..Self::default()
},
TransportHold { tx },
)
}
/// A transport that refuses **without writing a byte**, the way a transfer
/// whose repository will not open does.
///
/// Distinct from [`BannerTransport::that_then_fails`] on the one axis that
/// matters: that one writes its banner first, so the client has already had
/// output and the framing is no longer the session's to add to. This one has
/// written nothing, which is the *only* condition under which the session may
/// still hand the client a reason.
pub fn refusing(message: impl Into<String>) -> Self {
Self {
refuse_before_writing: Some(message.into()),
..Self::default()
}
}
/// …and then reports that the transfer failed.
///
/// The other direction of the same guard: a **non-zero** status has to
/// survive the stream being released too. Without this, the exit-status
/// assertions could all be satisfied by a server that hard-coded zero.
pub fn that_then_fails(mut self) -> Self {
self.fail_at_the_end = true;
self
}
/// Everything it has served, as `"<verb> <ns>/<repo> <principal-id>"`.
pub fn served(&self) -> Vec<String> {
self.served.lock().expect("banner transport lock").clone()
}
/// The `GIT_PROTOCOL` value that arrived with each served request, in the
/// same order as [`BannerTransport::served`].
///
/// Recorded separately from the banner so that a test can assert *what the
/// wire would negotiate with* without changing the byte the existing
/// end-to-end assertions read back over SSH.
pub fn git_protocols(&self) -> Vec<Option<String>> {
self.git_protocols
.lock()
.expect("banner transport lock")
.clone()
}
/// The line this transport writes for `request`.
pub fn banner(request: &ServiceRequest) -> String {
format!(
"{} {} {}\n",
request.service.verb(),
request.repo.key(),
request.principal.id
)
}
}
impl GitTransport for BannerTransport {
fn serve<'a>(
&'a self,
request: ServiceRequest,
mut stream: Box<dyn DuplexStream>,
) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
Box::pin(async move {
use tokio::io::AsyncWriteExt;
// Before the banner, before `served` is recorded: a refusal that had
// already written or already counted itself would not be the thing
// under test.
if let Some(message) = &self.refuse_before_writing {
return Err(TransportError::Refused(message.clone()));
}
let banner = BannerTransport::banner(&request);
self.served
.lock()
.expect("banner transport lock")
.push(banner.trim_end().to_string());
self.git_protocols
.lock()
.expect("banner transport lock")
.push(request.git_protocol.clone());
stream.write_all(banner.as_bytes()).await?;
stream.flush().await?;
if let Some(hold) = &self.hold {
// Wait, still holding the stream, which is what an admitted,
// in-flight transfer looks like to the admission gate.
// A transfer told to stop (its client gone past the grace)
// stops, as the real transport does at its next safe point.
let mut hold = hold.clone();
let released = async {
while !*hold.borrow_and_update() {
if hold.changed().await.is_err() {
break;
}
}
};
tokio::select! {
() = released => {}
() = request.cancel.cancelled() => {
return Err(TransportError::Io(std::io::Error::new(
std::io::ErrorKind::ConnectionAborted,
"the banner double was told to stop: its client is gone",
)));
}
}
}
stream.shutdown().await?;
if self.settle_after_release {
// Explicit, so the drop happens here rather than at the end of
// the block, and anything it spawned has somewhere to run.
drop(stream);
for _ in 0..16 {
tokio::task::yield_now().await;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
if self.fail_at_the_end {
return Err(TransportError::Protocol(
"the banner double was asked to report a failed transfer".to_string(),
));
}
Ok(())
})
}
}