crates/gunnar-server/src/mirror_fetch.rs
//! **What `Mirror.Sync` actually does**: one fetch from an upstream, into this
//! box's own storage engine, and one compare-and-swap that moves the
//! references.
//!
//! # There is one client, and two ways to reach a remote
//!
//! [`gunnar_wire::fetch_pack`] is the whole protocol — the advertisement, the
//! `ls-refs`, the want/have negotiation, the sideband demultiplex, the packfile
//! streamed into a [`PackIngest`](gunnar_wire::PackIngest). It takes a
//! `&mut dyn BufRead` and a `&mut dyn Write` and cannot tell what is behind
//! them, which is the seam this module fills twice and only twice:
//!
//! | remote | the pair | who built it |
//! |---|---|---|
//! | `ssh://`, `git+ssh://`, `git@host:path` | a russh channel's two halves | `nornir-git`, through `gunnar_client::connect_upload_pack` — the **same dialler** replication pushes over |
//! | `https://`, `http://` | a buffered request/response adapter | [`crate::mirror_http`] |
//!
//! Neither is a second protocol implementation, and that is the point. A bug in
//! negotiation is one bug, in one file, in `gunnar-wire`.
//!
//! # This module writes the references, and `fetch_pack` deliberately does not
//!
//! `fetch_pack`'s module doc states the rule: it returns a **plan**
//! (`FetchReport::updates`, each entry a name, the old value and the new one)
//! and applies nothing, because applying a whole mirror's plan with no gate, no
//! atomicity across the batch and no way for an operator to see it first is not
//! something a protocol crate should be able to do.
//!
//! So the apply is here, and it is **one [`GitOps::put_refs_cas`]** with the
//! plan's own `old` as every expectation. That is not tidiness:
//!
//! * A mirror is not the only writer. `git push` reaches the same repository
//! over SSH and over smart HTTP, and a fetch that read a tip, spent four
//! minutes on the network and then wrote unconditionally would **silently
//! destroy** whatever landed in between. The `old` in the plan is read from
//! the same `GitOps::refs()` snapshot the `have` list came from, so the CAS
//! is checking exactly the value the fetch reasoned about.
//! * A CAS failure is a **FAILED job that names the reference**, never a skip.
//! "Synced" about a repository where one branch quietly did not move is the
//! answer that makes a mirror untrustworthy, and it reads identically to a
//! mirror that worked.
//! * Every edit or none. A half-applied plan leaves the repository at a state
//! no upstream ever had.
//!
//! # …and the batch goes through branch protection first
//!
//! A mirror moves references, and until 2026-09-11 it was the only writer on
//! this box that moved them without asking [`BranchProtectionGate`]. That is
//! not a smaller version of a push: `Mirror.Sync` needs **Write** and
//! `Policy.SetProtection` needs **Admin**, so a principal who may not configure
//! a protection rule could register a mirror against the repository it protects
//! and have the upstream's tips applied over it — a non-fast-forward onto a ref
//! whose rule says `allow_force_push: false`, applied by the one path that
//! never looked. See [`Protection`] for what is decided here and what is
//! refused as undecidable.
//!
//! # The credential is never in this module's output
//!
//! The value is fetched from the sealed secret store, held for the length of
//! one dial, and never formatted. An SSH key is materialised into a
//! `0600` temporary file because `nornir-git` authenticates from a path and
//! this tree does not fork a second dialler to change that (one implementation); the file is
//! unlinked when the fetch returns, on every path including a panic, because
//! [`tempfile::NamedTempFile`] unlinks on `Drop`. A token becomes an
//! `Authorization` header value inside [`crate::mirror_http`] and nothing else.
//! Errors that mention a URL go through [`scrub`] first.
use std::io::{BufRead, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use gix_hash::ObjectId;
use gunnar_core::MirrorCredentialKind;
use gunnar_store::git_storage_trait::{GitOps, RefCas};
use gunnar_wire::fetch_pack::{self, FetchPack, FetchProgress, FetchReport, LocalRefs};
use crate::job_queue::Reporter;
use crate::mirror_queue::{Mirror, MirrorOrder};
use crate::stores::ServerStores;
/// The tracing target a sync's lines carry — the same one every other git
/// transfer on this box uses, so `RUST_LOG=gunnar::git=info` selects a mirror
/// as well as a clone.
const TARGET_GIT: &str = "gunnar::git";
/// **The `known_hosts` a mirror's SSH dial trusts**, under the data root.
///
/// Not `~/.ssh/known_hosts`, and this is the same ruling replication's
/// `REPLICATION_KNOWN_HOSTS` makes for the push direction: a server's identity
/// is the box, not a home directory. The appliance has no home directory at
/// all, so the default would be a file that cannot be written and a
/// trust-on-first-use that silently never remembers — every sync a first
/// contact, which is the same as no host-key checking with extra steps.
///
/// A separate file from replication's, because they are separate rosters: the
/// standbys this box pushes to and the upstreams it pulls from are different
/// sets of machines, and a pin acquired for one is not consent for the other.
pub const MIRROR_KNOWN_HOSTS: &str = "mirror_known_hosts";
/// **The file that makes a mirror's SSH dial refuse an unknown host key**,
/// under the data root beside [`MIRROR_KNOWN_HOSTS`].
///
/// Without it a mirror dials trust-on-first-use: an upstream this box has never
/// seen is recorded and trusted, and only a *changed* key is refused. That is
/// the right default for a developer's laptop and a defensible one for a server
/// being brought up without a provisioning step — it is what replication's own
/// `strict_host_key` defaults to, and its comment already says that on "is the
/// right setting for anything that outlives an afternoon". The mirror side had
/// no way to say on at all: the policy was a literal in [`dial_ssh`].
///
/// A file rather than a field on `MirrorRecord`, for the reason
/// [`MIRROR_KNOWN_HOSTS`] is a file under the data root: the roster of hosts
/// this box will pull from is the box's, not one repository's, and the pin it
/// is checked against is already per-box. Create it once, and every mirror on
/// the box requires a `known_hosts` entry that somebody put there.
pub const MIRROR_STRICT_HOST_KEY: &str = "mirror_strict_host_key";
/// Does this box refuse an unknown upstream host key? See
/// [`MIRROR_STRICT_HOST_KEY`].
///
/// The presence of the file is read at each dial rather than cached, so an
/// operator who creates it does not have to restart the server to mean it.
pub fn strict_host_keys(data_root: &std::path::Path) -> bool {
data_root.join(MIRROR_STRICT_HOST_KEY).exists()
}
/// **The file that lets this box's mirrors dial an address that is not
/// publicly routable**, under the data root beside [`MIRROR_KNOWN_HOSTS`].
///
/// A file and not an environment variable, for the reason `wg.rs` writes out at
/// length: an env var is process-global, so a rule that can only be exercised
/// by `set_var` is a rule whose tests flake the moment `cargo test` runs them
/// on two threads. A file under the data root is per-box, survives a restart,
/// is visible to an operator looking at the box, and each test gets its own.
///
/// It exists because refusing every private destination outright would be a
/// product regression, not a fix: mirroring from a forge on the operator's own
/// LAN is an ordinary thing for an appliance to do. What must not be ordinary
/// is a **low-privileged** caller getting it for free — see
/// [`check_destination`].
pub const MIRROR_ALLOW_PRIVATE: &str = "mirror_allow_private_destinations";
/// **The credential a mirror dials with**, in the one shape it exists in
/// memory.
///
/// Deliberately not `Debug`, `Clone` or `Serialize`. The compiler is the only
/// mechanism that reliably stops a secret being formatted into a log line by
/// somebody adding a `{:?}` three months from now, so the derive that would
/// allow it is simply absent.
pub enum Credential {
/// No credential. A public upstream needs none, and a mirror of one works.
None,
/// An OpenSSH private key, for an `ssh://` upstream.
SshKey(Vec<u8>),
/// A token, for an `https://` upstream. `username` empty means `Bearer`.
Token { username: String, secret: String },
}
impl Credential {
fn kind(&self) -> MirrorCredentialKind {
match self {
Credential::None => MirrorCredentialKind::None,
Credential::SshKey(_) => MirrorCredentialKind::SshKey,
Credential::Token { .. } => MirrorCredentialKind::Token,
}
}
}
/// **Which transport a remote URL names.**
///
/// Decided by the URL and by nothing else — not by which credential happens to
/// be stored, and not by trying one and falling back. A fallback would turn "I
/// gave you the wrong kind of key" into "your upstream is down", and those go
/// to two different people.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Wire {
/// `ssh://`, `git+ssh://`, or the scp-like `git@host:ns/repo.git`.
Ssh,
/// `https://` or `http://`.
Http,
}
impl Wire {
/// The word an operator sees in a refusal.
fn name(self) -> &'static str {
match self {
Wire::Ssh => "ssh",
Wire::Http => "smart http",
}
}
}
/// **What transport `url` names, or a refusal that says what gunnar can dial.**
///
/// The scp-like form is recognised the way git recognises it: a `:` before any
/// `/`, with no scheme. A URL that is neither is refused rather than guessed
/// at — a mirror registered against a path on the local disk would otherwise
/// look like it was configured and then fail on the first sync with a
/// connection error.
pub fn classify(url: &str) -> Result<Wire, String> {
let url = url.trim();
if url.starts_with("ssh://") || url.starts_with("git+ssh://") {
return Ok(Wire::Ssh);
}
if url.starts_with("https://") || url.starts_with("http://") {
return Ok(Wire::Http);
}
if url.contains("://") {
return Err(format!(
"{} names a scheme gunnar cannot mirror from. The two transports are `ssh://` \
and `https://`",
scrub(url)
));
}
// `git@host:ns/repo.git` — a colon before any slash, and a host in front
// of it.
match (url.find(':'), url.find('/')) {
(Some(colon), slash) if colon > 0 && slash.is_none_or(|slash| colon < slash) => {
Ok(Wire::Ssh)
}
_ => Err(format!(
"{url:?} is not a remote gunnar can mirror from. Write it as \
`ssh://git@host/ns/repo.git`, `git@host:ns/repo.git` or `https://host/ns/repo.git`"
)),
}
}
// ── where a mirror may dial ──────────────────────────────────────────────────
/// **The host and port a mirror URL names**, in the three shapes
/// [`classify`] admits.
///
/// Deliberately not a URL parser: the only question asked of this string is
/// *which machine would be dialled*, and the answer is the authority. A
/// userinfo half is dropped — `git@host` names `host` — and an IPv6 literal
/// keeps its brackets off so it can be parsed as an address.
pub(crate) fn authority(url: &str) -> Result<(String, u16), String> {
let url = url.trim();
let (default_port, rest) = if let Some(rest) = url.strip_prefix("ssh://") {
(22, rest)
} else if let Some(rest) = url.strip_prefix("git+ssh://") {
(22, rest)
} else if let Some(rest) = url.strip_prefix("https://") {
(443, rest)
} else if let Some(rest) = url.strip_prefix("http://") {
(80, rest)
} else {
// The scp-like `git@host:ns/repo.git`. `classify` has already decided
// this shape is ssh; the colon here separates a PATH, never a port.
// The userinfo goes first so a `@` inside it cannot be read as a host,
// and a bracketed literal is taken whole — `[::1]:path` is all colons
// and splitting on the first one would yield `[`.
let rest = url.rsplit('@').next().unwrap_or_default();
let host = match rest.strip_prefix('[') {
Some(inside) => inside.split(']').next().unwrap_or_default(),
None => rest.split(':').next().unwrap_or_default(),
};
return non_empty(host, url).map(|host| (host, 22));
};
let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
let authority = authority.rsplit('@').next().unwrap_or_default();
// `[::1]:9418` and `[::1]`, before the bare-host split, because an IPv6
// literal is all colons.
if let Some(rest) = authority.strip_prefix('[') {
let Some((host, tail)) = rest.split_once(']') else {
return Err(format!("{} names an unterminated IPv6 literal", scrub(url)));
};
let port = match tail.strip_prefix(':') {
None => default_port,
Some(digits) => digits
.parse()
.map_err(|_| format!("{} names a port that is not a number", scrub(url)))?,
};
return non_empty(host, url).map(|host| (host, port));
}
match authority.split_once(':') {
None => non_empty(authority, url).map(|host| (host, default_port)),
Some((host, digits)) => {
let port = digits
.parse()
.map_err(|_| format!("{} names a port that is not a number", scrub(url)))?;
non_empty(host, url).map(|host| (host, port))
}
}
}
fn non_empty(host: &str, url: &str) -> Result<String, String> {
if host.is_empty() {
return Err(format!("{} names no host to dial", scrub(url)));
}
Ok(host.to_owned())
}
/// **Refuse a destination this server can reach and its caller cannot** — the
/// pure half, with the addresses handed in.
///
/// `Mirror.Add` needs Admin and `Mirror.Sync` needs Write, but *every enrolled
/// principal is intrinsically Admin in its own namespace*
/// (`AuthTable::owns_account`), so `Mirror.Add{create_repo}` against
/// `alice/anything` is available to anybody with a key. Before this check the
/// only thing standing between that caller and an arbitrary request from the
/// server's own network position was `remote_url.trim().is_empty()`: the cloud
/// metadata endpoint at `169.254.169.254`, a control plane on `127.0.0.1`, a
/// printer on `10.0.0.0/8` — all dialled by the appliance, with the appliance's
/// routes, and the outcome partly readable back through `Mirror.Status`.
///
/// **Every** resolved address has to be routable, not merely the first: a name
/// that resolves to one public address and one loopback address is the
/// rebinding shape, and taking the best of the two is how a check like this
/// gets to be decorative.
///
/// The ranges are written out rather than taken from `IpAddr::is_global`, which
/// is unstable, and each line is the reason it is there.
pub(crate) fn address_verdict(host: &str, addrs: &[std::net::IpAddr]) -> Result<(), String> {
if addrs.is_empty() {
return Err(format!("{host} resolves to no address at all"));
}
for addr in addrs {
if let Some(why) = not_routable(*addr) {
return Err(format!(
"{host} resolves to {addr}, which is {why}. A mirror dials from the server, so a \
destination the caller could not reach on their own is the server's network \
position being borrowed. Create `{MIRROR_ALLOW_PRIVATE}` under this box's data \
root to allow it deliberately."
));
}
}
Ok(())
}
/// Why `addr` is not somewhere an arbitrary caller should be able to point this
/// server, or `None` when it is an ordinary public address.
fn not_routable(addr: std::net::IpAddr) -> Option<&'static str> {
use std::net::IpAddr;
match addr {
IpAddr::V4(v4) => {
let [a, b, c, _] = v4.octets();
match () {
_ if v4.is_unspecified() => Some("the unspecified address"),
_ if v4.is_loopback() => Some("loopback — this server itself"),
_ if v4.is_private() => Some("a private network (RFC 1918)"),
_ if v4.is_link_local() => {
Some("link-local (RFC 3927), the range cloud metadata services answer on")
}
_ if v4.is_broadcast() => Some("the broadcast address"),
_ if v4.is_multicast() => Some("multicast"),
_ if v4.is_documentation() => Some("a documentation range (RFC 5737)"),
// 100.64.0.0/10, carrier-grade NAT: the provider's own network,
// not the caller's.
_ if a == 100 && (64..128).contains(&b) => Some("carrier-grade NAT (RFC 6598)"),
// 192.0.0.0/24, IETF protocol assignments.
_ if a == 192 && b == 0 && c == 0 => Some("an IETF protocol assignment"),
// 198.18.0.0/15, benchmarking.
_ if a == 198 && (b == 18 || b == 19) => Some("a benchmarking range (RFC 2544)"),
// 240.0.0.0/4, reserved.
_ if a >= 240 => Some("a reserved range"),
_ => None,
}
}
// A v4 address written the long way is the same address. Checked
// first, or `::ffff:127.0.0.1` walks straight past every arm below it.
IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => {
not_routable(IpAddr::V4(v6.to_ipv4_mapped().expect("just matched")))
}
IpAddr::V6(v6) => {
let segments = v6.segments();
match () {
_ if v6.is_unspecified() => Some("the unspecified address"),
_ if v6.is_loopback() => Some("loopback — this server itself"),
_ if v6.is_multicast() => Some("multicast"),
// fc00::/7, unique local.
_ if segments[0] & 0xfe00 == 0xfc00 => Some("a unique-local address (RFC 4193)"),
// fe80::/10, link-local.
_ if segments[0] & 0xffc0 == 0xfe80 => Some("link-local (RFC 4291)"),
// 2001:db8::/32, documentation.
_ if segments[0] == 0x2001 && segments[1] == 0x0db8 => {
Some("a documentation range (RFC 3849)")
}
_ => None,
}
}
}
}
/// [`address_verdict`] with the resolution and the opt-out file read — the
/// edge, and the only part of this that touches the world.
///
/// # The window this does not close, said rather than glossed
///
/// The name is resolved here and again by `ureq` or by the ssh dialler a moment
/// later, so a resolver that answers differently the second time (DNS
/// rebinding) is not stopped by this. Closing that needs the dial pinned to the
/// address that was checked, which is a change inside `ureq`'s and
/// `nornir-git`'s connect paths rather than in this file. What this does close
/// is the whole of the *static* case — a URL that simply names an internal
/// address — which is what `Mirror.Add` accepted unchecked.
pub fn check_destination(url: &str, data_root: &std::path::Path) -> Result<(), String> {
use std::net::ToSocketAddrs as _;
if data_root.join(MIRROR_ALLOW_PRIVATE).exists() {
return Ok(());
}
let (host, port) = authority(url)?;
// A literal needs no resolver, and asking one for it would let a DNS server
// answer a question that has no question in it.
if let Ok(addr) = host.parse::<std::net::IpAddr>() {
return address_verdict(&host, &[addr]);
}
let addrs: Vec<std::net::IpAddr> = (host.as_str(), port)
.to_socket_addrs()
.map_err(|err| format!("{host} could not be resolved: {err}"))?
.map(|socket| socket.ip())
.collect();
address_verdict(&host, &addrs)
}
/// **Refuse a credential that cannot be offered over the transport the URL
/// names — by name, before anything is dialled.**
///
/// The failure this prevents is the quiet one. An SSH key against an `https://`
/// remote is not "authentication failed": there is nowhere in an HTTP request
/// to put a private key, so the request goes out anonymous and the upstream
/// answers 401 — and the operator, who can see that a credential IS stored,
/// concludes the key is wrong and rotates it. Forever. The same in reverse: a
/// token has no place in an SSH publickey exchange, so the dial falls through
/// to whatever else is offerable and fails on authentication with no mention of
/// the token at all.
///
/// A stored credential of no kind is not a mismatch — a public upstream mirrors
/// fine with nothing, and that path must keep working.
pub fn check_credential(wire: Wire, credential: &Credential) -> Result<(), String> {
match (wire, credential) {
(_, Credential::None) => Ok(()),
(Wire::Ssh, Credential::SshKey(_)) => Ok(()),
(Wire::Http, Credential::Token { .. }) => Ok(()),
(wire, held) => Err(format!(
"this mirror's remote is dialled over {}, and the credential stored for it is {}. \
{} cannot carry it, so the fetch would go out unauthenticated and be refused as \
though no credential were set. Store {} instead, with `Mirror.SetCredential`.",
wire.name(),
held.kind().name(),
wire.name(),
match wire {
Wire::Ssh => "an ssh private key",
Wire::Http => "a token",
}
)),
}
}
// ── the gate ─────────────────────────────────────────────────────────────────
/// **The repository's branch protection, over the batch a mirror is about to
/// apply.**
///
/// One implementation: this is `gunnar_policy::BranchProtectionGate`, the same
/// object `receive_pack`'s [`crate::transport::PolicyGate`] evaluates, reading
/// the same document through the same [`crate::transport::load_policy`]. A
/// second copy of "what does `allow_force_push` mean" is precisely the pair
/// that drifts, and the three independent ways past branch protection this
/// repository has had are all the same shape: a second writer that does what
/// the gated writer does, without the gate.
///
/// # What a mirror can decide, and the one thing it cannot
///
/// `allow_create`, `allow_delete` and `allow_force_push` are decidable from the
/// batch and the objects, which are already in the store when [`apply`] runs.
/// `require_signature` is decidable too — but only against the set of objects
/// the fetch introduced, so a repository carrying any protection rule makes the
/// ingest ask for that set ([`crate::storage::ObjectSet::Required`]) instead of
/// paying nothing for it, exactly as a push does when a gate is live.
///
/// `allowed_principals` is the one that cannot be decided here, and it is
/// refused rather than skipped. The caller's identity does not cross the
/// control-plane seam on this verb — `mirror_sync(&self, path)` carries a
/// repository and nothing about who asked — so there is no principal to compare
/// against the list. "I could not tell" is not "it is fine", so a rule that
/// names its principals refuses the whole sync by name, and the operator either
/// drops the rule or stops mirroring onto the refs it covers.
struct Protection {
gate: gunnar_policy::BranchProtectionGate,
keys: gunnar_policy::StaticKeyDirectory,
repo: gunnar_policy::RepoId,
}
impl Protection {
/// The rules covering `repo`, or `None` when it has none configured.
///
/// Read **before** the fetch, because whether any rule exists is what
/// decides if the ingest pays for an object-set enumeration — the same
/// ordering, for the same reason, as `receive_pack`'s.
///
/// `Err` is a policy document that could not be read, and it REFUSES the
/// sync — the same fail-closed rule the push path has (GUN-007). A mirror
/// that synced with its gates disabled because the policy did not parse
/// would be the bypass GUN-032 closed, back through the side door.
fn open(
stores: &ServerStores,
row: &gunnar_core::Repo,
git_dir: &std::path::Path,
) -> std::result::Result<Option<Self>, String> {
let config = crate::transport::load_policy(
stores,
row,
&git_dir.join(crate::transport::POLICY_FILE),
)
.map_err(|why| format!("refusing to mirror into {}: {why}", row.key()))?;
if config.protection.is_empty() {
return Ok(None);
}
Ok(Some(Protection {
gate: gunnar_policy::BranchProtectionGate::new(config.protection.clone()),
keys: crate::transport::key_directory(stores),
repo: gunnar_policy::RepoId::new(&row.namespace, &row.name),
}))
}
/// **Refuse the batch, or let it through.** `Err` is the refusal a FAILED
/// job carries, and nothing has been written when it is returned.
///
/// `new_objects` is the fetch's own receipt. An ingest that did not
/// enumerate hands back the reason rather than an empty set, and an empty
/// set here would be a silent always-allow for `require_signature` — so the
/// `Err` arm refuses, quoting why.
fn check(
&self,
git: &dyn gunnar_store::git_storage_trait::GitServe,
updates: &[gunnar_policy::RefUpdate],
new_objects: std::result::Result<&[ObjectId], &str>,
) -> Result<(), String> {
use gunnar_policy::Gate as _;
// The field with no answer on this path, refused first and by name.
for update in updates {
for rule in self.gate.rules() {
if !rule.pattern.matches_ref(&update.name) {
continue;
}
if rule.allowed_principals.is_some() {
return Err(format!(
"{} is protected by {}, which names the principals that may move it. A \
mirror sync carries no principal — the control-plane verb takes a \
repository and not a caller — so that rule cannot be satisfied and this \
sync is refused rather than applied unchecked. Nothing was moved.",
update.name, rule.pattern
));
}
break;
}
}
let new_objects = new_objects.map_err(|reason| {
format!(
"this repository has branch protection and the fetched pack's object set could \
not be enumerated ({reason}), so the gate could not decide. Nothing was moved."
)
})?;
// Never consulted: every rule that could read it has already refused
// above. Named rather than left as a plausible-looking value.
let principal = gunnar_policy::Principal::machine("mirror");
let ctx = gunnar_policy::PushContext::new(
&principal,
&self.repo,
updates,
new_objects,
git,
&self.keys,
);
match self.gate.check(&ctx) {
Ok(gunnar_policy::Verdict::Allow) => Ok(()),
Ok(gunnar_policy::Verdict::Reject(rejection)) => Err(format!(
"branch protection refused this mirror's reference batch: {}. Nothing was moved.",
rejection.reason
)),
// A gate that could not decide refuses; it does not allow.
Err(err) => Err(format!(
"branch protection could not decide this mirror's reference batch: {err}. \
Nothing was moved."
)),
}
}
}
// ── the local half ───────────────────────────────────────────────────────────
/// **What this repository already holds**, as `fetch_pack`'s have-source.
///
/// One `GitOps::refs()` call, one snapshot, taken inside `FetchPack::fetch` so
/// the `have` list and every `old` in the plan come from the same read — a
/// reference that moves under the fetch cannot make the plan promise an old
/// value the store never held.
///
/// `HEAD` and the pseudo-refs are not rows in `refs()` at all, and a purely
/// symbolic row under `refs/` carries no oid and is skipped: it is not a thing
/// to offer as a `have`, and `fetch_pack`'s own doc says an unborn ref should
/// simply be absent.
struct StoreRefs<'a> {
git: &'a dyn gunnar_store::git_storage_trait::GitServe,
/// Objects this store holds that no reference names yet — an interrupted
/// import's ([`already_here`]). Offered as `have`s under names the plan
/// never matches, so they change what is sent and nothing that is written.
extra_haves: Vec<ObjectId>,
}
impl LocalRefs for StoreRefs<'_> {
fn tips(&self) -> gunnar_wire::Result<Vec<(String, ObjectId)>> {
let rows = GitOps::refs(self.git).map_err(|err| {
gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
"reading this repository's refs to offer them as haves: {err:#}"
)))
})?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
if row.name == "HEAD" {
continue;
}
let Some(raw) = row.oid.as_deref() else {
continue;
};
let Ok(oid) = gix_hash::oid::try_from_bytes(raw) else {
return Err(gunnar_wire::Error::Store(gunnar_store::Error::backend(
format!("{} holds an oid this store cannot decode", row.name),
)));
};
out.push((row.name, oid.to_owned()));
}
for (n, oid) in self.extra_haves.iter().enumerate() {
out.push((format!("refs/gunnar/resume/{n}"), *oid));
}
Ok(out)
}
}
/// **The queue's progress field, written from the wire and from the walk.**
///
/// Three publishers, one job record, and each of the three is the only place on
/// the system that knows its own number:
///
/// | published by | what | when |
/// |---|---|---|
/// | [`FetchProgress::objects_total`] | the denominator | once, the instant the packfile header is read |
/// | [`FetchProgress::pack_bytes`] | the byte count | as the stream is consumed, while the network is the bottleneck |
/// | [`PackProgress::indexed`] | the numerator | once per object, as the engine's pack walk splits them |
///
/// # Why the numerator arrives from the other side of the seam
///
/// `gunnar-wire` reads bytes 8..12 of the header and can publish `(0, N)`
/// before an object is stored — twelve bytes, and it is the reason the
/// denominator is live at all. It cannot publish the numerator: counting
/// objects as they pass means inflating every one of them, which is a second
/// pack decoder, and gunnar has exactly one (one implementation, and `fetch_pack`'s own
/// module doc refuses it in those words). The one decoder is the storage
/// engine's walk, which has to split every entry anyway, so the count is free
/// exactly there and nowhere else. This type is where the two halves meet.
///
/// # `total` is remembered here, and that is not a second counter
///
/// [`PackProgress::indexed`] carries its own `total` — the count the engine
/// read from the same header — and it is *checked* against the one the wire
/// announced rather than trusted, because two readings of one field that
/// disagree mean the ingest is being handed a different pack from the one that
/// was negotiated. `fetch_pack` makes the same comparison at the end, on the
/// receipt; this is the same fact asserted while there is still a transfer to
/// abandon.
struct QueueProgress<'a> {
report: &'a Reporter<Mirror>,
total: AtomicU64,
/// A second sink, for a caller that has to see the *series* rather than the
/// job's latest reading. See [`sync_once_watching`].
watching: Option<&'a dyn git_storage_trait::PackProgress>,
}
impl QueueProgress<'_> {
/// **The one place a sync's progress pair is written.**
///
/// All three publishers go through here — the header's `(0, N)`, the
/// engine's per-object readings, and [`finish`]'s `(N, N)` — so the job and
/// any watcher see the same values in the same order. Two routes to the
/// same field would be two places the series could differ, and the series
/// is the whole subject.
fn publish(&self, done: u64, total: u64) {
self.report.progress(done, total);
if let Some(watcher) = self.watching {
watcher.indexed(done, total);
}
}
}
impl FetchProgress for QueueProgress<'_> {
fn objects_total(&self, objects: u32) {
self.total.store(u64::from(objects), Ordering::SeqCst);
self.publish(0, u64::from(objects));
}
fn pack_bytes(&self, streamed: u64) {
self.report.report(|mirror| mirror.bytes = streamed);
}
}
impl git_storage_trait::PackProgress for QueueProgress<'_> {
fn indexed(&self, done: u64, total: u64) {
// The wire's denominator wins. An engine that read a different total
// from the same twelve bytes is a bug worth a log line, not a reason to
// renumber a progress bar mid-flight and leave a console watching a
// percentage that jumped sideways.
let announced = self.total.load(Ordering::SeqCst);
let total = if announced == 0 || announced == total {
total
} else {
tracing::warn!(
target: TARGET_GIT,
announced,
indexed_total = total,
"the packfile header the wire read and the one the store read declare \
different object counts; the wire's is the one being rendered"
);
announced
};
self.publish(done, total);
}
}
// ── the fetch ────────────────────────────────────────────────────────────────
/// **One sync, start to finish.** The [`crate::mirror_queue::MirrorRunner`]
/// body.
///
/// Errors are `String` because they are going onto a job a console will read,
/// not up a call stack — the same reason `gc_queue`'s runner returns one.
pub fn sync_once(
stores: &Arc<ServerStores>,
order: &MirrorOrder,
credential: &Credential,
report: &Reporter<Mirror>,
) -> Result<(), String> {
sync_once_watching(stores, order, credential, report, None)
}
/// [`sync_once`], with a **second** place the object count is published to.
///
/// # Why this door exists, and why it is not a test hook
///
/// [`Reporter::progress`] *overwrites*, by design — its own doc says a progress
/// that accumulated would be a second counter to keep right — so the job record
/// holds the latest reading and never the series. That is correct for a console,
/// which only ever wants the latest reading, and it means the one thing that
/// cannot be established from the job record afterwards is **whether the count
/// moved at all**. A publisher that fires `(0, N)` and then `(N, N)` and a
/// publisher that fires five hundred times in between leave the job in exactly
/// the same state, so from the outside they are the same publisher.
///
/// That is measure-never-assert in its plainest form: a stub is
/// indistinguishable from an
/// implementation unless something can see it work — and it is precisely the
/// defect this seam was opened to close: `(0, N) → (N, N)` was the shape for as
/// long as no engine published from inside its pack walk. So the series gets a
/// door, and the acceptance test stands at it. There is one other caller,
/// [`sync_once`], which passes `None`.
///
/// The sink is *additional*: the job is written exactly as before, in the same
/// order, whether or not anybody is watching. A test that changed what the
/// production path publishes would be testing itself.
pub fn sync_once_watching(
stores: &Arc<ServerStores>,
order: &MirrorOrder,
credential: &Credential,
report: &Reporter<Mirror>,
watching: Option<&dyn git_storage_trait::PackProgress>,
) -> Result<(), String> {
let wire = classify(&order.upstream)?;
check_credential(wire, credential)?;
// Checked again on the worker, and not only at the door where
// `Mirror.Add` checked it: a registration made before this rule existed,
// or a name whose A record has moved since, reaches the dial through here
// and nowhere else.
check_destination(&order.upstream, stores.root())?;
// Resolved here, on the worker, and not at enqueue time: a repository can
// be deleted while its sync waits, and discovering that when the work
// starts is a FAILED job saying so, which is the truth.
let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
format!(
"`{}` is not a valid <namespace>/<name>: {err:#}",
order.path
)
})?;
let existing = stores
.catalog()
.get(&repo_ref)
.map_err(|err| format!("reading the catalog row for {}: {err:#}", order.path))?;
// **An import's repository is made HERE, by the first sync that reaches
// the upstream** (owner order 2026-10-18; `MirrorRecord::creates_repo`).
let to_create = existing.is_none()
&& stores
.mirrors()
.get(&repo_ref)
.map_err(|err| format!("reading the registration for {}: {err:#}", order.path))?
.is_some_and(|record| record.creates_repo);
if existing.is_none() && !to_create {
return Err(format!("{} is no longer a live repository", order.path));
}
// **An encrypted repository is never fetched into** (wiring tracer B
// 2026-10-08): the ingest below runs no encrypted gate (`PushIngest::new`),
// so an upstream's plaintext would be stored and framed for the twin.
// `Mirror.Add` / `Repos.SetEncrypted` refuse making such a mirror; this is
// a registration older than that rule, or a flag set below the door.
if existing.as_ref().is_some_and(|row| row.encrypted) {
return Err(format!(
"{} is encrypted, so it is not fetched into: a mirror copies another server's \
plaintext, and an encrypted repository holds only what its members sealed \
(remove the mirror)",
order.path
));
}
// **The namespace's room, asked of the same gates a push is held to**
// (2026-10-02). A suspended namespace or one already at its cap fetches
// nothing — refused here, before the dial, in the gate's own words — and
// the fetch is held to the tightest ceiling in the chain as it streams.
let ceiling = room_for(&stores.plan_gates(), repo_ref.namespace()).map_err(|(why, line)| {
report.report(|r| r.refused = why);
line
})?;
let auth = match credential {
Credential::Token { username, secret } if username.is_empty() => {
Some(format!("Bearer {secret}"))
}
Credential::Token { username, secret } => {
Some(format!("Basic {}", base64_basic(username, secret)))
}
_ => None,
};
let row = match existing {
Some(row) => row,
None => create_from_upstream(stores, &repo_ref, order, wire, credential, auth.clone())?,
};
let fetched = fetch_into(
stores, &row, order, wire, credential, auth, ceiling, report, watching,
)
.and_then(|advertised| landed(stores, &row, order, &advertised).map(|()| advertised));
match &fetched {
Ok(advertised) => after_landing(stores, &repo_ref, &row, order, advertised),
Err(_) if to_create => discard_created(stores, &repo_ref, order),
Err(_) => {}
}
fetched.map(|_| ())
}
/// **A sync is DONE only when what the upstream advertised is here**
/// (import-big, 2026-10-26).
///
/// Every branch and tag the upstream advertised must be a reference in this
/// repository when the sync says it is done, and an upstream that advertised
/// none is not an import that worked: it is an empty import, said as a
/// failure. Read back from the store, after the apply, rather than inferred
/// from the plan — the plan is what was meant, this is what landed. Before it,
/// a fetch whose plan came out empty applied nothing and settled DONE over a
/// repository with no references at all.
fn landed(
stores: &Arc<ServerStores>,
row: &gunnar_core::model::Repo,
order: &MirrorOrder,
advertised: &[gunnar_wire::upload_pack::AdvertisedRef],
) -> Result<(), String> {
let wanted: Vec<&str> = advertised
.iter()
.filter(|r| fetch_pack::mirror(&r.name) && r.target.is_some())
.map(|r| r.name.as_str())
.collect();
let upstream = scrub(&order.upstream);
if wanted.is_empty() {
return Err(format!(
"{upstream} has no branches or tags, so there was nothing to import into {}. \
Check the address, or push a first commit there and sync again.",
order.path
));
}
let store = stores
.repo_store(row)
.map_err(|err| format!("opening {} to check what landed: {err:#}", order.path))?;
let held = GitOps::refs(store.serve()).map_err(|err| {
format!(
"reading {}'s references to check what landed: {err:#}",
order.path
)
})?;
let missing: Vec<&str> = wanted
.iter()
.copied()
.filter(|name| !held.iter().any(|r| r.name == *name && r.oid.is_some()))
.collect();
if let Some(first) = missing.first() {
return Err(format!(
"the fetch from {upstream} ended without its references landing: {} of {} \
(first {first}) are not in {}. Nothing is reported as imported that is not \
here; sync again.",
missing.len(),
wanted.len(),
order.path
));
}
Ok(())
}
/// **Ask the upstream what it is, then make the repository to match.**
///
/// One `ls-refs` at the upstream's own object format (`ls_remote_any`): the
/// format becomes the repository's, and `HEAD`'s `symref-target` its default
/// branch. Nothing is created when the upstream cannot be reached or refuses
/// (a private repository without its token), so a failed first import leaves
/// no repository behind.
fn create_from_upstream(
stores: &Arc<ServerStores>,
repo_ref: &gunnar_core::RepoRef,
order: &MirrorOrder,
wire: Wire,
credential: &Credential,
auth: Option<String>,
) -> Result<gunnar_core::model::Repo, String> {
let advertisement = ls_remote(stores, order, wire, credential, auth)?;
let hash = match advertisement.hash_kind {
gix_hash::Kind::Sha1 => gunnar_core::HashKind::Sha1,
_ => gunnar_core::HashKind::Sha256,
};
let mut new = gunnar_core::NewRepo::new(repo_ref.clone())
.hash_kind(hash)
.visibility(gunnar_core::Visibility::Private);
new.description = format!("mirror of {}", scrub(&order.upstream));
if let Some(branch) = advertised_head(&advertisement.refs) {
new = new.default_branch(branch);
}
tracing::info!(
target: TARGET_GIT,
event = "mirror.create_from_upstream",
repo = %order.path,
hash = ?hash,
"an import's repository is made at the upstream's object format and default branch"
);
let row = stores
.create_repo(new, false)
.map_err(|err| format!("creating {} for the import: {err:#}", order.path))?;
stores
.repo_store(&row)
.map_err(|err| format!("opening {} to mirror into it: {err:#}", order.path))?;
Ok(row)
}
/// **One `ls-refs` against the upstream**, at its own object format.
fn ls_remote(
stores: &Arc<ServerStores>,
order: &MirrorOrder,
wire: Wire,
credential: &Credential,
auth: Option<String>,
) -> Result<gunnar_wire::fetch_pack::Advertisement, String> {
let listed =
|err: gunnar_wire::Error| format!("asking {} what it holds: {err}", scrub(&order.upstream));
match wire {
Wire::Http => {
let (mut reader, mut writer) = open_http(&order.upstream, &order.ca, auth)?;
gunnar_wire::ls_remote_any(&mut reader, &mut writer).map_err(listed)
}
Wire::Ssh => dial_ssh(
&order.upstream,
credential,
&stores.root().join(MIRROR_KNOWN_HOSTS),
strict_host_keys(stores.root()),
|reader, writer| gunnar_wire::ls_remote_any(reader, writer),
)?
.map_err(listed),
}
}
/// **What an interrupted import already brought in**, offered back to the
/// upstream as `have`s (import-big, 2026-10-26).
///
/// A repository that holds no branch and no tag is either new or an import
/// that stopped between its pack landing and its references being written —
/// a restart in the index, the longest step of a big import. The pack's bytes
/// were durable before that step, and the indexer resumes them on its own after
/// the restart; what was missing is only the reference write. So the resumed
/// sync asks the upstream what it advertises, lets the index catch up, and
/// offers every advertised tip this store already has as a `have`: the
/// upstream then sends nothing it sent before, and a 6.6 GB kernel pack is not
/// fetched (or stored) a second time. A new repository has nothing, and pays
/// one `ls-refs` for finding out.
#[allow(clippy::too_many_arguments)]
fn already_here(
stores: &Arc<ServerStores>,
git: &dyn gunnar_store::git_storage_trait::GitServe,
git_dir: &std::path::Path,
order: &MirrorOrder,
wire: Wire,
credential: &Credential,
auth: Option<String>,
report: &Reporter<Mirror>,
) -> Result<Vec<ObjectId>, String> {
let held =
GitOps::refs(git).map_err(|err| format!("reading {}'s references: {err:#}", order.path))?;
if held
.iter()
.any(|r| fetch_pack::mirror(&r.name) && r.oid.is_some())
{
return Ok(Vec::new());
}
let advertisement = ls_remote(stores, order, wire, credential, auth)?;
// An interrupted import's pack is indexed here: held to a cgroup's memory as the
// first try was (IMPORT-2); its object count is not known again, so no refusal.
let held = crate::import_room::MemoryLimit::read()
.and_then(|l| CgroupHold::start(stores, git_dir, &l));
index_landed(git, git_dir, stores.engine(), report, held.as_ref()).map_err(|err| {
format!(
"{} holds objects this server could not index: {err}",
order.path
)
})?;
let mut out = Vec::new();
for advertised in &advertisement.refs {
let Some(target) = advertised.target else {
continue;
};
if !fetch_pack::mirror(&advertised.name) || out.contains(&target) {
continue;
}
if GitOps::has(git, target.as_bytes()).unwrap_or(false) {
out.push(target);
}
}
if !out.is_empty() {
tracing::info!(
target: TARGET_GIT,
event = "mirror.sync.resumed",
repo = %order.path,
haves = out.len(),
"an import that stopped after its pack landed resumes without fetching it again"
);
}
Ok(out)
}
/// `HEAD`'s branch, as the upstream advertised it, when it is a branch.
fn advertised_head(refs: &[gunnar_wire::upload_pack::AdvertisedRef]) -> Option<String> {
refs.iter()
.find(|r| r.name == "HEAD")
.and_then(|r| r.symref_target.clone())
.filter(|target| {
target.starts_with("refs/heads/") && gunnar_store::refname::validate(target).is_ok()
})
}
/// A first import that failed after its repository was made: the repository
/// goes again, so nothing empty is left behind. Best effort — a removal that
/// fails is logged, and the registration still says the repository is to come.
fn discard_created(
stores: &Arc<ServerStores>,
repo_ref: &gunnar_core::RepoRef,
order: &MirrorOrder,
) {
let removed = stores
.catalog()
.delete(repo_ref)
.map_err(|err| format!("{err:#}"))
.and_then(|_| {
stores
.purge_repo(repo_ref)
.map_err(|err| format!("{err:#}"))
});
if let Err(err) = removed {
tracing::warn!(
target: TARGET_GIT,
repo = %order.path,
%err,
"a failed first import could not remove the repository it had made"
);
}
}
/// **What a landed sync owes the mirror besides its references**: the
/// upstream's default branch (when it holds that branch now) and whether the
/// upstream uses Git LFS. Neither fails the sync — the references landed.
fn after_landing(
stores: &Arc<ServerStores>,
repo_ref: &gunnar_core::RepoRef,
row: &gunnar_core::model::Repo,
order: &MirrorOrder,
advertised: &[gunnar_wire::upload_pack::AdvertisedRef],
) {
let mut row = row.clone();
if let Some(branch) = advertised_head(advertised) {
let held = stores
.repo_store(&row)
.ok()
.and_then(|store| GitOps::refs(store.serve()).ok())
.is_some_and(|refs| refs.iter().any(|r| r.name == branch && r.oid.is_some()));
if held && branch != row.default_branch {
match stores.catalog().set_default_branch(repo_ref, &branch) {
Ok(moved) => {
row = moved;
if let Err(err) = stores.repoint_head(&row) {
tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "pointing HEAD at the upstream's default branch");
}
}
Err(err) => {
tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "taking the upstream's default branch");
}
}
}
}
let lfs = uses_lfs(stores, &row);
if let Err(err) = stores.mirrors().set_lfs_detected(repo_ref, lfs) {
tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "recording whether the upstream uses LFS");
}
}
/// **Does the default branch route files through Git LFS?** A `filter=lfs`
/// in its root `.gitattributes` is the sign every LFS repository carries; the
/// files it names arrive as pointer files, and their content is not mirrored.
fn uses_lfs(stores: &Arc<ServerStores>, row: &gunnar_core::model::Repo) -> bool {
use gunnar_store::git_storage_trait::ObjType;
use gunnar_store::object;
let Ok(store) = stores.repo_store(row) else {
return false;
};
let git = store.serve();
let hash = store.hash_kind();
let Some(tip) = GitOps::refs(git)
.ok()
.and_then(|refs| refs.into_iter().find(|r| r.name == row.default_branch))
.and_then(|r| r.oid)
.and_then(|raw| ObjectId::try_from(&raw[..]).ok())
else {
return false;
};
let read = |id: &ObjectId, want: ObjType| match git.read(id.as_bytes()) {
Ok(Some((kind, bytes))) if kind == want => Some(bytes),
_ => None,
};
let Some(commit) =
read(&tip, ObjType::Commit).and_then(|bytes| object::commit(&bytes, hash).ok())
else {
return false;
};
let Some(tree) = read(&commit.tree, ObjType::Tree) else {
return false;
};
let Ok(Some(entry)) = object::tree_entry_named(&tree, hash, b".gitattributes") else {
return false;
};
read(&entry.oid, ObjType::Blob).is_some_and(|attributes| {
String::from_utf8_lossy(&attributes)
.lines()
.any(|line| !line.trim_start().starts_with('#') && line.contains("filter=lfs"))
})
}
/// **The fetch and the apply**, into a repository that exists. Answers what
/// the upstream advertised, for [`after_landing`].
#[allow(clippy::too_many_arguments)]
fn fetch_into(
stores: &Arc<ServerStores>,
row: &gunnar_core::model::Repo,
order: &MirrorOrder,
wire: Wire,
credential: &Credential,
auth: Option<String>,
ceiling: Option<gunnar_wire::PackCeiling>,
report: &Reporter<Mirror>,
watching: Option<&dyn git_storage_trait::PackProgress>,
) -> Result<Vec<gunnar_wire::upload_pack::AdvertisedRef>, String> {
let row = row.clone();
let store = stores
.repo_store(&row)
.map_err(|err| format!("opening {} to mirror into it: {err:#}", order.path))?;
let git = store.serve();
let hash_kind = store.hash_kind();
// Opened BEFORE the ingest, and that ordering is load-bearing for the same
// reason it is on the push path: whether this repository has a protection
// rule at all is what decides if the fetch pays for an object-set
// enumeration.
let protection = Protection::open(stores, &row, store.git_dir())?;
let extra_haves = already_here(
stores,
git,
store.git_dir(),
order,
wire,
credential,
auth.clone(),
report,
)?;
let local = StoreRefs { git, extra_haves };
let progress = QueueProgress {
report,
total: AtomicU64::new(0),
watching,
};
// The ingest a PUSH lands through, unchanged. A fetched pack and a pushed
// pack are the same bytes (`fetch_pack`'s module doc), so this is the same
// storage authority and not a second one.
//
// The object set is asked for exactly when something reads it. A repository
// with a protection rule has a gate that does (`require_signature` walks the
// commits this fetch introduced), so the enumeration is `Required` and an
// engine that cannot produce one must refuse rather than let the gate decide
// on nothing. With no rule configured nobody reads it, and `NotNeeded` keeps
// the fetch from waiting on index work no answer depends on.
//
// The last argument is the numerator's whole route: the ingest hands it to
// the engine, the engine hands it to the pack walk, and the walk publishes
// once per object into `progress` above — which is the same value the wire
// is publishing the denominator and the byte count into. One sink, three
// publishers, no second decoder anywhere on the path.
let ingest = crate::transport::PushIngest::new(
store.git_dir().to_path_buf(),
hash_kind,
stores.engine(),
if protection.is_some() {
crate::storage::ObjectSet::Required
} else {
crate::storage::ObjectSet::NotNeeded
},
Some(&progress),
// A fetched pack is a transaction exactly as a pushed one is: framed before it lands.
Some(crate::transport::PackFrame::of(stores, &row)),
)
// …and held to the data set's room the way a push is (cov-13).
.with_space(stores.space_ledger());
// **Heimdall first** (feature `heimdall`, `crate::heimdall_gate`): with the
// gate installed, the fetched pack is parsed in a Heimdall job and only a
// pack whose signed result checks out reaches the ingest above.
#[cfg(feature = "heimdall")]
let checked = stores
.pack_gate()
.map(|gate| gate.checked_as(&ingest, hash_kind));
#[cfg(feature = "heimdall")]
let ingest: &dyn gunnar_wire::PackIngest = match &checked {
Some(c) => c,
None => &ingest,
};
#[cfg(not(feature = "heimdall"))]
let ingest: &dyn gunnar_wire::PackIngest = &ingest;
let mut fetch = FetchPack::new(hash_kind, &fetch_pack::mirror, &local).with_progress(&progress);
if let Some(ceiling) = ceiling {
fetch = fetch.with_ceiling(ceiling);
}
let outcome = match wire {
Wire::Http => {
let (mut reader, mut writer) = open_http(&order.upstream, &order.ca, auth)?;
fetch.fetch(ingest, &mut reader, &mut writer)
}
Wire::Ssh => dial_ssh(
&order.upstream,
credential,
&stores.root().join(MIRROR_KNOWN_HOSTS),
strict_host_keys(stores.root()),
|reader, writer| fetch.fetch(ingest, reader, writer),
)?,
}
.map_err(|err| {
// Cut at the namespace's ceiling: the plan's refusal, as a field.
if matches!(err, gunnar_wire::Error::OverCeiling(_)) {
report.report(|r| r.refused = crate::mirror_queue::Refusal::PackCap);
}
format!(
"fetching {} from {}: {err}",
order.path,
scrub(&order.upstream)
)
})?;
// **Index the fetched pack ONCE, here, before a single read by oid**
// (import-big, 2026-10-26) — the line `receive_pack::apply` has carried
// since 2026-08-30 and this path never got.
//
// The znippy engine stores a pack's bytes before its oids exist and indexes
// them behind the ack. `apply` reads by oid (the branch-protection gate, and
// `put_refs_cas`, whose dangling-target check asks `has` for every new tip),
// so without this its first read fell into the engine's read-time fallback:
// a whole-pack resolve on this thread beside the indexer resolving the same
// bytes. MEASURED on oden 2026-10-26, torvalds/linux from GitHub (6.60 GB,
// 11 986 735 objects): the import landed only after the two resolvers had
// driven the process to a 19.4 GB peak — past a 16 GB Mini, which is where
// the owner's import of it died with nothing to say so.
//
// A refusal is the backend's own sentence (its memory watchdog gives up by
// name, with what was available and what was needed): the pack is stored
// and could not be indexed on this box, which is "this repository does not
// fit here", never a sync that waits for ever or reads around it.
if outcome.receipt.is_some() {
// **IMPORT-2: the index must fit the memory this server may really use**: the box's
// MemTotal, or the cgroup's memory.max when that is smaller (a container, a
// MemoryMax= unit, where /proc/meminfo is the host's). Refused by name before
// the index when it does not fit with the margin; under a cgroup the store's
// memory guard reads the cgroup's memory, kept fresh while the index runs.
let objects = progress.total.load(Ordering::SeqCst);
let limit = crate::import_room::MemoryLimit::read();
if let Some(limit) = &limit {
crate::import_room::import_room(&order.path, objects, limit)?;
}
let held = limit
.as_ref()
.and_then(|l| CgroupHold::start(stores, store.git_dir(), l));
index_landed(git, store.git_dir(), stores.engine(), report, held.as_ref()).map_err(
|err| {
format!(
"{} was fetched and stored, and this server could not index it: {err}. Nothing \
was moved. The repository is larger than this server can take in; a server \
with more memory imports it.",
order.path
)
},
)?;
}
apply(git, order, &outcome, &progress, protection.as_ref())?;
Ok(outcome.advertised)
}
/// **`GitServe::index_now`, with the index's own progress on the job.**
///
/// For a kernel-sized pack the index is minutes after the last byte arrived,
/// and a job that sat at `N/N` for all of them would read as stalled. So while
/// it runs, the engine's index row for this repository (`Engine::index_progress`,
/// the same row the dashboard reads) is published as the job's progress: the
/// objects indexed of the pack's total. When it returns the pair is the walk's
/// again (`finish`).
fn index_landed(
git: &dyn gunnar_store::git_storage_trait::GitServe,
git_dir: &std::path::Path,
engine: crate::storage::Engine,
report: &Reporter<Mirror>,
held: Option<&CgroupHold>,
) -> Result<usize, String> {
let indexed = indexed_with_progress(
|| git.index_now().map_err(|err| format!("{err:#}")),
|| {
if let Some(held) = held {
held.refresh();
}
if let Some((_, row)) = engine
.index_progress()
.into_iter()
.find(|(dir, row)| dir == git_dir && row.active && row.objects_total > 0)
{
report.progress(row.objects_done, row.objects_total);
}
},
);
if let Ok(packs) = &indexed {
tracing::info!(
target: TARGET_GIT,
event = "mirror.sync.indexed",
packs,
"the fetched pack was indexed once, before the references were checked"
);
}
indexed
}
/// **The cgroup's memory, held under one import's index** (IMPORT-2).
///
/// Only when the limit is a cgroup's: the store's memory guard is pointed at a
/// meminfo-shaped file of the cgroup's memory (`import_room::cgroup_meminfo`), refreshed
/// every half second by the index's progress poller, with the import's margin as the
/// floor. Dropped after the index: the file goes and the store gets the process-wide
/// guard back, so a later push into the same repository reads no stale file.
struct CgroupHold {
engine: crate::storage::Engine,
git_dir: std::path::PathBuf,
cgroup: std::path::PathBuf,
file: std::path::PathBuf,
}
impl CgroupHold {
fn start(
stores: &Arc<ServerStores>,
git_dir: &std::path::Path,
limit: &crate::import_room::MemoryLimit,
) -> Option<CgroupHold> {
let cgroup = limit.cgroup_dir()?.to_path_buf();
let dir = stores.root().join("import-memory");
std::fs::create_dir_all(&dir).ok()?;
let name: String = git_dir
.to_string_lossy()
.chars()
.map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
.collect();
let held = CgroupHold {
engine: stores.engine(),
git_dir: git_dir.to_path_buf(),
cgroup,
file: dir.join(format!("{name}.meminfo")),
};
if !held.refresh() {
return None;
}
let floor = limit.bytes / 100 * crate::import_room::MARGIN_PERCENT;
if !held
.engine
.hold_index_memory(&held.git_dir, Some((&held.file, floor)))
{
let _ = std::fs::remove_file(&held.file);
return None;
}
tracing::info!(
target: TARGET_GIT,
event = "mirror.import.memory_held",
repo = %held.git_dir.display(),
limit = limit.bytes,
floor,
"the import's index reads this server's cgroup memory ({}) and pauses under {}",
limit.words(),
crate::budget::human(floor)
);
Some(held)
}
/// Write the cgroup's memory now (a temp file and a rename, so the guard never
/// reads half a file).
fn refresh(&self) -> bool {
let Some(c) = crate::import_room::cgroup_memory_in(&self.cgroup) else {
return false;
};
let tmp = self.file.with_extension("meminfo.tmp");
std::fs::write(&tmp, crate::import_room::cgroup_meminfo(&c)).is_ok()
&& std::fs::rename(&tmp, &self.file).is_ok()
}
}
impl Drop for CgroupHold {
fn drop(&mut self) {
self.engine.hold_index_memory(&self.git_dir, None);
let _ = std::fs::remove_file(&self.file);
}
}
/// **`index` on this thread while `progress` is asked every half second on a scoped one**
/// ([`index_landed`]). The poller stops when the index ends, however it ends: a panic in
/// the index is an `Err` naming it (RA4-9).
fn indexed_with_progress(
index: impl FnOnce() -> Result<usize, String>,
progress: impl Fn() + Sync,
) -> Result<usize, String> {
let done = std::sync::atomic::AtomicBool::new(false);
std::thread::scope(|scope| {
// A32-14, stays a scoped thread: it reads the index row for the length of one
// `index_now` on the mirror job's own lane and is joined before this returns.
scope.spawn(|| {
while !done.load(Ordering::SeqCst) {
progress();
for _ in 0..10 {
if done.load(Ordering::SeqCst) {
return;
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
}
});
// RA4-9: a panic in the index is caught here, so `done` is set however it ends and the
// scope (which waits on the poller) returns; the mirror job then fails by name.
let indexed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(index));
done.store(true, Ordering::SeqCst);
indexed.unwrap_or_else(|panic| {
let words = panic
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| panic.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "a panic with no words".into());
Err(format!("the index panicked: {words}"))
})
})
}
/// **How much a mirror sync into `namespace` may fetch, decided by the front
/// doors' own plan gates** ([`ServerStores::install_plan_gates`]).
///
/// The push path's questions, in its order (`gunnar_ssh::server`'s
/// `ask_plan`): the gate's verdict on a `Write`, then — for a write that grows
/// data, which a fetch is — `pack_quota` (is the namespace already at its
/// cap?) and the TIGHTEST `pack_ceiling` in the chain. The first `Refuse` is
/// the answer, verbatim. The ceiling is bound to the namespace's shared pack
/// ledger, marked here before any gate measured, so a mirror fetch and a push
/// landing at the same moment share the room instead of each spending it.
///
/// `Ok(None)` is "nothing caps it": a chain with no gate that answers, which is
/// the community appliance's ordinary state.
///
/// A refusal comes back with WHICH question refused it — the plan's verdict
/// on a write ([`crate::mirror_queue::Refusal::Plan`]) or the pack room
/// ([`crate::mirror_queue::Refusal::PackCap`]) — so the job can say so as a
/// field (`MirrorJob.refused_reason`).
pub(crate) fn room_for(
gates: &[Arc<dyn gunnar_ssh::PlanGate>],
namespace: &str,
) -> Result<Option<gunnar_wire::PackCeiling>, (crate::mirror_queue::Refusal, String)> {
use gunnar_ssh::{PlanAction, PlanVerdict};
if gates.is_empty() {
return Ok(None);
}
// Taken BEFORE any gate measures, for the reason the push path gives:
// only one of the two roundings is a quota that leaks.
let mark = gunnar_lfs::AllowanceLedger::shared_pack().mark(namespace);
let mut tightest: Option<gunnar_ssh::PackCeiling> = None;
for gate in gates {
if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
return Err((crate::mirror_queue::Refusal::Plan, line));
}
if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
return Err((crate::mirror_queue::Refusal::PackCap, line));
}
if let Some(asked) = gate.pack_ceiling(namespace) {
if tightest
.as_ref()
.is_none_or(|held| asked.bytes < held.bytes)
{
tightest = Some(asked);
}
}
}
Ok(tightest.map(|ceiling| gunnar_wire::PackCeiling {
bytes: ceiling.bytes,
refusal: ceiling.refusal,
ledger: Some(
Arc::new(crate::transport::PackMark { mark }) as Arc<dyn gunnar_wire::PackLedger>
),
}))
}
/// **Apply the plan, in one transaction, against the values the fetch
/// reasoned about.**
///
/// See the module doc for why this is `put_refs_cas` and why a rejection is a
/// failure rather than a skip.
fn apply(
git: &dyn gunnar_store::git_storage_trait::GitServe,
order: &MirrorOrder,
outcome: &FetchReport,
progress: &QueueProgress<'_>,
protection: Option<&Protection>,
) -> Result<(), String> {
let zero = "0".repeat(outcome.hash_kind.len_in_hex());
// Owned first, because `RefCas` borrows its oids and the batch has to
// outlive every one of them.
// (name, old, new, peeled)
type OwnedUpdate = (String, Option<Vec<u8>>, Option<Vec<u8>>, Option<Vec<u8>>);
let mut owned: Vec<OwnedUpdate> = Vec::new();
let mut updates: Vec<(String, String, String)> = Vec::new();
for plan in &outcome.updates {
// **The name came off somebody else's wire, so it goes through the ref
// name rules before it is written.**
//
// `plan.name` is the string an upstream's `ls-refs` answered with,
// carried untouched from `fetch_pack`, and the only filter it passed on
// the way is `fetch_pack::mirror`'s prefix test — which admits
// `refs/heads/../../etc/passwd` exactly as readily as `refs/heads/main`.
// Receive-pack puts every client-supplied name through
// `gunnar_store::refname::validate` at the same point in its own path
// (`receive_pack.rs`'s `funny refname`), and a fetch is the same
// untrusted input arriving through a different door.
//
// It is checked HERE rather than relied on from the storage engine. The
// memory backend does validate in `put_refs_cas`, so a name like this
// would be refused there today — but that is a property of one engine,
// and "a rule one backend enforces by accident is not a rule" is the
// sentence `refname` itself is written under. The engines are swappable
// (`GUNNAR_ENGINE`), and this is the one caller that can guarantee the
// check ran.
//
// Only the names that came from the upstream. The prune half below
// reads its names out of this repository's OWN refs, and a bad name
// already in the store is one a prune should be able to delete rather
// than one it must refuse to touch.
gunnar_store::refname::validate(plan.name.as_bytes()).map_err(|err| {
format!(
"the upstream advertised a reference name gunnar will not write: {err}. Nothing \
was moved. A name is not a path, and this one would not have been a name."
)
})?;
// The peel the remote already sent, rather than a peel computed here.
// `ls-refs` was asked with `peel`, so an annotated tag arrived with the
// commit it resolves to; recomputing it would be an object read per tag
// for an answer already in hand (the tag re-peel, from the other side).
let peeled = outcome
.advertised
.iter()
.find(|advertised| advertised.name == plan.name)
.and_then(|advertised| advertised.peeled)
.map(|oid| oid.as_bytes().to_vec());
updates.push((
plan.name.clone(),
plan.old
.map(|old| old.to_string())
.unwrap_or_else(|| zero.clone()),
plan.new.to_string(),
));
owned.push((
plan.name.clone(),
plan.old.map(|old| old.as_bytes().to_vec()),
Some(plan.new.as_bytes().to_vec()),
peeled,
));
}
if order.prune {
// **What the upstream no longer has.** Only over the same namespace the
// ref plan covers (`refs/heads/*`, `refs/tags/*`): a prune that reached
// wider would delete this box's own `refs/gunnar/*` bookkeeping on the
// grounds that an upstream never advertised it.
let held = GitOps::refs(git)
.map_err(|err| format!("reading {}'s refs to prune them: {err:#}", order.path))?;
for row in held {
if !fetch_pack::mirror(&row.name) {
continue;
}
if outcome
.advertised
.iter()
.any(|advertised| advertised.name == row.name)
{
continue;
}
let Some(old) = row.oid else { continue };
updates.push((row.name.clone(), hex(&old), zero.clone()));
owned.push((row.name, Some(old), None, None));
}
}
if owned.is_empty() {
finish(outcome, progress, Vec::new());
return Ok(());
}
// **The gate, between the plan and the write.** Before this line the batch
// below was the one reference write on the box that no gate had seen; a
// repository with no protection rule configured has no `Protection` at all,
// so a mirror of an unprotected repository pays nothing for this.
if let Some(protection) = protection {
let gated: Vec<gunnar_policy::RefUpdate> = owned
.iter()
.map(|(name, old, new, _)| {
// The null-oid spelling never appears here: `owned` already
// carries absence as `None`, which is what `RefUpdate`'s three
// constructors take. `try_from_bytes` cannot fail on a value the
// store or the wire produced, and an oid this store cannot
// decode is the one `StoreRefs::tips` already refuses by name.
let decode = |raw: &Vec<u8>| {
gix_hash::oid::try_from_bytes(raw)
.map(gix_hash::oid::to_owned)
.map_err(|err| {
format!("{name} holds an oid this server cannot decode: {err}")
})
};
Ok(gunnar_policy::RefUpdate {
name: name.clone(),
old: old.as_ref().map(&decode).transpose()?,
new: new.as_ref().map(&decode).transpose()?,
})
})
.collect::<Result<_, String>>()?;
let new_objects = match outcome.receipt.as_ref() {
// Nothing was fetched, so nothing was introduced. That is a measured
// empty set and not an unasked question — see `FetchReport::receipt`.
None => Ok(Vec::new()),
Some(receipt) => receipt.new_objects.clone(),
};
protection.check(git, &gated, new_objects.as_deref().map_err(String::as_str))?;
}
let batch: Vec<RefCas<'_>> = owned
.iter()
.map(|(name, old, new, peeled)| RefCas {
name: name.clone(),
old: old.as_deref(),
new: new.as_deref(),
peeled: peeled.clone(),
})
.collect();
// ONE call. Every edit or none — a loop with a break leaves what it already
// applied applied, and a half-mirrored repository is a state no upstream
// ever had.
GitOps::put_refs_cas(git, &batch).map_err(|err| {
format!(
"the fetch succeeded and the reference update did not: {err:#}. Nothing was moved. \
The usual cause is a push that landed on {} while the fetch was running; the \
expected values came from the snapshot the fetch negotiated against, and \
overwriting a concurrent push is the one thing a mirror must never do. Sync again.",
order.path
)
})?;
tracing::info!(
target: TARGET_GIT,
event = "mirror.sync.applied",
repo = %order.path,
refs = updates.len(),
objects = outcome.objects_done,
bytes = outcome.pack_bytes,
"a mirror sync moved this repository's references"
);
finish(outcome, progress, updates);
Ok(())
}
/// **Close the job's numbers out**, on both paths out of [`apply`].
///
/// One function and not two copies, because the pair below is the whole of what
/// a console reads when a sync ends and the two paths differ only in whether
/// any reference moved.
///
/// The `progress` call is the second half of the pair `FetchProgress` opened.
/// `objects_total` published `(0, N)` the instant the packfile header arrived;
/// **this publishes `(N, N)`**, and without it a finished job sits for ever at
/// nought per cent of a total it knows — which is worse than no progress bar,
/// because it reads as a transfer that stalled. `fetch_pack` has already
/// refused the transfer if the declared count and the indexed count disagree,
/// so `objects_done` is the same `N` and this cannot report a completion the
/// store did not make.
///
/// A fetch that wanted nothing — the ordinary answer for a poll of an idle
/// upstream — has no header, no `N`, and therefore **no progress at all**: the
/// pair stays `None`, and a caller reads `has_progress = false` rather than a
/// fabricated `0/0`.
fn finish(
outcome: &FetchReport,
progress: &QueueProgress<'_>,
updates: Vec<(String, String, String)>,
) {
if outcome.receipt.is_some() {
let total = u64::from(outcome.objects_total);
progress.publish(u64::from(outcome.objects_done), total);
}
progress.report.report(|mirror| {
mirror.objects = u64::from(outcome.objects_done);
mirror.bytes = outcome.pack_bytes;
mirror.updates = updates;
});
}
/// Lower-case hex, the spelling `gunnar.v1.RefUpdate` uses.
fn hex(raw: &[u8]) -> String {
let mut out = String::with_capacity(raw.len() * 2);
for byte in raw {
use std::fmt::Write as _;
let _ = write!(out, "{byte:02x}");
}
out
}
/// `base64(user:secret)`, the Basic-auth pair.
///
/// Written out rather than pulled in: the whole alphabet is 64 characters and
/// the encoder is nine lines, against a dependency that would be in the
/// appliance's graph for this one call. It is not a general base64 — it does
/// exactly the one thing, and `mirror_http`'s tests assert on the bytes it
/// produces.
fn base64_basic(user: &str, secret: &str) -> String {
const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let raw = format!("{user}:{secret}");
let raw = raw.as_bytes();
let mut out = String::with_capacity(raw.len().div_ceil(3) * 4);
for chunk in raw.chunks(3) {
let b = [
chunk[0],
chunk.get(1).copied().unwrap_or(0),
chunk.get(2).copied().unwrap_or(0),
];
let packed = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
out.push(ALPHABET[(packed >> 18) as usize & 63] as char);
out.push(ALPHABET[(packed >> 12) as usize & 63] as char);
out.push(if chunk.len() > 1 {
ALPHABET[(packed >> 6) as usize & 63] as char
} else {
'='
});
out.push(if chunk.len() > 2 {
ALPHABET[packed as usize & 63] as char
} else {
'='
});
}
out
}
// ── the ssh half ─────────────────────────────────────────────────────────────
/// **Dial `git-upload-pack` over SSH and hand the channel's two halves to
/// `run`.**
///
/// `gunnar_client::connect_upload_pack` is the fetch-side twin of the
/// `connect_receive_pack` replication already pushes with, over
/// `nornir-git`'s russh transport. There is no second dialler here and there
/// must not be: that transport owns the connect, the publickey exchange, the
/// `known_hosts` policy, the blocking bridge and — load-bearing — the `Drop`
/// that tears the channel down *inside* the driving runtime.
///
/// # The key is a file for the length of one dial, and then it is not
///
/// `nornir_git::ssh::SshAuth` authenticates from a **path**. gunnar's mirror
/// credential is sealed bytes in the control database, because a private key
/// sitting in the clear on `/data` is the thing that store exists to prevent.
/// The two are bridged by a [`tempfile::NamedTempFile`] created `0600` and
/// dropped — and therefore unlinked — before this function returns, on every
/// path out of it including a panic.
///
/// The alternative was an in-memory key API on `nornir-git`, which is the right
/// long-term shape, and it is a change to a crate outside this repository. It
/// is named here rather than left as a silent compromise.
#[cfg(feature = "replication")]
fn dial_ssh<T>(
url: &str,
credential: &Credential,
known_hosts: &std::path::Path,
strict: bool,
run: impl FnOnce(&mut dyn BufRead, &mut dyn Write) -> T,
) -> Result<T, String> {
use gunnar_client::ssh::{HostKeyPolicy, SshOptions};
use std::io::BufReader;
let mut options = SshOptions::new();
// A server has no ssh-agent and no interactive user. Offering the agent
// would be offering identities nobody configured for this mirror — the
// silent substitution `gunnar_client::resolve_identity` exists to forbid.
options.use_agent = false;
// Held for the whole of `run`: dropping it unlinks the file, and the dial
// reads it during `connect_upload_pack`.
let key_file = match credential {
Credential::SshKey(bytes) => {
let mut file = tempfile::Builder::new()
.prefix("gunnar-mirror-key-")
.tempfile()
.map_err(|err| format!("no room to materialise the mirror's ssh key: {err}"))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
std::fs::set_permissions(file.path(), std::fs::Permissions::from_mode(0o600))
.map_err(|err| {
format!("the mirror's ssh key could not be made private: {err}")
})?;
}
file.write_all(bytes)
.and_then(|()| file.flush())
.map_err(|err| format!("writing the mirror's ssh key: {err}"))?;
options.key_path = Some(file.path().to_path_buf());
Some(file)
}
// No key named means the ladder, which on a server finds nothing; that
// is the honest behaviour for a public upstream over ssh and it is what
// `gunnar-client` warns about at start-up.
_ => None,
};
// Trust-on-first-use against a `known_hosts` under the data root is what
// replication does when an operator has not said otherwise, and a
// mirror is dialling the same kind of box. See [`MIRROR_KNOWN_HOSTS`] for
// why the file is NOT `~/.ssh/known_hosts`.
//
// **Unless the box says otherwise**, which it could not before: the policy
// was this literal and nothing else, so a first sync trusted whatever
// answered on the upstream's address, on a path that is about to offer it a
// private key. See [`MIRROR_STRICT_HOST_KEY`].
options.host_key = if strict {
HostKeyPolicy::StrictKnownHosts
} else {
HostKeyPolicy::TofuKnownHosts
};
options.known_hosts = Some(known_hosts.to_path_buf());
let location = nornir_git::ssh::parse_ssh_url(url)
.map_err(|err| format!("{} is not an ssh url: {err}", scrub(url)))?;
let mut channel = gunnar_client::connect_upload_pack(&location, &options).map_err(|err| {
// `{:#}` over an `anyhow::Error`, and not `{err}`: `gunnar_client::Error`
// is a thiserror enum whose `Ssh` variant Displays as the bare word
// "ssh" and carries the real sentence — which host key, which policy —
// as its `source`. A FAILED job reading `opening git-upload-pack on …:
// ssh` tells an operator nothing they can act on, and the reason was
// one formatting flag away the whole time.
format!(
"opening git-upload-pack on {}: {:#}",
scrub(url),
anyhow::Error::new(err)
)
})?;
let out = {
let (reader, writer) = channel.io();
let mut reader = BufReader::new(reader);
run(&mut reader, writer)
};
drop(key_file);
Ok(out)
}
/// **What a build with no ssh dialler says**, rather than what it does.
///
/// `replication` is the feature that carries `gunnar-client`'s wire half and
/// `nornir-git`'s russh transport; without it there is no dialler in the
/// binary. A mirror registered against an `ssh://` remote on such a build is a
/// registration that can never run, and it says so at the one place it can be
/// acted on — a FAILED job naming the flag — rather than failing as a
/// connection error that sends an operator to look at the network.
///
/// The smart-HTTP half is behind its own feature, `mirror-http` (A32-18), which every
/// appliance row names: [`NO_MIRROR_HTTP`] is its refusal.
#[cfg(not(feature = "replication"))]
fn dial_ssh<T>(
url: &str,
_credential: &Credential,
_known_hosts: &std::path::Path,
_strict: bool,
_run: impl FnOnce(&mut dyn BufRead, &mut dyn Write) -> T,
) -> Result<T, String> {
Err(format!(
"{} is an ssh remote and this build carries no ssh dialler: it was compiled without \
`--features replication`, which is what links `gunnar-client`'s russh transport. \
Mirror from the upstream's `https://` URL, or use a build that has the feature.",
scrub(url)
))
}
/// **Take any userinfo out of a URL before it is shown to anybody.**
///
/// `https://user:token@host/path` is a legal way to write a remote and an
/// operator may well have written one. Everything between `//` and the last
/// `@` of the authority is replaced, so a token that arrived in the URL cannot
/// leave through an error string.
///
/// Not a parser: this runs in an error path, and a URL this cannot make sense
/// of is returned with the whole authority elided rather than passed through.
pub(crate) fn scrub(url: &str) -> String {
let Some((scheme, rest)) = url.split_once("://") else {
return url.to_owned();
};
let (authority, path) = match rest.find('/') {
Some(cut) => (&rest[..cut], &rest[cut..]),
None => (rest, ""),
};
match authority.rfind('@') {
Some(cut) => format!("{scheme}://<redacted>@{}{path}", &authority[cut + 1..]),
None => url.to_owned(),
}
}
/// **An `https://` remote's two halves** ([`crate::mirror_http`], feature `mirror-http`).
#[cfg(feature = "mirror-http")]
fn open_http(
url: &str,
ca: &[Vec<u8>],
auth: Option<String>,
) -> Result<(crate::mirror_http::HttpRead, crate::mirror_http::HttpWrite), String> {
crate::mirror_http::open(url, ca, auth)
}
/// The refusal an `https://` remote gets in a build without `mirror-http` (A32-18): the
/// airgap default has no HTTP client, and says so where it can be acted on.
pub const NO_MIRROR_HTTP: &str = "this build carries no HTTP client: it was compiled without gunnar-server's `mirror-http` feature (the appliance rows name it)";
#[cfg(not(feature = "mirror-http"))]
fn open_http(
url: &str,
_ca: &[Vec<u8>],
_auth: Option<String>,
) -> Result<(std::io::Empty, std::io::Sink), String> {
Err(format!("{} is an https remote and {NO_MIRROR_HTTP}. Mirror from the upstream's ssh:// URL, or use a build that has the feature.", scrub(url)))
}
#[cfg(test)]
mod index_panic_tests {
use super::*;
/// ★ RA4-9 RED: **an index that panics ends the index step with a named refusal**, at
/// once — the progress poller is stopped however the index ends, so the mirror job
/// reaches FAILED instead of hanging for good.
#[test]
fn an_index_that_panics_is_a_named_failure_not_a_hung_job() {
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let r = std::panic::catch_unwind(|| {
indexed_with_progress(|| panic!("the index broke"), || {})
});
let _ = tx.send(r);
});
let r = rx
.recv_timeout(std::time::Duration::from_secs(5))
.expect("the index step ended within 5 s: no hang");
let said = r.expect("a panic in the index is an answer, not a panic up the mirror job");
let why = said.expect_err("a refusal");
assert!(
why.contains("the index panicked") && why.contains("the index broke"),
"{why}"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
/// **The userinfo scrub is the last thing between a token in a URL and the
/// log.** Asserted on the bytes, because "it does not appear" is the whole
/// property.
/// ★ A32-18: the airgap default carries no HTTP client, so an `https://` remote is refused
/// by name (and its token never shown), never dialled.
#[cfg(not(feature = "mirror-http"))]
#[test]
fn an_https_remote_is_refused_by_name_in_a_build_without_mirror_http() {
let e = open_http(
"https://user:tok3n@forge.mock.invalid/ns/repo.git",
&[],
Some("Bearer tok3n".into()),
)
.expect_err("refused");
assert!(
e.contains(NO_MIRROR_HTTP) && e.contains("mirror-http"),
"{e}"
);
assert!(!e.contains("tok3n"), "{e}");
}
#[test]
fn a_url_with_userinfo_never_shows_it() {
let scrubbed = scrub("https://alice:ghp_SUPERSECRET@example.test/ns/repo.git");
assert!(
!scrubbed.contains("ghp_SUPERSECRET"),
"the token survived the scrub: {scrubbed}"
);
assert!(
!scrubbed.contains("alice"),
"the whole userinfo goes, not just the password half: {scrubbed}"
);
assert_eq!(scrubbed, "https://<redacted>@example.test/ns/repo.git");
}
/// A URL with no userinfo is left exactly as the operator wrote it — an
/// error that mangled the ordinary case would be worse than no scrub.
#[test]
fn a_plain_url_is_untouched() {
let url = "https://example.test/ns/repo.git";
assert_eq!(scrub(url), url);
assert_eq!(
scrub("ssh://git@example.test/ns/repo.git"),
"ssh://<redacted>@example.test/ns/repo.git"
);
}
/// **Both transports are recognised from the URL alone**, and the scp-like
/// form is ssh — the shape every forge prints on its clone button.
#[test]
fn a_remote_url_names_its_transport() {
assert_eq!(classify("ssh://git@host/ns/repo.git"), Ok(Wire::Ssh));
assert_eq!(classify("git+ssh://git@host/ns/repo.git"), Ok(Wire::Ssh));
assert_eq!(classify("git@host:ns/repo.git"), Ok(Wire::Ssh));
assert_eq!(classify("https://host/ns/repo.git"), Ok(Wire::Http));
assert_eq!(
classify("http://127.0.0.1:9418/ns/repo.git"),
Ok(Wire::Http)
);
}
/// A remote gunnar cannot dial is refused when it is REGISTERED, not on the
/// first sync — and the refusal names the two forms that work.
#[test]
fn a_remote_that_is_neither_is_refused_by_name() {
let err = classify("/srv/git/ns/repo.git").expect_err("a local path is not a remote");
assert!(err.contains("ssh://") && err.contains("https://"), "{err}");
let err = classify("git://host/ns/repo.git").expect_err("the dumb git protocol");
assert!(err.contains("scheme gunnar cannot mirror from"), "{err}");
}
/// **The mismatch is refused by name, in both directions.**
///
/// The failure being prevented is not an error — it is a 401 (or an SSH
/// authentication failure) against an operator who can see that a
/// credential IS stored, and who will therefore rotate a key that was never
/// the problem.
#[test]
fn a_credential_of_the_wrong_kind_is_refused_by_name() {
let key = Credential::SshKey(b"-----BEGIN OPENSSH PRIVATE KEY-----".to_vec());
let token = Credential::Token {
username: String::new(),
secret: "ghp_SECRET".to_owned(),
};
let err = check_credential(Wire::Http, &key).expect_err("a key cannot ride on http");
assert!(
err.contains("smart http") && err.contains("ssh private key"),
"{err}"
);
assert!(
!err.contains("BEGIN OPENSSH"),
"the refusal quoted the key: {err}"
);
let err = check_credential(Wire::Ssh, &token).expect_err("a token cannot ride on ssh");
assert!(err.contains("ssh") && err.contains("a token"), "{err}");
assert!(
!err.contains("ghp_SECRET"),
"the refusal quoted the token: {err}"
);
// …and the matching pairs, plus the public case, are all fine.
check_credential(Wire::Ssh, &key).expect("a key over ssh");
check_credential(Wire::Http, &token).expect("a token over http");
check_credential(Wire::Ssh, &Credential::None).expect("a public ssh upstream");
check_credential(Wire::Http, &Credential::None).expect("a public http upstream");
}
/// **The authority, out of each of the three shapes** — and the default
/// port with it, because the port is half of what gets dialled.
#[test]
fn a_remote_url_names_the_machine_it_would_dial() {
for (url, want) in [
("https://host/ns/repo.git", ("host".to_owned(), 443)),
("http://host/ns/repo.git", ("host".to_owned(), 80)),
("http://host:9418/ns/repo.git", ("host".to_owned(), 9418)),
("ssh://git@host/ns/repo.git", ("host".to_owned(), 22)),
("ssh://git@host:2222/ns/repo.git", ("host".to_owned(), 2222)),
("git@host:ns/repo.git", ("host".to_owned(), 22)),
// The userinfo is not the host, and a `@` inside it must not become
// one either.
(
"https://user:pass@host/ns/repo.git",
("host".to_owned(), 443),
),
// An IPv6 literal is all colons; the brackets are what separates the
// address from the port, and splitting on the first colon would
// yield `[`.
("ssh://git@[::1]:2222/ns/repo.git", ("::1".to_owned(), 2222)),
(
"https://[2606:4700::1111]/ns/repo.git",
("2606:4700::1111".to_owned(), 443),
),
("git@[::1]:ns/repo.git", ("::1".to_owned(), 22)),
] {
assert_eq!(authority(url), Ok(want.clone()), "{url}");
}
assert!(authority("https:///ns/repo.git").is_err(), "no host at all");
}
/// **Every range a caller must not be able to point this server at**, and
/// the two that must still work.
///
/// Asserted on [`address_verdict`] rather than on [`check_destination`]
/// because the addresses are handed in: no resolver runs, so this is the
/// same answer on a laptop, on a build box with a corporate DNS and in a
/// container with none.
#[test]
fn a_mirror_refuses_every_address_only_the_server_can_reach() {
for addr in [
"127.0.0.1",
"0.0.0.0",
"10.0.0.5",
"172.16.0.1",
"192.168.1.1",
// The one the finding names: cloud instance metadata.
"169.254.169.254",
"100.64.0.1",
"192.0.0.1",
"198.18.0.1",
"240.0.0.1",
"255.255.255.255",
"224.0.0.1",
"::1",
"::",
"fc00::1",
"fe80::1",
"ff02::1",
// …written the long way. Without the mapped-address arm this one
// walks past every v6 test and is allowed.
"::ffff:127.0.0.1",
"::ffff:169.254.169.254",
] {
let ip: std::net::IpAddr = addr.parse().expect("a literal");
let err = address_verdict("upstream.invalid", &[ip])
.expect_err("{addr} must not be dialled on a caller's say-so");
assert!(
err.contains(addr) || err.contains(&ip.to_string()),
"{addr}: {err}"
);
assert!(err.contains(MIRROR_ALLOW_PRIVATE), "{addr}: {err}");
}
for addr in ["93.184.216.34", "8.8.8.8", "2606:4700:4700::1111"] {
let ip: std::net::IpAddr = addr.parse().expect("a literal");
address_verdict("upstream.invalid", &[ip])
.unwrap_or_else(|err| panic!("{addr} is an ordinary public address: {err}"));
}
}
/// **One bad answer out of several is a refusal.** A name that resolves to a
/// routable address and a loopback address is the rebinding shape, and a
/// check that took the first answer, or the best one, would pass it.
#[test]
fn one_private_answer_among_several_refuses_the_whole_name() {
let addrs: Vec<std::net::IpAddr> = ["93.184.216.34", "127.0.0.1"]
.iter()
.map(|a| a.parse().expect("a literal"))
.collect();
let err = address_verdict("both.invalid", &addrs).expect_err("one of them is loopback");
assert!(err.contains("127.0.0.1"), "{err}");
// And nothing at all is not "fine": a name with no address is refused
// rather than allowed through to a dial that would fail anyway.
assert!(address_verdict("nowhere.invalid", &[]).is_err());
}
/// The Basic pair is encoded the way every HTTP client encodes it. Checked
/// against a value anybody can verify by hand rather than against this
/// function's own output.
#[test]
fn basic_auth_is_ordinary_base64() {
assert_eq!(
base64_basic("Aladdin", "open sesame"),
"QWxhZGRpbjpvcGVuIHNlc2FtZQ=="
);
assert_eq!(base64_basic("a", "b"), "YTpi");
assert_eq!(base64_basic("ab", "b"), "YWI6Yg==");
}
}