diff --git a/knot2/crates/knot-git/src/repo.rs b/knot2/crates/knot-git/src/repo.rs index 2b214c0a..3f3f09bc 100644 --- a/knot2/crates/knot-git/src/repo.rs +++ b/knot2/crates/knot-git/src/repo.rs @@ -212,7 +212,17 @@ impl Layout { } pub fn create(&self, did: &RepoDid) -> Result { - self.init_repo(self.guarded_path(did)?) + self.create_with_format(did, self.object_format) + } + + pub fn create_with_format( + &self, + did: &RepoDid, + format: ObjectFormat, + ) -> Result { + let repo = Repo::create_with_format(self.guarded_path(did)?, format)?; + repo.set_head(&self.head)?; + Ok(repo) } pub fn remove(&self, did: &RepoDid) -> Result<(), GitError> { @@ -233,12 +243,6 @@ impl Layout { pub fn bootstrap_meta(&self, knot: &KnotId) -> Result { init_bare_idempotent(self.meta_path(knot)?) } - - fn init_repo(&self, path: PathBuf) -> Result { - let repo = Repo::create_with_format(path, self.object_format)?; - repo.set_head(&self.head)?; - Ok(repo) - } } pub(crate) fn init_bare_with_format( diff --git a/knot2/crates/knot-pack/src/fetch.rs b/knot2/crates/knot-pack/src/fetch.rs index d692caf3..bdf3165b 100644 --- a/knot2/crates/knot-pack/src/fetch.rs +++ b/knot2/crates/knot-pack/src/fetch.rs @@ -3,7 +3,7 @@ use std::io::Write; use axum::http::{HeaderMap, HeaderValue, Method, header}; use knot_git::{Filter, RefRecord, Repo}; use knot_runtime::{HttpRequest, HttpResponse, HttpTransport, NetworkError}; -use knot_types::{HttpStatus, Oid, RefName}; +use knot_types::{HttpStatus, ObjectFormat, Oid, RefName}; use url::Url; use crate::error::PackError; @@ -30,6 +30,7 @@ pub enum FetchError { #[derive(Debug, Clone, PartialEq, Eq)] pub struct UpstreamRefs { + pub object_format: ObjectFormat, pub head_symref: Option, pub refs: Vec, } @@ -81,7 +82,7 @@ async fn execute( Ok(response) } -pub fn parse_advertisement(body: &[u8]) -> Result<(), FetchError> { +pub fn parse_advertisement(body: &[u8]) -> Result { let lines = pkt::data_payloads_all(body).map_err(|error| protocol(error.to_string()))?; let lines: Vec<&str> = lines .iter() @@ -98,7 +99,14 @@ pub fn parse_advertisement(body: &[u8]) -> Result<(), FetchError> { if !has("ls-refs") || !has("fetch") { return Err(protocol("upstream is missing ls-refs or fetch v2 command")); } - Ok(()) + lines + .iter() + .find_map(|line| line.strip_prefix("object-format=")) + .map_or(Ok(ObjectFormat::SHA1), |token| { + ObjectFormat::from_capability(token).ok_or_else(|| { + protocol(format!("upstream advertises unknown object format {token}")) + }) + }) } pub fn ls_refs_request(prefixes: &[&str]) -> Result, PackError> { @@ -114,10 +122,11 @@ pub fn ls_refs_request(prefixes: &[&str]) -> Result, PackError> { Ok(buf) } -pub fn parse_ls_refs(body: &[u8]) -> Result { +pub fn parse_ls_refs(body: &[u8], object_format: ObjectFormat) -> Result { let lines = pkt::data_payloads(body).map_err(|error| protocol(error.to_string()))?; lines.iter().try_fold( UpstreamRefs { + object_format, head_symref: None, refs: Vec::new(), }, @@ -133,6 +142,12 @@ pub fn parse_ls_refs(body: &[u8]) -> Result { .ok_or_else(|| protocol(format!("malformed ref line: {text}")))?; let target = Oid::from_hex(oid).map_err(|_| protocol(format!("malformed ref oid: {oid}")))?; + if target.object_id().kind() != object_format.kind() { + return Err(protocol(format!( + "ref oid {oid} doesn't match advertised object format {}", + object_format.capability() + ))); + } let mut attributes = rest.split(' '); match attributes.next() { Some("HEAD") => { @@ -228,7 +243,7 @@ pub async fn remote_refs( }, ) .await?; - parse_advertisement(&response.body)?; + let object_format = parse_advertisement(&response.body)?; let upload = endpoint(base, "git-upload-pack")?; let response = execute( @@ -241,7 +256,7 @@ pub async fn remote_refs( }, ) .await?; - parse_ls_refs(&response.body) + parse_ls_refs(&response.body, object_format) } pub async fn remote_pack( @@ -277,6 +292,7 @@ pub fn local_refs(source: &Repo, prefixes: &[&str]) -> Result| { + let mut buf = Vec::new(); + data(&mut buf, b"version 2\n"); + data(&mut buf, b"ls-refs\n"); + data(&mut buf, b"fetch\n"); + if let Some(extra) = extra { + data(&mut buf, format!("{extra}\n").as_bytes()); + } + pkt::write_flush(&mut buf).unwrap(); + parse_advertisement(&buf) + }; + assert_eq!( + advertise(Some("object-format=sha256")).unwrap(), + ObjectFormat::SHA256 + ); + assert_eq!( + advertise(None).unwrap(), + ObjectFormat::SHA1, + "protocol v2 treats an absent object-format as sha1" + ); + assert!(matches!( + advertise(Some("object-format=sha3")), + Err(FetchError::Protocol(_)) + )); let mut v0 = Vec::new(); data(&mut v0, b"# service=git-upload-pack\n"); @@ -373,7 +414,7 @@ mod tests { } #[test] - fn ls_refs_lines_parse_with_symref_and_skip_head() { + fn ls_refs_lines_parse_with_symref_skip_head_and_enforce_the_object_format() { let mut body = Vec::new(); data( &mut body, @@ -384,7 +425,7 @@ mod tests { b"95d09f2b10159347eece71399a7e2e907ea3df4f refs/heads/main\n", ); pkt::write_flush(&mut body).unwrap(); - let refs = parse_ls_refs(&body).unwrap(); + let refs = parse_ls_refs(&body, ObjectFormat::SHA1).unwrap(); assert_eq!( refs.head_symref.as_ref().map(RefName::as_str), Some("refs/heads/main") @@ -392,6 +433,24 @@ mod tests { assert_eq!(refs.refs.len(), 1); assert_eq!(refs.refs[0].name.as_str(), "refs/heads/main"); assert_eq!(refs.tips().len(), 1); + + let single = |oid: &str| { + let mut buf = Vec::new(); + data(&mut buf, format!("{oid} refs/heads/main\n").as_bytes()); + pkt::write_flush(&mut buf).unwrap(); + buf + }; + let sha1 = "95d09f2b10159347eece71399a7e2e907ea3df4f"; + let sha256 = "6ef19b41225c5369f1c104d45d8d85efa9b057b53b14b4b9b939dd74decc5321"; + assert!(parse_ls_refs(&single(sha256), ObjectFormat::SHA256).is_ok()); + assert!(matches!( + parse_ls_refs(&single(sha256), ObjectFormat::SHA1), + Err(FetchError::Protocol(_)) + )); + assert!(matches!( + parse_ls_refs(&single(sha1), ObjectFormat::SHA256), + Err(FetchError::Protocol(_)) + )); } #[test] @@ -399,7 +458,10 @@ mod tests { let mut body = Vec::new(); data(&mut body, b"zzzz refs/heads/main\n"); pkt::write_flush(&mut body).unwrap(); - assert!(matches!(parse_ls_refs(&body), Err(FetchError::Protocol(_)))); + assert!(matches!( + parse_ls_refs(&body, ObjectFormat::SHA1), + Err(FetchError::Protocol(_)) + )); } #[test] @@ -408,7 +470,7 @@ mod tests { data(&mut body, b"ERR access denied\n"); pkt::write_flush(&mut body).unwrap(); assert!(matches!( - parse_ls_refs(&body), + parse_ls_refs(&body, ObjectFormat::SHA1), Err(FetchError::Remote(message)) if message == "access denied" )); } diff --git a/knot2/crates/knot-xrpc/src/forks.rs b/knot2/crates/knot-xrpc/src/forks.rs index 0fb3dc1f..56c1adec 100644 --- a/knot2/crates/knot-xrpc/src/forks.rs +++ b/knot2/crates/knot-xrpc/src/forks.rs @@ -14,7 +14,7 @@ use knot_index::Resolved; use knot_pack::{FetchError, HaveOids, PackLimits, UpstreamRefs, WantOids}; use knot_postreceive::{Actor, Ci}; use knot_runtime::{Clock, HttpTransport}; -use knot_types::{BranchName, Oid, OwnerDid, RefName, RepoDid}; +use knot_types::{BranchName, ObjectFormat, Oid, OwnerDid, RefName, RepoDid}; use crate::body::{ForkRef, RemoteRef, RepoAtUri, RepoNameArg, Revspec, SourceUrl}; use crate::branches::resolve_at_uri; @@ -84,6 +84,31 @@ pub(crate) fn resolve_upstream( Ok(Upstream::Remote(url.clone())) } +pub(crate) struct ForkSource { + pub(crate) origin: SourceUrl, + pub(crate) upstream: Upstream, + pub(crate) refs: UpstreamRefs, +} + +impl ForkSource { + pub(crate) async fn resolve( + state: &XrpcState, + origin: &SourceUrl, + ) -> Result { + let upstream = resolve_upstream(state, origin)?; + let prefixes = ["HEAD", "refs/heads/", "refs/tags/"] + .iter() + .map(|prefix| prefix.to_string()) + .collect(); + let refs = upstream_refs(state, &upstream, prefixes).await?; + Ok(Self { + origin: origin.clone(), + upstream, + refs, + }) + } +} + fn map_fetch(error: FetchError) -> XrpcError { match error { FetchError::PackTooLarge { limit } => XrpcError::request_too_large(format!( @@ -272,6 +297,7 @@ const FORK_DENIED: &str = "only repository owner or a collaborator may operate o struct ForkState { origin: SourceUrl, haves: Vec, + object_format: ObjectFormat, } fn load_fork_state(repo: &Repo) -> Result { @@ -285,7 +311,11 @@ fn load_fork_state(repo: &Repo) -> Result { .into_iter() .map(|record| record.target) .collect(); - Ok(ForkState { origin, haves }) + Ok(ForkState { + origin, + haves, + object_format: repo.object_format(), + }) } pub(crate) struct SyncResult { @@ -339,6 +369,13 @@ async fn pull_upstream_branch( let upstream = resolve_upstream(state, &fork.origin)?; let refs = upstream_refs(state, &upstream, vec![branch.as_str().to_string()]).await?; + if refs.object_format != fork.object_format { + return Err(XrpcError::conflict(format!( + "upstream stores {} objects but this fork stores {}", + refs.object_format.capability(), + fork.object_format.capability() + ))); + } let tip = refs .find(branch) .ok_or_else(|| XrpcError::not_found("upstream repository doesn't have that branch"))?; diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index b7248ff0..e8852650 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -167,13 +167,6 @@ pub(crate) async fn create_repo( } let input: CreateInput = decode(&body)?; - let source = input - .source - .as_ref() - .map(|source| { - crate::forks::resolve_upstream(&state, source).map(|upstream| (source, upstream)) - }) - .transpose()?; let head = input.default_branch.map(|branch| branch.head_ref()); let owner = OwnerDid::new(actor.as_str()).expect("account DID is always a valid owner DID"); @@ -195,6 +188,12 @@ pub(crate) async fn create_repo( Resolved::Ready(None) => {} } + let fork = match &input.source { + Some(origin) => Some(crate::forks::ForkSource::resolve(&state, origin).await?), + None => None, + }; + let object_format = fork.as_ref().map(|fork| fork.refs.object_format); + let (repo_did, provisioning) = provision_repo_did(&state, &actor, &input.rkey, input.repo_did).await?; let now = state.now(); @@ -214,7 +213,11 @@ pub(crate) async fn create_repo( let placed = repo_did.clone(); let lfs = lfs_store.clone(); let provisioning = run_blocking(move || { - let repo = layout.create(&placed).map_err(|error| match error { + let repo = match object_format { + Some(format) => layout.create_with_format(&placed, format), + None => layout.create(&placed), + } + .map_err(|error| match error { GitError::AlreadyExists(_) => XrpcError::conflict("repository already exists on disk"), GitError::ReservedDid(_) => { XrpcError::invalid_request("repoDid mustn't be knot's own identity") @@ -229,23 +232,21 @@ pub(crate) async fn create_repo( }) .await?; - let lfs_missing = match &source { - Some((source_url, upstream)) => { - match populate(&state, &repo_did, source_url, upstream).await { - Ok(missing) => missing, - Err(error) => { - let layout = state.layout.clone(); - let placed = repo_did.clone(); - let lfs = lfs_store.clone(); - let _ = run_blocking(move || { - rollback_local(&layout, lfs.as_deref(), &placed); - Ok(()) - }) - .await; - return Err(error); - } + let lfs_missing = match &fork { + Some(fork) => match populate(&state, &repo_did, fork).await { + Ok(missing) => missing, + Err(error) => { + let layout = state.layout.clone(); + let placed = repo_did.clone(); + let lfs = lfs_store.clone(); + let _ = run_blocking(move || { + rollback_local(&layout, lfs.as_deref(), &placed); + Ok(()) + }) + .await; + return Err(error); } - } + }, None => Vec::new(), }; @@ -333,26 +334,20 @@ pub(crate) async fn create_repo( async fn populate( state: &Arc>, repo_did: &RepoDid, - source: &SourceUrl, - upstream: &crate::forks::Upstream, + fork: &crate::forks::ForkSource, ) -> Result, XrpcError> { - let prefixes = vec![ - "HEAD".to_string(), - "refs/heads/".to_string(), - "refs/tags/".to_string(), - ]; - let refs = crate::forks::upstream_refs(state, upstream, prefixes).await?; - let tips = refs.tips(); + let tips = fork.refs.tips(); let pack = crate::forks::upstream_pack( state, - upstream, - knot_pack::WantOids::new(refs.tips()), + &fork.upstream, + knot_pack::WantOids::new(tips.clone()), knot_pack::HaveOids::default(), ) .await?; let layout = state.layout.clone(); let placed = repo_did.clone(); - let origin = source.clone(); + let origin = fork.origin.clone(); + let refs = fork.refs.clone(); run_blocking(move || { let repo = layout.open(&placed)?; crate::forks::populate_fork(&repo, &refs, &pack, &origin) @@ -360,7 +355,7 @@ async fn populate( .await?; crate::lfs::mirror_fork_objects( Arc::clone(state), - upstream.clone(), + fork.upstream.clone(), repo_did.clone(), knot_pack::WantOids::new(tips), knot_pack::HaveOids::default(), diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index 55344707..f7d256c9 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -267,11 +267,18 @@ fn only_git_event(world: &World) -> std::sync::Arc { events.remove(0) } -fn bootstrap(dir: &TempDir, rebuild: bool) -> (Layout, Arc, PathBuf) { +fn bootstrap( + dir: &TempDir, + rebuild: bool, + object_format: knot_types::ObjectFormat, +) -> (Layout, Arc, PathBuf) { let scan_path = dir.path().join("repos"); std::fs::create_dir_all(&scan_path).unwrap(); let knot = knot_did(); - let layout = Layout::new(&scan_path).reserving_meta(&knot).unwrap(); + let layout = Layout::new(&scan_path) + .with_object_format(object_format) + .reserving_meta(&knot) + .unwrap(); layout.bootstrap_meta(&knot).unwrap(); let meta_path = layout.meta_path(&knot).unwrap(); let index = Arc::new(Index::new(meta_path.clone(), layout.clone())); @@ -425,23 +432,56 @@ struct World { impl World { fn new() -> Self { - Self::build(256, 256, AdmissionPolicy::Closed, None) + Self::build( + 256, + 256, + AdmissionPolicy::Closed, + None, + knot_types::ObjectFormat::SHA1, + ) } fn open() -> Self { - Self::build(256, 256, AdmissionPolicy::Open, None) + Self::build( + 256, + 256, + AdmissionPolicy::Open, + None, + knot_types::ObjectFormat::SHA1, + ) } fn with_pending_limit(limit: usize) -> Self { - Self::build(limit, limit, AdmissionPolicy::Closed, None) + Self::build( + limit, + limit, + AdmissionPolicy::Closed, + None, + knot_types::ObjectFormat::SHA1, + ) } fn with_limits(global: usize, per_actor: usize) -> Self { - Self::build(global, per_actor, AdmissionPolicy::Closed, None) + Self::build( + global, + per_actor, + AdmissionPolicy::Closed, + None, + knot_types::ObjectFormat::SHA1, + ) } - fn with_git_http(git_http: Arc) -> Self { - Self::build(256, 256, AdmissionPolicy::Closed, Some(git_http)) + fn with_git_http( + git_http: Arc, + object_format: knot_types::ObjectFormat, + ) -> Self { + Self::build( + 256, + 256, + AdmissionPolicy::Closed, + Some(git_http), + object_format, + ) } fn build( @@ -449,9 +489,10 @@ impl World { per_actor: usize, admission: AdmissionPolicy, git_http: Option>, + object_format: knot_types::ObjectFormat, ) -> Self { let dir = tempfile::tempdir().unwrap(); - let (layout, index, meta_path) = bootstrap(&dir, true); + let (layout, index, meta_path) = bootstrap(&dir, true, object_format); let admin = signer(1); let member = signer(2); let stranger = signer(3); @@ -514,7 +555,7 @@ impl World { fn build_state(responder: Responder, rebuild: bool) -> (TempDir, SharedState) { let dir = tempfile::tempdir().unwrap(); - let (layout, index, meta_path) = bootstrap(&dir, rebuild); + let (layout, index, meta_path) = bootstrap(&dir, rebuild, knot_types::ObjectFormat::SHA1); let state = state_from( &dir, (layout, index, meta_path), @@ -2634,7 +2675,17 @@ mod fork_endpoints { use knot_runtime::HttpTransport; use knot_types::{Oid, RefName, UnixSeconds}; - const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; + const EMPTY_TREE_SHA1: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; + const EMPTY_TREE_SHA256: &str = + "6ef19b41225c5369f1c104d45d8d85efa9b057b53b14b4b9b939dd74decc5321"; + + fn empty_tree(repo: &Repo) -> Oid { + let hex = match repo.object_format() { + knot_types::ObjectFormat::SHA256 => EMPTY_TREE_SHA256, + _ => EMPTY_TREE_SHA1, + }; + Oid::from_hex(hex).unwrap() + } fn ident(time: i64) -> Identity { Identity { @@ -2648,7 +2699,7 @@ mod fork_endpoints { fn put_commit(repo: &Repo, parent: Option, path: &str, content: &str, time: i64) -> Oid { let base_tree = match parent { Some(parent) => repo.find_commit(parent).unwrap().tree, - None => Oid::from_hex(EMPTY_TREE).unwrap(), + None => empty_tree(repo), }; let staged = vec![StagedChange { path: knot_types::RepoPath::new(path).unwrap(), @@ -2991,21 +3042,20 @@ mod fork_endpoints { } #[tokio::test] - async fn a_fork_from_a_remote_knot_fetches_over_http() { + async fn a_fork_over_http_takes_the_upstream_object_format_and_conflicts_when_it_changes() { let upstream_dir = tempfile::tempdir().unwrap(); let upstream_path = upstream_dir.path().join("uni.git"); - let upstream = Repo::create(&upstream_path).unwrap(); - upstream - .set_head(&RefName::new("refs/heads/main").unwrap()) - .unwrap(); + let upstream = + Repo::create_with_format(&upstream_path, knot_types::ObjectFormat::SHA1).unwrap(); + upstream.set_head(&main_ref()).unwrap(); advance(&upstream, &main_ref(), "reef.txt", "kelp forest\n", 1_000); let tip = advance(&upstream, &main_ref(), "tide.txt", "rock pool\n", 1_001); - let served = upstream_path.clone(); + let served = Arc::new(std::sync::RwLock::new(upstream_path.clone())); + let path = Arc::clone(&served); let git_http: Arc = Arc::new(FakeHttp::new(move |request: &HttpRequest| { - let _keep = &upstream_dir; - let repo = Repo::open(&served).unwrap(); + let repo = Repo::open(path.read().unwrap().clone()).unwrap(); let body = if request.url.path().ends_with("/info/refs") { knot_pack::advertise_upload(&repo).unwrap() } else { @@ -3019,13 +3069,35 @@ mod fork_endpoints { }) })); - let world = World::with_git_http(git_http); + let world = World::with_git_http(git_http, knot_types::ObjectFormat::SHA256); add_member_helper(&world).await; + let plain = create_repo_helper(&world, "kelp").await; + assert_eq!( + world.layout.open(&plain).unwrap().object_format(), + knot_types::ObjectFormat::SHA256, + "a repo with no upstream still uses this knot's configured format" + ); + let remote = "https://barnacle.nel.pet/did:plc:squid/uni"; let fork_did = fork_repo(&world, remote, "uni").await; let fork = world.layout.open(&fork_did).unwrap(); + assert_eq!( + fork.object_format(), + knot_types::ObjectFormat::SHA1, + "a fork of a sha1 upstream must be sha1 so the upstream's objects ingest" + ); assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(tip)); assert_eq!(fork.origin_url().as_deref(), Some(remote)); + assert_eq!( + fork.read_blob( + fork.entry_at(tip, &knot_types::RepoPath::new("tide.txt").unwrap()) + .unwrap() + .unwrap() + .oid + ) + .unwrap(), + b"rock pool\n" + ); let new_tip = advance( &Repo::open(&upstream_path).unwrap(), @@ -3038,8 +3110,26 @@ mod fork_endpoints { sync_fork(&world, &world.member, MEMBER_HOST, "main").await, StatusCode::OK ); - let fork = world.layout.open(&fork_did).unwrap(); - assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(new_tip)); + assert_eq!( + world + .layout + .open(&fork_did) + .unwrap() + .find_ref(&main_ref()) + .unwrap(), + Some(new_tip) + ); + + let replaced = upstream_dir.path().join("replaced.git"); + let sha256 = Repo::create_with_format(&replaced, knot_types::ObjectFormat::SHA256).unwrap(); + sha256.set_head(&main_ref()).unwrap(); + advance(&sha256, &main_ref(), "reef.txt", "kelp forest\n", 1_000); + *served.write().unwrap() = replaced; + assert_eq!( + sync_fork(&world, &world.member, MEMBER_HOST, "main").await, + StatusCode::CONFLICT, + "the fork reports a format mismatch as a conflict" + ); } }