From 10666ff5b6a6bfcb7b61b43331050d2d807b3ee7 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Tue, 25 Aug 2026 21:18:04 +0900 Subject: [PATCH] gitmirror: `git.mergeCommit` Signed-off-by: Seongmin Lee --- Cargo.lock | 14 + Cargo.toml | 1 + docker-compose.yml | 2 + gitmirror/Cargo.toml | 16 + gitmirror/src/auth.rs | 215 ++++++++++ gitmirror/src/git.rs | 8 +- gitmirror/src/git_repo_ext.rs | 114 ++++++ gitmirror/src/git_transport.rs | 412 +++++++++++++++++++ gitmirror/src/main.rs | 50 ++- gitmirror/src/repo_did.rs | 29 ++ gitmirror/src/xrpc.rs | 57 ++- gitmirror/src/xrpc_git.rs | 720 +++++++++++++++++++++++++++++++++ nix/modules/knotmirror.nix | 2 + 13 files changed, 1633 insertions(+), 7 deletions(-) create mode 100644 gitmirror/src/auth.rs create mode 100644 gitmirror/src/git_repo_ext.rs create mode 100644 gitmirror/src/git_transport.rs create mode 100644 gitmirror/src/repo_did.rs create mode 100644 gitmirror/src/xrpc_git.rs diff --git a/Cargo.lock b/Cargo.lock index 985248372..98f9880b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2745,17 +2745,31 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "base64", + "bobbin-knot-proxy", + "bobbin-runtime", "bobbin-types", + "bytes", "chrono", "clap", + "futures-lite", "gix", + "gix-pack", + "gix-rebase", + "gix-receive-pack", + "gix-transport", "jacquard-axum", "jacquard-common", + "jacquard-identity", + "k256", "line-numbers", + "multibase", + "reqwest 0.13.1", "rustc-hash", "serde", "serde_json", "tempfile", + "thiserror 2.0.18", "tokio", "tokio-stream", "tracing", diff --git a/Cargo.toml b/Cargo.toml index 5baa684f0..d5b928654 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ knot-capability = { path = "crates/knot-capability" } lexicons = { path = "crates/lexicons" } trusted-proxies = { path = "crates/trusted-proxies" } gix-rebase = { path = "crates/gix-rebase" } +gix-receive-pack = { path = "crates/gix-receive-pack" } bobbin-types = { path = "bobbin/crates/types" } bobbin-codesearch = { path = "bobbin/crates/codesearch" } diff --git a/docker-compose.yml b/docker-compose.yml index b0716aea3..66b70df27 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -378,6 +378,8 @@ services: RUST_LOG: gitmirror=debug GITMIRROR_ADDR: 0.0.0.0:9001 GITMIRROR_REPO_BASE: /data/repos + GITMIRROR_PLC_URL: https://plc.tngl.boltless.dev + GITMIRROR_SERVICE_DID: did:web:mirror.tngl.boltless.dev volumes: - knotmirror-data:/data - ./localinfra/certs/root.crt:/usr/local/share/ca-certificates/caddy.crt:ro diff --git a/gitmirror/Cargo.toml b/gitmirror/Cargo.toml index bc7f78572..71d8f93f8 100644 --- a/gitmirror/Cargo.toml +++ b/gitmirror/Cargo.toml @@ -8,11 +8,14 @@ rust-version.workspace = true [dependencies] anyhow = { workspace = true } axum = { workspace = true } +bobbin-knot-proxy = { workspace = true } +bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } chrono = { workspace = true } clap = { workspace = true } jacquard-axum = { workspace = true } jacquard-common = { workspace = true } +jacquard-identity = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["net", "rt-multi-thread", "sync"] } @@ -22,7 +25,20 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } line-numbers = "0.4.0" rustc-hash = "2.1.2" gix = { version = "0.84", features = ["parallel", "blob-diff", "merge", "sha1", "sha256", "revision", "tree-editor"] } +gix-pack = { workspace = true } +gix-receive-pack = { workspace = true } +gix-transport = { workspace = true, features = ["async-client"] } +gix-rebase = { workspace = true } +futures-lite = { workspace = true } +reqwest = { workspace = true } tempfile = "3" +thiserror = { workspace = true } + +[dev-dependencies] +base64 = { workspace = true } +bytes = { workspace = true } +k256 = { workspace = true } +multibase = "0.9" [lints] workspace = true diff --git a/gitmirror/src/auth.rs b/gitmirror/src/auth.rs new file mode 100644 index 000000000..9c3e6fb9d --- /dev/null +++ b/gitmirror/src/auth.rs @@ -0,0 +1,215 @@ +use axum::extract::FromRequestParts; +use axum::http::header; +use axum::http::request::Parts; +use bobbin_runtime::ReqwestHttp; +use jacquard_axum::service_auth::{ + ReplayKey, ReplayStore, ServiceAuth, ServiceAuthConfig, ServiceAuthError, +}; +use jacquard_common::IntoStatic as _; +use jacquard_common::deps::fluent_uri::Uri; +use jacquard_common::deps::smol_str::SmolStr; +use jacquard_common::service_auth; +use jacquard_common::types::crypto::KeyCodec; +use jacquard_common::types::string::Did; +use jacquard_identity::JacquardResolver; +use jacquard_identity::resolver::{IdentityResolver as _, PlcSource}; + +use crate::xrpc::XrpcState; + +/// jacquard's own `HttpClient` is implemented for reqwest 0.12; the workspace is on 0.13, and +/// `bobbin_runtime::ReqwestHttp` is the adapter that already bridges the two. +pub(crate) type Directory = JacquardResolver; + +/// Service auth against `audience`, which is the knot the token will be forwarded to. `plc` is the +/// directory `did:plc:` documents are read from — the token's issuer, and the repo whose knot is +/// pushed to. +pub(crate) fn config( + http: reqwest::Client, + audience: Did, + plc: &str, +) -> anyhow::Result> { + Ok(ServiceAuthConfig::new( + audience, + JacquardResolver::new(ReqwestHttp::new(http), Default::default()) + .with_plc_source(plc_source(plc)?), + ) + .disable_replay_protection()) +} + +/// `jacquard-identity`'s `DidStep::PlcHttp` joins the base and the DID by concatenation, so a base +/// missing its trailing slash would ask for `https://plc.exampledid:plc:...`. +fn plc_source(url: &str) -> anyhow::Result { + let base = format!("{}/", url.trim_end_matches('/')); + Ok(PlcSource::PlcDirectory { + base: Uri::parse(base.as_str()) + .map_err(|e| anyhow::anyhow!("{url:?} is not a usable PLC directory: {e}"))? + .to_owned(), + }) +} + +impl ServiceAuth for XrpcState { + type Resolver = Directory; + + fn service_did(&self) -> Did<&str> { + self.service_auth.service_did() + } + + fn resolver(&self) -> &Self::Resolver { + self.service_auth.resolver() + } + + fn require_lxm(&self) -> bool { + // Qualified: `ServiceAuthConfig`'s inherent `require_lxm` is the builder setter. + ServiceAuth::require_lxm(&*self.service_auth) + } + + fn allowed_services(&self) -> &[SmolStr] { + self.service_auth.allowed_services() + } + + fn replay_protection_enabled(&self) -> bool { + self.service_auth.replay_protection_enabled() + } + + fn replay_store(&self) -> &dyn ReplayStore { + self.service_auth.replay_store() + } +} + +fn bearer_token_from_parts(parts: &Parts) -> Result, ServiceAuthError> { + let Some(auth_header) = parts.headers.get(header::AUTHORIZATION) else { + return Ok(None); + }; + + let auth_str = auth_header + .to_str() + .map_err(|_| ServiceAuthError::InvalidAuthHeader)?; + let token = auth_str + .strip_prefix("Bearer ") + .ok_or(ServiceAuthError::InvalidAuthHeader)?; + Ok(Some(token)) +} + +/// [`jacquard_axum::service_auth::ExtractServiceAuth`] without the audience check, plus the raw +/// token +pub struct ExtractAnyServiceAuth(pub service_auth::ServiceAuthClaims, pub String); + +impl FromRequestParts for ExtractAnyServiceAuth +where + S: ServiceAuth + Send + Sync, + S::Resolver: Send + Sync, +{ + type Rejection = ServiceAuthError; + + async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { + let token = bearer_token_from_parts(&parts)?.ok_or(ServiceAuthError::MissingAuthHeader)?; + let claims = verify_service_auth(state, token).await?; + + Ok(Self(claims, token.to_owned())) + } +} + +async fn verify_service_auth( + state: &S, + token: &str, +) -> Result +where + S: ServiceAuth + Send + Sync, + S::Resolver: Send + Sync, +{ + let parsed = service_auth::parse_jwt(token)?; + let claims = parsed.claims(); + + let doc = state + .resolver() + .resolve_did_doc_owned(&claims.iss) + .await + .map_err(|e| ServiceAuthError::DidResolutionFailed { + did: claims.iss.clone().into_static(), + source: Box::new(e), + })?; + + let signing_key = doc + .atproto_public_key() + .map_err(|e| ServiceAuthError::InvalidKey(e.to_string()))? + .ok_or_else(|| ServiceAuthError::NoSigningKey(claims.iss.clone().into_static())) + .map(|key| match key.codec { + KeyCodec::P256 => Ok(service_auth::PublicKey::from_p256_bytes(&key.bytes)), + KeyCodec::Secp256k1 => Ok(service_auth::PublicKey::from_k256_bytes(&key.bytes)), + codec => { + Err(ServiceAuthError::InvalidKey(format!( + "{codec:?} does not sign service auth tokens" + ))) + } + })???; + + service_auth::verify_signature(&parsed, &signing_key)?; + let now = chrono::Utc::now().timestamp(); + if claims.exp <= now { + return Err(service_auth::ServiceAuthError::Expired { exp: claims.exp, now }.into()); + } + + if state.require_lxm() && claims.lxm.is_none() { + return Err(ServiceAuthError::MethodBindingRequired); + } + + if state.replay_protection_enabled() { + let jti = claims.jti.as_ref().ok_or(ServiceAuthError::MissingJti)?; + let key = ReplayKey::new( + claims.iss.clone().into_static(), + claims.aud.clone().into_static(), + jti.clone(), + ); + state + .replay_store() + .check_and_insert(key, claims.exp) + .await?; + } + + Ok(parsed.into_claims()) +} + +// // TODO(boltless): make more flexible AtprotoService to replace +// // `jacquard_axum::service_auth::ServiceAuth`. +// // It will only verify the parsed signature, `claim.exp` and replayed state. +// pub trait AtprotoService { +// /// The identity resolver type +// type Resolver: IdentityResolver; +// +// /// Get the service DID (expected audience) +// fn service_did(&self) -> Did<&str>; +// +// /// Get a reference to the identity resolver +// fn resolver(&self) -> &Self::Resolver; +// +// /// Replay store used by replay protection. +// fn replay_store(&self) -> Option<&dyn ReplayStore>; +// } +// +// #[derive(Debug, Clone)] +// pub struct VerifiedServiceAuth<'a> { +// did: Did, +// aud: DidService, +// lxm: Option, +// jti: Option>, +// raw: CowStr<'a>, +// } +// +// pub struct ExtractServiceAuth(pub VerifiedServiceAuth<'static>); +// pub struct ExtractOptionalServiceAuth(pub Option>); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_plc_base_carries_the_separator_the_resolver_leaves_out() { + let base = |url| match plc_source(url).unwrap() { + PlcSource::PlcDirectory { base } => base.to_string(), + other => panic!("{other:?}"), + }; + assert_eq!(base("https://plc.example"), "https://plc.example/"); + assert_eq!(base("https://plc.example/"), "https://plc.example/"); + assert!(plc_source("not a url").is_err()); + } +} diff --git a/gitmirror/src/git.rs b/gitmirror/src/git.rs index 74135c1c9..a66529947 100644 --- a/gitmirror/src/git.rs +++ b/gitmirror/src/git.rs @@ -17,9 +17,15 @@ pub(crate) struct TempRepository { impl TempRepository { pub fn open(dir: TempDir) -> Result { - let repo = gix::open(dir.path())?.with_object_memory(); + let repo = gix::open(dir.path())?; Ok(Self { repo, _dir: dir }) } + + #[allow(dead_code)] + pub fn with_object_memory(mut self) -> Self { + self.repo.objects.enable_object_memory(); + self + } } impl std::ops::Deref for TempRepository { diff --git a/gitmirror/src/git_repo_ext.rs b/gitmirror/src/git_repo_ext.rs new file mode 100644 index 000000000..7d0cc46bd --- /dev/null +++ b/gitmirror/src/git_repo_ext.rs @@ -0,0 +1,114 @@ +//! Merge operations layered onto [`gix::Repository`], so they read as repository methods rather +//! than as free functions taking one. + +use std::collections::{HashMap, HashSet}; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum MergeError { + /// Not a failure so much as an answer: these paths couldn't be reconciled, and nothing + /// reachable was written. + #[error("the merge conflicted in {}", .0.join(", "))] + Conflicts(Vec), + #[error("could not walk the commits to replay")] + Walk(#[from] gix::revision::walk::Error), + #[error("could not read a commit to replay")] + ReadCommit(#[from] gix::revision::walk::iter::Error), + #[error("could not set up the blob merge")] + BlobMerge(#[from] Box), + #[error("could not set up the tree diff")] + TreeDiff(#[from] Box), + #[error("could not replay the commits")] + Replay(#[from] gix_rebase::replay::Error), +} + +pub(crate) trait RepositoryExt { + /// Replay `source` and the ancestors of it that `onto` doesn't already have on top of `onto`. + fn rebase( + &self, + source: gix::ObjectId, + onto: gix::ObjectId, + committer: gix::actor::Signature, + ) -> Result; +} + +impl RepositoryExt for gix::Repository { + fn rebase( + &self, + source: gix::ObjectId, + onto: gix::ObjectId, + committer: gix::actor::Signature, + ) -> Result { + let picks = replay_list(self, source, onto)?; + + let options: gix_rebase::replay::Options = Default::default(); + let treat_as_unresolved = options.treat_as_unresolved; + + let mut blob_merge = self + .merge_resource_cache(Default::default()) + .map_err(Box::new)?; + let mut diff_cache = self.diff_resource_cache_for_tree_diff().map_err(Box::new)?; + let mut diff_state = gix::diff::tree::State::default(); + + let outcome = gix_rebase::replay( + &onto, + &picks, + &committer, + Default::default(), + self, + &mut diff_state, + &mut diff_cache, + &mut blob_merge, + options, + )?; + + match outcome { + gix_rebase::replay::Outcome::Complete(done) => Ok(done.tip), + gix_rebase::replay::Outcome::Conflict(conflict) => Err(MergeError::Conflicts( + conflict + .unresolved_paths(treat_as_unresolved) + .map(|path| path.to_string()) + .collect(), + )), + } + } +} + +/// The commits [`RepositoryExt::rebase`] replays, oldest first: everything reachable from `source` +/// that `onto` doesn't have, without the merge commits — the set `git rebase` puts on its todo list. +fn replay_list( + repo: &gix::Repository, + source: gix::ObjectId, + onto: gix::ObjectId, +) -> Result, MergeError> { + let mut parents_of = HashMap::new(); + for info in repo.rev_walk([source]).with_hidden([onto]).all()? { + let info = info?; + parents_of.insert(info.id, info.parent_ids); + } + + // `rev_walk` offers no topological sorting through the gix façade, and replaying a commit + // before its own parent would rebase it onto the wrong tree, so lay the picks out with an + // explicit post-order walk. Iterative, because a branch can be far deeper than the stack. + let mut order = Vec::with_capacity(parents_of.len()); + let mut seen = HashSet::new(); + let mut stack = vec![(source, false)]; + while let Some((id, parents_done)) = stack.pop() { + let Some(parents) = parents_of.get(&id) else { + continue; + }; + if parents_done { + // Merge commits are left out, exactly as `git rebase` does without + // `--rebase-merges`; the commits they brought in are replayed on their own. + if parents.len() < 2 { + order.push(id); + } + continue; + } + if !seen.insert(id) { + continue; + } + stack.push((id, true)); + stack.extend(parents.iter().map(|parent| (*parent, false))); + } + Ok(order) +} diff --git a/gitmirror/src/git_transport.rs b/gitmirror/src/git_transport.rs new file mode 100644 index 000000000..747cf0999 --- /dev/null +++ b/gitmirror/src/git_transport.rs @@ -0,0 +1,412 @@ +//! `gix_transport`'s async [`Transport`] over gitmirror's own `reqwest` client, which is what +//! [`gix_receive_pack::Session`] talks through. +//! +//! Smart HTTP, both halves of it: +//! +//! - `handshake` — `GET /info/refs?service=git-receive-pack`, whose body is fed to a +//! packetline reader after the `application/x-git-receive-pack-advertisement` content-type check +//! and the `# service=` line, exactly as `gix_transport::client::blocking_io::http::Transport` +//! does; [`Handshake::from_lines_with_version_detection`] takes it from there. +//! - `request` — `POST /git-receive-pack` with `application/x-git-receive-pack-request`, and a +//! [`RequestWriter`] over the body it carries. +//! +//! `gix-transport`'s own blocking HTTP transport is the reference implementation; its async +//! counterpart doesn't exist, which is why this is here. +//! +//! Neither body is streamed: `request` is not `async`, so the POST cannot be awaited where it is +//! built, and both bodies are small or already in memory — the pack `gix-receive-pack` writes is a +//! `&[u8]` to begin with, and the reply is one line per ref. So the request body is collected until +//! the response is first read, and the response is read in full before its first byte is handed +//! out. The one thing lost is live side-band progress, which arrives with the report instead; the +//! caller passes `gix::progress::Discard` anyway. + +use std::any::Any; +use std::borrow::Cow; +use std::cell::RefCell; +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::rc::Rc; +use std::task::{Context, Poll, ready}; + +use futures_lite::AsyncReadExt as _; +use gix_transport::async_trait::async_trait; +use gix_transport::bstr::BStr; +use gix_transport::client::async_io::{self, HandleProgress, RequestWriter, SetServiceResponse}; +use gix_transport::client::capabilities::async_recv::Handshake; +use gix_transport::client::{self, MessageKind, TransportWithoutIO, WriteMode}; +use gix_transport::futures_io::{AsyncRead, AsyncWrite}; +use gix_transport::packetline::PacketLineRef; +use gix_transport::packetline::async_io::StreamingPeekableIter; +use gix_transport::{Protocol, Service}; + +/// The `User-Agent` header, which is not the `agent=` capability — that one is the caller's. +const USER_AGENT: &str = concat!("gitmirror/", env!("CARGO_PKG_VERSION")); + +/// A v1 advertisement and a v1 report both end at a flush. +const DELIMITERS: &[PacketLineRef<'static>] = &[PacketLineRef::Flush]; + +pub(crate) struct HttpTransport { + http: reqwest::Client, + /// The caller's service-auth JWT, sent on both halves: a knot authorizes the + /// `git-receive-pack` advertisement, not just the push that follows it. + token: String, + /// The repository's base URL, without the `/info/refs` or `/git-receive-pack` suffix. + url: String, + /// The service `handshake` was performed for; `None` until then. + service: Option, + /// The reader over the response in flight. Created by `handshake` and reused by `request`, both + /// for its line buffers and because the POST is only legal after the advertisement. + lines: Option>, +} + +impl HttpTransport { + pub(crate) fn new(http: reqwest::Client, url: String, token: String) -> Self { + Self { + http, + token, + url, + service: None, + lines: None, + } + } +} + +impl TransportWithoutIO for HttpTransport { + fn to_url(&self) -> Cow<'_, BStr> { + Cow::Borrowed(BStr::new(&self.url)) + } + + /// v0 only: `git-receive-pack` has no v2 command set, and `V1` is the only value that sends no + /// `Git-Protocol` header at all. + fn supported_protocol_versions(&self) -> &[Protocol] { + &[Protocol::V1] + } + + /// Stateless RPC: every turn is its own request. + fn connection_persists_across_multiple_requests(&self) -> bool { + false + } + + fn configure(&mut self, _config: &dyn Any) -> Result<(), Box> { + Ok(()) + } +} + +#[async_trait(?Send)] +impl async_io::Transport for HttpTransport { + /// `extra_parameters` is ignored: it is only ever sent in the `Git-Protocol` header, which v1 + /// does not carry, and this transport speaks nothing else. + async fn handshake<'a>( + &mut self, + service: Service, + _extra_parameters: &'a [(&'a str, Option<&'a str>)], + ) -> Result, client::Error> { + let url = append_url(&self.url, &format!("info/refs?service={}", service.as_str())); + let response = self + .http + .get(url) + .header(reqwest::header::USER_AGENT, USER_AGENT) + .bearer_auth(&self.token) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(io::Error::other)?; + check_content_type(&response, service, "advertisement")?; + let body = response.bytes().await.map_err(io::Error::other)?; + + let line_reader = self.lines.insert(StreamingPeekableIter::new( + ResponseBody::ready(body.to_vec()), + DELIMITERS, + false, + )); + + // The service announcement is only sent by some servers; eat it when it is there, and check + // that it names the service we asked for. Reading to the flush that follows is what lets + // the advertisement itself be read next: `WithSidebands` resets the iterator when dropped. + let announced_service = { + let line_ = line_reader + .peek_line() + .await + .ok_or(client::Error::ExpectedLine("capabilities, version or service"))???; + let line = line_.as_text().ok_or(client::Error::ExpectedLine("text"))?; + line.as_bstr().strip_prefix(b"# service=").map(<[u8]>::to_vec) + }; + if let Some(announced_service) = announced_service { + if announced_service != service.as_str().as_bytes() { + return Err(io::Error::other(format!( + "expected service {:?}, but the remote announced {:?}", + service.as_str(), + String::from_utf8_lossy(&announced_service) + )) + .into()); + } + + line_reader.as_read().read_to_end(&mut Vec::new()).await?; + } + + let Handshake { + capabilities, + refs, + protocol: actual_protocol, + } = Handshake::from_lines_with_version_detection(line_reader).await?; + self.service = Some(service); + Ok(SetServiceResponse { + actual_protocol, + capabilities, + refs, + }) + } + + fn request( + &mut self, + write_mode: WriteMode, + on_into_read: MessageKind, + trace: bool, + ) -> Result, client::Error> { + let service = self.service.ok_or(client::Error::MissingHandshake)?; + let url = append_url(&self.url, service.as_str()); + let http = self.http.clone(); + let token = self.token.clone(); + + // Written into by the `RequestWriter` below, taken by the POST when the response is first + // read — by then the caller has written everything and dropped the writer. + let body = Rc::new(RefCell::new(Vec::new())); + let post = { + let body = Rc::clone(&body); + async move { + let body = std::mem::take(&mut *body.borrow_mut()); + let response = http + .post(url) + .header(reqwest::header::USER_AGENT, USER_AGENT) + .header(reqwest::header::CONTENT_TYPE, content_type(service, "request")) + .header(reqwest::header::ACCEPT, content_type(service, "result")) + .bearer_auth(&token) + .body(body) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(io::Error::other)?; + check_content_type(&response, service, "result")?; + Ok(response.bytes().await.map_err(io::Error::other)?.to_vec()) + } + }; + + let lines = self.lines.as_mut().ok_or(client::Error::MissingHandshake)?; + lines.replace(ResponseBody::pending(post)); + Ok(RequestWriter::new_from_bufread( + BodyWriter(body), + Box::new(lines.as_read_without_sidebands::>()), + write_mode, + on_into_read, + trace, + )) + } +} + +fn append_url(base: &str, suffix: &str) -> String { + format!("{}/{suffix}", base.trim_end_matches('/')) +} + +/// What a smart-HTTP body of the given `kind` is announced as, like +/// `application/x-git-receive-pack-request`. +fn content_type(service: Service, kind: &str) -> String { + format!("application/x-{}-{kind}", service.as_str()) +} + +/// Reject anything but the smart protocol, which is the only one either side speaks. +fn check_content_type(response: &reqwest::Response, service: Service, kind: &str) -> io::Result<()> { + let wanted = content_type(service, kind); + let found = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + if found.trim() != wanted { + return Err(io::Error::other(format!( + "expected a {wanted:?} response to indicate the 'smart' protocol, but got {found:?}" + ))); + } + Ok(()) +} + +/// Collects the request body for [`ResponseBody`] to send. +struct BodyWriter(Rc>>); + +impl AsyncWrite for BodyWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.0.borrow_mut().extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + +/// The POST that a [`ResponseBody`] is still waiting on. +type PendingResponse = Pin>>>>; + +/// A response body, either already in hand (the advertisement) or still to be asked for (the POST, +/// which `request` had no `await` to send it in). +struct ResponseBody { + /// The request that will produce `body`, until it has answered. + pending: Option, + body: io::Cursor>, +} + +impl ResponseBody { + fn ready(body: Vec) -> Self { + Self { + pending: None, + body: io::Cursor::new(body), + } + } + + fn pending(request: impl Future>> + 'static) -> Self { + Self { + pending: Some(Box::pin(request)), + body: io::Cursor::new(Vec::new()), + } + } +} + +impl AsyncRead for ResponseBody { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll> { + let this = self.get_mut(); + if let Some(pending) = this.pending.as_mut() { + let body = ready!(pending.as_mut().poll(cx)); + // Cleared first: a failed request must not be polled again on the next read. + this.pending = None; + this.body = io::Cursor::new(body?); + } + Poll::Ready(io::Read::read(&mut this.body, buf)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_lite::{AsyncBufReadExt as _, AsyncWriteExt as _}; + use gix_transport::client::async_io::Transport as _; + use gix_transport::client::async_io::ReadlineBufRead as _; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + fn pkt(line: &str) -> String { + format!("{:04x}{line}", line.len() + 4) + } + + /// One canned reply per request, in order, on a throwaway port. Returns the base URL and, per + /// request, the head the client sent (lowercased) and its body. + async fn serve( + replies: Vec, + ) -> (String, tokio::task::JoinHandle)>>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/repo", listener.local_addr().unwrap()); + let handle = tokio::spawn(async move { + let mut requests = Vec::new(); + for reply in replies { + let (mut socket, _) = listener.accept().await.unwrap(); + // Read the head, then exactly as many body bytes as were announced. + let mut request = Vec::new(); + let mut byte = [0u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + assert_eq!(socket.read(&mut byte).await.unwrap(), 1, "request cut short"); + request.push(byte[0]); + } + let head = String::from_utf8(request).unwrap().to_lowercase(); + let len: usize = head + .split("content-length: ") + .nth(1) + .map_or(0, |v| v.split("\r\n").next().unwrap().parse().unwrap()); + let mut body = vec![0; len]; + socket.read_exact(&mut body).await.unwrap(); + requests.push((head.clone(), body)); + + let kind = if head.starts_with("get ") { + "advertisement" + } else { + "result" + }; + socket + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/x-git-receive-pack-{kind}\r\n\ + Content-Length: {}\r\n\r\n{reply}", + reply.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + socket.shutdown().await.unwrap(); + } + requests + }); + (url, handle) + } + + /// The whole exchange against a canned knot: the advertisement is parsed, and the POST body is + /// only sent once the reply is read. + #[tokio::test] + async fn a_handshake_and_a_request() { + let tip = "0123456789012345678901234567890123456789"; + let (url, server) = serve(vec![ + format!( + "{}0000{}0000", + pkt("# service=git-receive-pack\n"), + pkt(&format!("{tip} refs/heads/main\0report-status side-band-64k\n")), + ), + format!("{}{}0000", pkt("unpack ok\n"), pkt("ok refs/heads/main\n")), + ]) + .await; + + let mut transport = + HttpTransport::new(reqwest::Client::new(), url, "a.service.jwt".to_owned()); + // Scoped: the response borrows the transport, which the request below needs back. + { + let response = transport.handshake(Service::ReceivePack, &[]).await.unwrap(); + assert_eq!(response.actual_protocol, Protocol::V1); + assert!(response.capabilities.contains("report-status")); + let mut refs = response.refs.expect("v1 advertises refs"); + let mut advertised = String::new(); + refs.readline_str(&mut advertised).await.unwrap(); + assert_eq!(advertised, format!("{tip} refs/heads/main\n")); + } + + let (mut writer, mut reader) = transport + .request(WriteMode::Binary, MessageKind::Flush, false) + .unwrap() + .into_parts(); + writer.write_all(b"the pack").await.unwrap(); + writer.flush().await.unwrap(); + drop(writer); + + let mut report = String::new(); + loop { + let line = reader.fill_buf().await.unwrap(); + if line.is_empty() { + break; + } + let consumed = line.len(); + report.push_str(std::str::from_utf8(line).unwrap()); + reader.consume(consumed); + } + assert_eq!(report, "unpack ok\nok refs/heads/main\n"); + + let requests = server.await.unwrap(); + assert_eq!(requests[0].1, b"", "the advertisement is a GET"); + assert_eq!(requests[1].1, b"the pack"); + // Both halves are authorized by the knot, so both carry the credential. + for (head, _) in &requests { + assert!( + head.contains("authorization: bearer a.service.jwt\r\n"), + "missing credential in {head:?}" + ); + } + } +} diff --git a/gitmirror/src/main.rs b/gitmirror/src/main.rs index ef22364ff..1c2e5cd61 100644 --- a/gitmirror/src/main.rs +++ b/gitmirror/src/main.rs @@ -3,11 +3,16 @@ use std::path::PathBuf; use clap::{Parser, Subcommand}; +mod auth; mod diff; mod git; +mod git_repo_ext; +mod git_transport; mod layout; mod merge; +mod repo_did; mod xrpc; +mod xrpc_git; #[derive(Parser)] #[command(name = "gitmirror", about = "Git mirror service")] @@ -31,6 +36,33 @@ struct ServeArgs { /// Base directory holding bare mirror repos, one per DID (/). #[arg(long, env = "GITMIRROR_REPO_BASE", default_value = "repos")] repo_base: PathBuf, + + /// This service's own DID, e.g. `did:web:mirror.tangled.network`. + #[arg(long, env = "GITMIRROR_SERVICE_DID")] + service_did: String, + + /// Allow knot endpoints on private or loopback addresses. Off in production; on for local + /// testing against a knot on localhost. + #[arg( + long, + env = "GITMIRROR_KNOT_ALLOW_PRIVATE", + default_value_t = false, + default_missing_value = "true" + )] + knot_allow_private: bool, + + /// PLC directory to read `did:plc:` documents from. + #[arg(long, env = "GITMIRROR_PLC_URL", default_value = "https://plc.directory")] + plc_url: String, + + /// Require https on knot endpoints. Disable only when pushing to a local knot. + #[arg( + long, + env = "GITMIRROR_KNOT_REQUIRE_HTTPS", + default_value_t = true, + default_missing_value = "true" + )] + knot_require_https: bool, } #[tokio::main] @@ -42,6 +74,22 @@ async fn main() -> anyhow::Result<()> { .init(); match Cli::parse().cmd { - Command::Serve(args) => xrpc::serve(args.xrpc_addr, args.repo_base).await, + Command::Serve(args) => { + // Parsed here so a malformed DID is a startup failure rather than a surprise on the + // first request that reaches for it. + let service_did = jacquard_common::types::did::Did::new_owned(&args.service_did) + .map_err(|e| anyhow::anyhow!("--service-did {:?}: {e}", args.service_did))?; + xrpc::serve( + args.xrpc_addr, + args.repo_base, + service_did, + xrpc_git::KnotPolicy { + allow_private: args.knot_allow_private, + require_https: args.knot_require_https, + }, + args.plc_url, + ) + .await + } } } diff --git a/gitmirror/src/repo_did.rs b/gitmirror/src/repo_did.rs new file mode 100644 index 000000000..a8204a226 --- /dev/null +++ b/gitmirror/src/repo_did.rs @@ -0,0 +1,29 @@ +use jacquard_common::Bos; +use jacquard_common::deps::fluent_uri::Uri; +use jacquard_common::types::did_doc::DidDocument; +use jacquard_common::types::string::AtprotoStr; +use jacquard_common::types::value::Data; + +const KNOT_SERVICE_TYPE: &str = "TangledKnot"; + +pub(crate) trait DidDocumentExt { + /// The `tangled_knot` service endpoint, if the document declares one. + fn knot_endpoint(&self) -> Option>; +} + +impl + AsRef + Clone> DidDocumentExt for DidDocument { + fn knot_endpoint(&self) -> Option> { + self.service.as_ref().and_then(|services| { + services.iter().find_map(|service| { + if service.r#type.as_ref() != KNOT_SERVICE_TYPE { + return None; + } + match &service.service_endpoint { + Some(Data::String(AtprotoStr::Uri(uri))) => Uri::parse(uri.as_ref()).ok(), + Some(Data::String(AtprotoStr::String(raw))) => Uri::parse(raw.as_ref()).ok(), + _ => None, + } + }) + }) + } +} diff --git a/gitmirror/src/xrpc.rs b/gitmirror/src/xrpc.rs index 628ff704d..910a79ca6 100644 --- a/gitmirror/src/xrpc.rs +++ b/gitmirror/src/xrpc.rs @@ -1,30 +1,39 @@ use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use axum::extract::State; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; -use axum::routing::get; +use axum::routing::{get, post}; use axum::{Json, Router}; use bobbin_types::sh_tangled; use bobbin_types::sh_tangled::git::temp2::{get_diff, get_interdiff, list_commits, merge_check}; use gix::ObjectId; use jacquard_axum::{ExtractXrpc, XrpcResponse}; use jacquard_common::ToSmolStr; +use jacquard_common::types::did::Did; use jacquard_common::types::string::Datetime; use serde_json::json; use tracing::{error, info}; use crate::diff; use crate::git::{commit_log_walk, open_scratch}; +use crate::layout::Layout; const DEFAULT_LIMIT: u32 = 50; const MAX_LIMIT: u32 = 100; #[derive(Clone)] -struct XrpcState { +pub(crate) struct XrpcState { + // TODO: deprecate in favor of `.layout` repo_base: Arc, + pub(crate) layout: Arc, + pub(crate) http: reqwest::Client, + /// Service auth for the endpoints that take a credential; see [`crate::auth`]. + pub(crate) service_auth: Arc>, + pub(crate) knot_policy: crate::xrpc_git::KnotPolicy, } #[derive(Debug)] @@ -34,6 +43,9 @@ pub(crate) enum XrpcError { RefNotFound { rev: String, detail: String }, RevisionNotFound { rev: String }, CompareError(String), + MergeConflict(Vec), + PushRejected(String), + Unauthorized(String), Internal(String), } @@ -70,6 +82,23 @@ impl IntoResponse for XrpcError { "failed to compare revisions".to_owned(), ) } + Self::MergeConflict(paths) => ( + StatusCode::CONFLICT, + "MergeConflict", + format!("merge produced conflicts in {}", paths.join(", ")), + ), + Self::PushRejected(detail) => { + error!(error = %detail, "push rejected"); + ( + StatusCode::CONFLICT, + "PushRejected", + "knot rejected the push".to_owned(), + ) + } + Self::Unauthorized(detail) => { + error!(error = %detail, "authentication failed"); + (StatusCode::UNAUTHORIZED, "AuthenticationRequired", detail) + } Self::Internal(m) => { error!(error = %m, "xrpc request failed"); ( @@ -198,7 +227,7 @@ impl From for sh_tangled::git::FileDiff { } /// Resolve a full hex oid to a commit that actually exists in `repo`. -fn find_commit(repo: &gix::Repository, sha: &str) -> Result { +pub(crate) fn find_commit(repo: &gix::Repository, sha: &str) -> Result { let oid = gix::ObjectId::from_hex(sha.as_bytes()) .map_err(|e| XrpcError::InvalidRequest(format!("bad commit sha {sha:?}: {e}")))?; repo.find_commit(oid) @@ -394,7 +423,17 @@ async fn list_commits( }) } -pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> { +pub async fn serve( + addr: SocketAddr, + repo_base: PathBuf, + service_did: Did, + knot_policy: crate::xrpc_git::KnotPolicy, + plc: String, +) -> anyhow::Result<()> { + let http = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(300)) + .build()?; let app = Router::new() .route("/xrpc/sh.tangled.git.temp2.getDiff", get(get_diff)) .route( @@ -403,8 +442,16 @@ pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> { ) .route("/xrpc/sh.tangled.git.temp2.listCommits", get(list_commits)) .route("/xrpc/sh.tangled.git.temp2.mergeCheck", get(merge_check)) + .route( + "/xrpc/sh.tangled.git.mergeCommit", + post(crate::xrpc_git::merge_commit), + ) .with_state(XrpcState { - repo_base: Arc::new(repo_base), + repo_base: Arc::new(repo_base.clone()), + layout: Arc::new(Layout::new(repo_base)), + http: http.clone(), + service_auth: Arc::new(crate::auth::config(http, service_did, &plc)?), + knot_policy, }); let listener = tokio::net::TcpListener::bind(addr).await?; diff --git a/gitmirror/src/xrpc_git.rs b/gitmirror/src/xrpc_git.rs new file mode 100644 index 000000000..f67189b66 --- /dev/null +++ b/gitmirror/src/xrpc_git.rs @@ -0,0 +1,720 @@ +use axum::extract::State; +use bobbin_knot_proxy::KnotHost; +use bobbin_types::sh_tangled::git::merge_commit; +use jacquard_axum::service_auth::ServiceAuth; +use jacquard_axum::{ExtractXrpc, XrpcResponse}; +use jacquard_identity::resolver::IdentityResolver as _; +use tracing::info; + +use crate::auth::ExtractAnyServiceAuth; +use crate::git_repo_ext::{MergeError, RepositoryExt as _}; +use crate::repo_did::DidDocumentExt as _; +use crate::xrpc::{find_commit, XrpcError, XrpcState}; + +const AGENT: &str = concat!("gitmirror/", env!("CARGO_PKG_VERSION")); + +const COMMITTER_NAME: &str = "Tangled"; +const COMMITTER_EMAIL: &str = "noreply@tangled.sh"; +const REPO_PUSH_NSID: &str = "sh.tangled.repo.push"; + +enum MergeStyle { + Rebase, + // TODO(post-1.0): support more merge strategies + // Merge, + // RebaseMerge, + // SquashRebase, + // SquashRebaseMerge, + // FastForwardOnly, +} + +impl MergeStyle { + fn parse(style: &str) -> Option { + match style { + "rebase" => Some(Self::Rebase), + _ => None, + } + } +} + +pub(crate) async fn merge_commit( + State(state): State, + ExtractAnyServiceAuth(auth, token): ExtractAnyServiceAuth, + ExtractXrpc(input): ExtractXrpc, +) -> Result, XrpcError> { + if !auth + .lxm + .as_ref() + .is_some_and(|lxm| lxm.as_str() == REPO_PUSH_NSID) + { + return Err(XrpcError::Unauthorized(format!( + "the token is bound to {:?}, which is not the push method", + auth.lxm.as_ref().map(|lxm| lxm.as_str()) + ))); + } + + // 1. validate inputs + let merge_style = MergeStyle::parse(input.style.as_str()).ok_or_else(|| { + XrpcError::InvalidRequest(format!("unknown merge style: {:?}", input.style.as_str())) + })?; + let target_branch = + gix::refs::FullName::try_from(format!("refs/heads/{}", input.target.branch.as_str())) + .map_err(|e| { + XrpcError::InvalidRequest(format!( + "bad branch name {:?}: {e}", + input.target.branch.as_str() + )) + })?; + + // 2. load target & source repos as scratch repo. + let scratch = state + .layout + .open_scratch(&[&input.target.repo, &input.source.repo]) + .map_err(|e| XrpcError::Internal(e.to_string()))? + .ok_or_else(|| XrpcError::RepoNotFound { detail: "repo not found".to_string() })?; + let source_commit_id = find_commit(&scratch, input.source.commit.as_str())?; + let target_commit_id = find_commit(&scratch, input.target.commit.as_str())?; + + // TODO(boltless): check knot reachability here + + let mut hidden = vec![target_commit_id]; + if input.source.repo == input.target.repo { + hidden.push(source_commit_id); + } + + let remote = { + let repo = input.target.repo.as_str(); + let document = state + .resolver() + .resolve_did_doc_owned(&input.target.repo) + .await + .map_err(|e| XrpcError::Internal(format!("could not resolve {repo}: {e}")))?; + let endpoint = document + .knot_endpoint() + .ok_or_else(|| { + XrpcError::InvalidRequest(format!("{repo} declares no knot service")) + })?; + let knot = safe_knot_host(endpoint.as_str(), state.knot_policy).map_err(|_reason| { + XrpcError::InvalidRequest(format!("{repo} declares invalid knot service endpoint")) + })?; + + let audience = knot_service_did(&knot); + if auth.aud.audience().as_str() != audience { + return Err(XrpcError::Unauthorized(format!( + "the token is audienced to {}, but {repo} is hosted on {audience}", + auth.aud.as_str() + ))); + } + + format!("{}{}", knot.url(), repo) + }; + + let http = state.http.clone(); + let committer = gix::actor::Signature { + name: COMMITTER_NAME.into(), + email: COMMITTER_EMAIL.into(), + time: gix::date::Time::new(chrono::Utc::now().timestamp(), 0), + }; + // NOTE: gix's async traits doesn't implement `Send`, so we should basically do a blocking IO + // while pushing. Not using `gix_transport::blocking_io` because that one is for one-shot and + // loose connection between handshake & git-receive-pack calls. + // TODO(boltless): implement `gix_transport::async_io_send` or something like that. + let runtime = tokio::runtime::Handle::current(); + let (merged, statuses) = tokio::task::spawn_blocking(move || { + // 3. merge + let merged = match merge_style { + MergeStyle::Rebase => scratch.rebase(source_commit_id, target_commit_id, committer), + // TODO(post-1.0): support more merge strategies + } + .map_err(|e| match e { + MergeError::Conflicts(paths) => XrpcError::MergeConflict(paths), + _ => XrpcError::Internal(e.to_string()), + })?; + + // 4. push the merged tip to the target repository + let mut transport = crate::git_transport::HttpTransport::new(http, remote.clone(), token); + let mut session = runtime + .block_on(gix_receive_pack::Session::handshake( + &mut transport, + Some(AGENT), + )) + .map_err(|e| XrpcError::Internal(format!("{remote}: {e}")))?; + + let pack = build_pack(&scratch, session.advertised(), merged, hidden) + .map_err(|e| XrpcError::Internal(format!("{e:#}")))?; + + let statuses = runtime + .block_on(session.push( + false, + &[gix_receive_pack::Update { + name: target_branch, + expected_old: target_commit_id, + new: merged, + }], + pack.as_deref(), + gix_receive_pack::Options { + atomic: true, + push_options: &[], + quiet: false, + }, + &mut gix::progress::Discard, + &gix::interrupt::IS_INTERRUPTED, + )) + .map_err(|e| XrpcError::Internal(format!("{remote}: {e}")))?; + Ok::<_, XrpcError>((merged, statuses)) + }) + .await + .map_err(|e| XrpcError::Internal(e.to_string()))??; + + if let Some(reason) = statuses.iter().find_map(|s| s.status.as_ref().err()) { + return Err(XrpcError::PushRejected(reason.to_string())); + } + + info!( + pusher = %auth.iss.as_str(), + repo = %input.target.repo.as_str(), + branch = %input.target.branch.as_str(), + commit = %merged, + "merged" + ); + Ok(XrpcResponse(())) +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct KnotPolicy { + pub allow_private: bool, + pub require_https: bool, +} + +fn safe_knot_host(endpoint: &str, policy: KnotPolicy) -> Result { + let host = KnotHost::parse(endpoint).map_err(|_| "is not a usable base URL")?; + if policy.require_https && host.url().scheme() != "https" { + return Err("is not https"); + } + if !policy.allow_private && host.private_literal_reason().is_some() { + return Err("is a private address"); + } + Ok(host) +} + +/// The DID a knot serving this host is addressed by, built exactly as Go's `serviceauth.DidWeb` +/// (xrpc/serviceauth/service_auth.go:22) builds it: the authority, with `:` percent-encoded. The +/// `url` crate leaves out a default port, and so does the hostname the appview mints from. +fn knot_service_did(knot: &KnotHost) -> String { + let url = knot.url(); + match url.port() { + // `did:web` percent-encodes the port separator; everything after it is the method-specific + // id, where a bare `:` would read as a path segment. + Some(port) => format!("did:web:{}%3A{port}", url.host_str().unwrap_or_default()), + None => format!("did:web:{}", url.host_str().unwrap_or_default()), + } +} + +/// The packfile the knot is missing: every object reachable from `new_commit_id` that neither +/// `hidden` nor anything `advertised` already accounts for — its ref tips and its `.have` lines, +/// which name the tips of its alternates. +/// +/// Only the ids this repository can actually load make it into the exclusion set: the walk reads +/// each one to mark its ancestry uninteresting, so an advertised tip a lagging mirror does not yet +/// have would otherwise fail the walk rather than merely widen the pack. +/// +/// `None` when the knot already holds every object the update names, which a fast-forward within +/// one repository is: there is then nothing to build a pack out of. A knot reads the body to its +/// end and stages nothing (`knot-pack::receive::stage_pack_bytes` treats empty as no pack), so the +/// body simply stops after the command block. +fn build_pack( + scratch: &crate::git::TempRepository, + advertised: &gix_receive_pack::ReceivePackAdvertisement, + new_commit_id: gix::ObjectId, + hidden: Vec, +) -> anyhow::Result>> { + use gix::parallel::{reduce::Finalize as _, InOrderIter}; + use gix::progress::Discard; + use gix_pack::data::{output, Version}; + + const CHUNK_SIZE: usize = 50; + + anyhow::ensure!( + advertised.capabilities.contains("ofs-delta"), + "the knot does not read offset deltas" + ); + let object_hash = advertised.object_hash()?; + anyhow::ensure!( + object_hash == scratch.object_hash(), + "the knot speaks {object_hash} but the mirror is {}", + scratch.object_hash() + ); + + let mut objects = scratch.objects.clone().into_inner(); + objects.prevent_pack_unload(); + + // The walk loads every hidden id to mark its ancestry uninteresting, so one it cannot load + // fails the whole walk — and the knot advertises refs a mirror that is behind may not have yet. + // Peeling is what turns an advertised annotated tag into the commit the walk wants. + let hidden: Vec<_> = hidden + .into_iter() + .chain(advertised.remote_refs.values().copied()) + .chain(advertised.advertised_haves.iter().copied()) + .filter_map(|id| Some(scratch.find_object(id).ok()?.peel_to_commit().ok()?.id)) + .collect(); + let commits: Vec = scratch + .rev_walk([new_commit_id]) + .with_hidden(hidden) + .all()? + .map(|info| info.map(|info| info.id)) + .collect::>()?; + if commits.is_empty() { + return Ok(None); + } + + // ponytail: the exclusion is commit-level only, so an object the knot already has can still be + // sent when a commit's own diff reintroduces it — `gix-pack` keeps the set of objects it has + // seen private and offers no way to seed it with the remote's. The fix is hand-rolled counting + // straight into `Vec`. + let (counts, _) = output::count::objects( + objects.clone(), + Box::new(commits.into_iter().map(Ok)), + &Discard, + &gix::interrupt::IS_INTERRUPTED, + output::count::objects::Options { + // TODO(boltless): get thread limit like knot does + thread_limit: Some(1), + chunk_size: CHUNK_SIZE, + input_object_expansion: + output::count::objects::ObjectExpansion::TreeAdditionsComparedToAncestor, + }, + )?; + + let num_entries = u32::try_from(counts.len()) + .map_err(|_| anyhow::anyhow!("{} objects do not fit into a pack", counts.len()))?; + let counted = output::entry::iter_from_counts( + counts, + objects, + Box::new(Discard), + output::entry::iter_from_counts::Options { + // TODO(boltless): get thread limit like knot does + thread_limit: Some(1), + mode: output::entry::iter_from_counts::Mode::PackCopyAndBaseObjects, + allow_thin_pack: false, + chunk_size: CHUNK_SIZE, + version: Version::V2, + }, + ); + + // Entries have to reach the writer in the order they were counted, because an offset delta + // refers to its base by index into that very sequence. + let mut in_order = InOrderIter::from(counted); + let mut pack = Vec::new(); + // The `PACK` header and the trailing checksum are the writer's own doing; driving it to + // exhaustion is what produces them. + let mut writer = output::bytes::FromEntriesIter::new( + in_order.by_ref(), + &mut pack, + num_entries, + Version::V2, + object_hash, + ); + writer.try_fold((), |(), written| written.map(|_| ()))?; + + let stats = in_order.inner.finalize()?; + anyhow::ensure!( + stats.missing_objects == 0, + "{} object(s) could not be found, which would have produced a corrupt pack", + stats.missing_objects + ); + + Ok(Some(pack)) +} + +#[cfg(test)] +mod tests { + use super::*; + use jacquard_common::types::did::Did; + use std::io::Write as _; + use std::path::{Path, PathBuf}; + use std::process::{Command, Stdio}; + + use crate::layout::Layout; + + const TARGET_DID: &str = "did:plc:upstream"; + const SOURCE_DID: &str = "did:plc:fork"; + const CHANGE_ID: &str = "zzzzkzkzkzkzkzkzkzkzkzkzkzkz"; + + fn committer() -> gix::actor::Signature { + gix::actor::Signature { + name: COMMITTER_NAME.into(), + email: COMMITTER_EMAIL.into(), + time: gix::date::Time::new(chrono::Utc::now().timestamp(), 0), + } + } + + fn git(dir: &Path, args: &[&str]) -> String { + let out = Command::new("git") + .args(args) + .current_dir(dir) + .env("GIT_AUTHOR_NAME", "t") + .env("GIT_AUTHOR_EMAIL", "t@t") + .env("GIT_COMMITTER_NAME", "t") + .env("GIT_COMMITTER_EMAIL", "t@t") + .output() + .expect("run git"); + assert!( + out.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8(out.stdout).unwrap().trim().to_owned() + } + + /// `git hash-object` and friends need the object on stdin. + fn git_stdin(dir: &Path, args: &[&str], stdin: &[u8]) -> String { + let mut child = Command::new("git") + .args(args) + .current_dir(dir) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .expect("spawn git"); + child.stdin.take().unwrap().write_all(stdin).unwrap(); + let out = child.wait_with_output().expect("run git"); + assert!(out.status.success(), "git {args:?} failed"); + String::from_utf8(out.stdout).unwrap().trim().to_owned() + } + + /// Loose objects under a bare repo's `objects` dir, ignoring `info` and `pack`. + fn count_loose_objects(objects: &Path) -> usize { + let mut n = 0; + for shard in std::fs::read_dir(objects).unwrap() { + let shard = shard.unwrap(); + let name = shard.file_name(); + if name == "info" || name == "pack" || !shard.file_type().unwrap().is_dir() { + continue; + } + n += std::fs::read_dir(shard.path()).unwrap().count(); + } + n + } + + struct Fixture { + root: tempfile::TempDir, + layout: Layout, + /// The target branch tip, living only in the target mirror. + onto: gix::ObjectId, + /// The source tip, living only in the source mirror. + source: gix::ObjectId, + /// The tree real `git rebase` produces for the same replay. + expected_tree: gix::ObjectId, + } + + impl Fixture { + fn objects(&self, did: &str) -> PathBuf { + self.layout.repo_path(&Did::raw(did.into())).join("objects") + } + } + + /// Two mirrors sharing a base commit, the real gitmirror shape: the target tip is only in the + /// target's mirror and the source tip only in the source's. The source's first commit carries a + /// hand-written `change-id`, which `git commit` cannot produce. + fn two_mirrors(target_a: &str, source_a: &str) -> Fixture { + let root = tempfile::tempdir().unwrap(); + let repo_base = root.path().join("repos"); + std::fs::create_dir_all(&repo_base).unwrap(); + + let target = root.path().join("target"); + std::fs::create_dir_all(&target).unwrap(); + git(&target, &["init", "-q", "-b", "main"]); + std::fs::write(target.join("a.txt"), "line1\nline2\nline3\n").unwrap(); + git(&target, &["add", "."]); + git(&target, &["commit", "-q", "-m", "shared ancestor"]); + + // Fork before the target moves on, so neither tip is reachable from the other. + let source = root.path().join("source"); + git( + root.path(), + &[ + "clone", + "-q", + target.to_str().unwrap(), + source.to_str().unwrap(), + ], + ); + + std::fs::write(target.join("a.txt"), target_a).unwrap(); + git(&target, &["add", "-A"]); + git(&target, &["commit", "-q", "-m", "target moves on"]); + + std::fs::write(source.join("a.txt"), source_a).unwrap(); + git(&source, &["add", "-A"]); + let tree = git(&source, &["write-tree"]); + let parent = git(&source, &["rev-parse", "HEAD"]); + let raw = format!( + "tree {tree}\nparent {parent}\n\ + author A 1700000060 +0000\n\ + committer C 1700000060 +0000\n\ + change-id {CHANGE_ID}\n\n\ + source edit with a change-id\n" + ); + let id = git_stdin( + &source, + &["hash-object", "-t", "commit", "-w", "--stdin"], + raw.as_bytes(), + ); + git(&source, &["update-ref", "refs/heads/main", &id]); + git(&source, &["reset", "-q", "--hard", "main"]); + std::fs::write(source.join("source.txt"), "mine\n").unwrap(); + git(&source, &["add", "-A"]); + git(&source, &["commit", "-q", "-m", "plain follow-up"]); + + let onto = git(&target, &["rev-parse", "HEAD"]); + let source_tip = git(&source, &["rev-parse", "HEAD"]); + + for (did, src) in [(TARGET_DID, &target), (SOURCE_DID, &source)] { + git( + root.path(), + &[ + "clone", + "-q", + "--bare", + src.to_str().unwrap(), + repo_base.join(did).to_str().unwrap(), + ], + ); + } + + // The baseline: what real `git rebase` makes of the same replay. Run after the mirrors are + // cloned, so it can't influence what they hold. + git(&source, &["remote", "add", "up", target.to_str().unwrap()]); + git(&source, &["fetch", "-q", "up"]); + let rebase = Command::new("git") + .args(["rebase", "-q", &onto]) + .current_dir(&source) + .env("GIT_AUTHOR_NAME", "t") + .env("GIT_AUTHOR_EMAIL", "t@t") + .env("GIT_COMMITTER_NAME", "t") + .env("GIT_COMMITTER_EMAIL", "t@t") + .output() + .expect("run git rebase"); + let expected_tree = if rebase.status.success() { + git(&source, &["rev-parse", "HEAD^{tree}"]) + } else { + // A conflicting fixture has no baseline tree; the test asserting the conflict ignores it. + onto.clone() + }; + + Fixture { + layout: Layout::new(repo_base), + onto: gix::ObjectId::from_hex(onto.as_bytes()).unwrap(), + source: gix::ObjectId::from_hex(source_tip.as_bytes()).unwrap(), + expected_tree: gix::ObjectId::from_hex(expected_tree.as_bytes()).unwrap(), + root, + } + } + + fn scratch(fixture: &Fixture) -> crate::git::TempRepository { + fixture + .layout + .open_scratch(&[&Did::raw(TARGET_DID.into()), &Did::raw(SOURCE_DID.into())]) + .unwrap() + .unwrap() + } + + #[test] + fn a_cross_mirror_rebase_matches_git_and_keeps_the_change_id() { + // The target edits the last line, the source the first: a clean three-way merge. + let fixture = two_mirrors("line1\nline2\nTARGET\n", "SOURCE\nline2\nline3\n"); + let before: Vec = [TARGET_DID, SOURCE_DID] + .iter() + .map(|did| count_loose_objects(&fixture.objects(did))) + .collect(); + + let repo = scratch(&fixture); + let tip = repo + .rebase(fixture.source, fixture.onto, committer()) + .expect("a clean rebase"); + assert_eq!( + repo.find_commit(tip).unwrap().tree_id().unwrap().detach(), + fixture.expected_tree, + "tree differs from what `git rebase` produces" + ); + + // The `change-id` rides along on the rewritten commit, which `git rebase` itself drops. + let replayed = repo.find_commit(tip).unwrap(); + let first = replayed.decode().unwrap().parents().next().unwrap(); + let first = repo.find_commit(first).unwrap(); + let decoded = first.decode().unwrap(); + assert_eq!( + decoded + .extra_headers() + .find("change-id") + .map(|v| v.to_string()) + .as_deref(), + Some(CHANGE_ID), + ); + // Author preserved, committer replaced by ours. + assert_eq!(decoded.author().unwrap().email, "a@example.com"); + assert_eq!(decoded.committer().unwrap().email, COMMITTER_EMAIL); + + // Nothing was written to either mirror: the replay lives in the scratch repo's memory. + for (did, before) in [TARGET_DID, SOURCE_DID].iter().zip(before) { + assert_eq!( + count_loose_objects(&fixture.objects(did)), + before, + "{did} mirror was mutated" + ); + } + drop(fixture.root); + } + + /// What a knot advertises for a target sitting at `onto`, plus a ref the mirror has never seen + /// — a branch pushed to the knot after the mirror last fetched. + fn advertisement(onto: gix::ObjectId) -> gix_receive_pack::ReceivePackAdvertisement { + let (capabilities, _) = gix_transport::client::Capabilities::from_bytes( + b"\0report-status delete-refs atomic ofs-delta side-band-64k", + ) + .unwrap(); + gix_receive_pack::ReceivePackAdvertisement { + server_protocol_version: gix_transport::Protocol::V1, + capabilities, + remote_refs: [ + (gix::refs::FullName::try_from("refs/heads/main").unwrap(), onto), + ( + gix::refs::FullName::try_from("refs/heads/elsewhere").unwrap(), + gix::ObjectId::from_hex(b"1111111111111111111111111111111111111111").unwrap(), + ), + ] + .into_iter() + .collect(), + advertised_haves: Vec::new(), + } + } + + #[test] + fn the_pack_carries_everything_the_knot_is_missing() { + let fixture = two_mirrors("line1\nline2\nTARGET\n", "SOURCE\nline2\nline3\n"); + let repo = scratch(&fixture); + let merged = repo + .rebase(fixture.source, fixture.onto, committer()) + .expect("a clean rebase"); + + let pack = build_pack( + &repo, + &advertisement(fixture.onto), + merged, + vec![fixture.onto], + ) + .expect("the pack is built despite an advertised ref the mirror lacks") + .expect("a rebase writes new commits, so there is a pack"); + + // A stand-in for the knot: it holds exactly what the target mirror holds, which is what its + // advertisement said. Anything the merged tip needs beyond that has to be in the pack. + let knot = fixture.root.path().join("knot"); + git( + fixture.root.path(), + &[ + "clone", + "-q", + "--bare", + fixture + .layout + .repo_path(&Did::raw(TARGET_DID.into())) + .to_str() + .unwrap(), + knot.to_str().unwrap(), + ], + ); + git_stdin(&knot, &["unpack-objects", "-q"], &pack); + git(&knot, &["update-ref", "refs/heads/main", &merged.to_string()]); + // Walks every object the branch reaches and fails on one that is not there. + git(&knot, &["fsck", "--connectivity-only", "--no-dangling"]); + assert_eq!( + git(&knot, &["rev-parse", "main^{tree}"]), + fixture.expected_tree.to_string(), + "the knot ends up with the tree `git rebase` produces" + ); + drop(fixture.root); + } + + #[test] + fn a_tip_the_knot_already_has_needs_no_pack() { + // What a fast-forward within one repository will look like: the update names a commit the + // knot advertised, so the walk comes up empty and there is nothing to send. + let fixture = two_mirrors("line1\nline2\nTARGET\n", "SOURCE\nline2\nline3\n"); + let repo = scratch(&fixture); + let pack = build_pack( + &repo, + &advertisement(fixture.onto), + fixture.onto, + vec![fixture.onto], + ) + .expect("the walk runs"); + assert!(pack.is_none(), "no objects, so no pack"); + drop(fixture.root); + } + + #[test] + fn a_knot_endpoint_becomes_a_base_url_and_only_a_public_https_one() { + let deployed = KnotPolicy { + allow_private: false, + require_https: true, + }; + let base = |endpoint| safe_knot_host(endpoint, deployed).map(|host| host.url().to_string()); + assert_eq!( + base("https://knot.example/repo/abc").ok().as_deref(), + Some("https://knot.example/"), + "the tagged repo path is not part of the knot's base" + ); + assert_eq!(base("http://knot.example"), Err("is not https")); + assert_eq!(base("https://127.0.0.1/repo/abc"), Err("is a private address")); + assert_eq!(base("https://localhost/repo/abc"), Err("is a private address")); + assert_eq!(base("not a url"), Err("is not a usable base URL")); + + // The local-development hatch: the same endpoints, minus both guards. + let dev = KnotPolicy { + allow_private: true, + require_https: false, + }; + let base = |endpoint| safe_knot_host(endpoint, dev).map(|host| host.url().to_string()); + assert_eq!( + base("http://127.0.0.1:5555/repo/abc").ok().as_deref(), + Some("http://127.0.0.1:5555/") + ); + assert_eq!( + base("http://localhost:5555").ok().as_deref(), + Some("http://localhost:5555/") + ); + // Not a hatch for nonsense: parsing still has to succeed. + assert_eq!(base("not a url"), Err("is not a usable base URL")); + } + + /// The audience a push token has to name to be forwarded to this knot. The handler compares + /// `claims.aud.audience()` against exactly this string, so a token minted for one knot cannot + /// be spent at another. + #[test] + fn a_knot_is_addressed_by_the_did_web_of_its_authority() { + let dev = KnotPolicy { + allow_private: true, + require_https: false, + }; + let did = |endpoint| knot_service_did(&safe_knot_host(endpoint, dev).unwrap()); + + assert_eq!(did("https://knot.example/repo/abc"), "did:web:knot.example"); + // The port separator is percent-encoded, or it would read as a `did:web` path segment. + assert_eq!(did("http://localhost:5555"), "did:web:localhost%3A5555"); + // A default port is not part of the authority the appview mints from either. + assert_eq!(did("https://knot.example:443"), "did:web:knot.example"); + } + + #[test] + fn a_conflicting_rebase_names_the_path_and_writes_no_commit() { + // Both sides rewrite the same line. + let fixture = two_mirrors("line1\nTARGET\nline3\n", "line1\nSOURCE\nline3\n"); + let repo = scratch(&fixture); + let err = repo + .rebase(fixture.source, fixture.onto, committer()) + .expect_err("a conflict"); + assert!( + matches!(err, MergeError::Conflicts(ref paths) if *paths == ["a.txt"]), + "{err:?}" + ); + drop(fixture.root); + } +} diff --git a/nix/modules/knotmirror.nix b/nix/modules/knotmirror.nix index c8441e263..16c94464b 100644 --- a/nix/modules/knotmirror.nix +++ b/nix/modules/knotmirror.nix @@ -195,6 +195,8 @@ in Environment = [ "GITMIRROR_ADDR=${cfg.v2.listenAddr}" "GITMIRROR_REPO_BASE=/var/lib/knotmirror/repos" + "GITMIRROR_PLC_URL=${cfg.atpPlcUrl}" + "GITMIRROR_SERVICE_DID=did:web:${cfg.hostname}" ]; ExecStart = "${getExe cfg.v2.package} serve"; Restart = "always"; -- 2.54.0