From b51522019c2693c0261884909eddf7ba216ecd04 Mon Sep 17 00:00:00 2001 From: hmirin Date: Thu, 29 Oct 2026 01:01:51 +0900 Subject: [PATCH] Use older valid carrier copies during recovery and re-import packs after a prune - Carrier history reads stopped at the first corrupt or missing copy of an object, even when an older carrier commit still held a valid copy, so recovery failed with a content-id mismatch. recovery_history and restore_historical_object now try each copy newest to oldest within the existing commit budget and use the first whose content hash matches. Bytes read from rejected copies count toward the byte budget. - After a user deleted a remote-tracking ref and ran gc --prune, every later fetch of that ref failed, because the import trusted the recorded pack list and skipped generations it believed were already imported. When the incremental connectivity check fails, the helper now re-imports every pack of the authenticated chain and checks the full graph. The pin is validated first and written only after that check passes, so it never moves backwards. --- docs/spec.md | 10 +++ src/repository.rs | 48 ++++++++--- src/storage.rs | 178 ++++++++++++++++++++++------------------- tests/carrier_git.rs | 69 ++++++++++++++++ tests/end_to_end.rs | 13 +-- tests/remote_helper.rs | 85 ++++++++++++++++++++ 6 files changed, 300 insertions(+), 103 deletions(-) diff --git a/docs/spec.md b/docs/spec.md index 5595b03..95795e3 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -460,6 +460,12 @@ inductive check of only the newly reachable range establishes completeness of the new state. The frontier advances only after pack authentication, import, and the incremental connectivity check all succeed. +Imported pack IDs do not prove that their objects remain in the local object +database; deleting a remote-tracking ref and pruning removes them. When the +incremental check fails, the client MUST re-import every pack of the +authenticated chain and run a full connectivity walk before failing. The +re-import does not move the floor; only a successful full check advances it. + A missing frontier in a v5 client state requires a full connectivity walk. A client state from another wire version is not accepted. The incremental rule assumes the trusted client's existing object database has not been corrupted outside this @@ -482,6 +488,10 @@ publication is a single-parent fast-forward, and a merge can hide a legitimate state behind a non-first parent. Discovery MUST deduplicate manifest IDs and validate candidates against the authenticated policy and predecessor chain. The byte, outer-commit, candidate, and cryptographic-operation budgets MUST be fixed. Exhaustion MUST fail closed. +Reading a historical manifest or restoring historical ciphertext MUST try each +stored copy in that bounded history newest first and use the first whose +content ID matches, so a corrupt copy cannot hide an older valid one. Bytes +read from rejected copies count against the byte budget. Discovery MUST report authenticated descendants and replays. If authenticated descendants conflict, recovery MUST refuse to choose a fork. A replayed older manifest MUST NOT be treated as a newer candidate. Directory storage has no diff --git a/src/repository.rs b/src/repository.rs index 7f040be..6f3bb22 100644 --- a/src/repository.rs +++ b/src/repository.rs @@ -733,10 +733,42 @@ impl EncryptedRepository { let state = read_client_state(state_path)?; self.validate_pinned_head(&chain, &state)?; - let stop_generation = state.imported_generation; let mut imported: HashSet = state.packs.into_iter().collect(); + self.import_chain_packs(repo, &chain, state.imported_generation, &mut imported)?; + let connectivity_timer = trace::Span::new("inner_ref_connectivity_check"); + if git::ensure_refs_connected_since(repo, ¤t.manifest.refs, &state.verified_refs) + .is_err() + { + // Imported pack IDs are cache state, not proof that their objects + // are still local: deleting a remote-tracking ref and pruning + // removes them. Re-import the whole authenticated chain and check + // the full graph; the pinned floor is unchanged until that passes. + imported.clear(); + self.import_chain_packs(repo, &chain, None, &mut imported)?; + git::ensure_refs_connected_since(repo, ¤t.manifest.refs, &BTreeMap::new())?; + } + drop(connectivity_timer); + write_client_state_locked( + state_path, + &imported, + ¤t.id, + ¤t.manifest, + ¤t.manifest.refs, + )?; + Ok(current.manifest) + } + + /// Imports, oldest first, the packs of every chain entry newer than + /// `after_generation`, recording each pack ID in `imported`. + fn import_chain_packs( + &self, + repo: &Path, + chain: &[OpenedState], + after_generation: Option, + imported: &mut HashSet, + ) -> Result<()> { for entry in chain.iter().rev() { - if stop_generation.is_some_and(|generation| entry.manifest.generation <= generation) { + if after_generation.is_some_and(|generation| entry.manifest.generation <= generation) { continue; } for descriptor in &entry.manifest.new_packs { @@ -773,17 +805,7 @@ impl EncryptedRepository { imported.insert(descriptor.id.clone()); } } - let connectivity_timer = trace::Span::new("inner_ref_connectivity_check"); - git::ensure_refs_connected_since(repo, ¤t.manifest.refs, &state.verified_refs)?; - drop(connectivity_timer); - write_client_state_locked( - state_path, - &imported, - ¤t.id, - ¤t.manifest, - ¤t.manifest.refs, - )?; - Ok(current.manifest) + Ok(()) } fn validate_remote_continuity(&self, state_path: &Path, chain: &[OpenedState]) -> Result<()> { diff --git a/src/storage.rs b/src/storage.rs index f4dd09d..50cefc6 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -12,7 +12,6 @@ use rand::rngs::OsRng; use sha2::{Digest, Sha256}; use thiserror::Error; -use crate::crypto::object_id; use crate::error::{BudgetExhausted, ObjectUnavailable}; use crate::trace; @@ -1125,55 +1124,50 @@ impl GitStorage { carrier_chunk_entries(self.root(), commit, kind, id) } - fn object_at_commit( + /// Commits in `window` whose tree changed the stored copy of `id`, newest + /// first. Carrier history is linear, so each one starts a distinct copy. + fn copy_commits(&self, window: &[String], kind: ObjectKind, id: &str) -> Result> { + validate_id(id)?; + let Some(newest) = window.first() else { + return Ok(Vec::new()); + }; + let directory = format!("{}/{}/{id}", kind.directory(), &id[..2]); + let output = carrier_git_command() + .arg("-C") + .arg(self.root()) + .args([ + "rev-list", + &format!("--max-count={}", window.len()), + newest, + "--", + &directory, + ]) + .output()?; + ensure_git_success(&output, "git rev-list carrier object copies")?; + let window: HashSet<&str> = window.iter().map(String::as_str).collect(); + Ok(String::from_utf8(output.stdout)? + .lines() + .filter(|commit| window.contains(commit)) + .map(ToOwned::to_owned) + .collect()) + } + + /// Streams the copy of `id` stored at `commit` into `sink` chunk by chunk. + /// Returns false when the commit holds no copy or the copy is malformed or + /// does not hash to `id`; `sink` errors and process failures propagate. + fn read_copy_at( &self, commit: &str, kind: ObjectKind, id: &str, - max_bytes: u64, - ) -> Result>> { - let Some(chunks) = self.chunk_entries_at(commit, kind, id)? else { - return Ok(None); - }; - let mut bytes = Vec::new(); - for (index, (path, _)) in chunks.iter().enumerate() { - let output = carrier_git_command() - .arg("-C") - .arg(self.root()) - .args(["show", &format!("{commit}:{path}")]) - .output()?; - ensure_git_success(&output, "git show historical carrier object")?; - let length = output.stdout.len(); - if (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE) - || (index + 1 == chunks.len() && (length == 0 || length > CARRIER_CHUNK_SIZE)) - { - bail!("historical carrier object chunk has invalid size") - } - let next_length = (bytes.len() as u64) - .checked_add(length as u64) - .context("historical carrier object size overflow")?; - if next_length > max_bytes { - bail!(BudgetExhausted( - "carrier recovery manifest byte budget exhausted".to_owned() - )) - } - bytes.extend_from_slice(&output.stdout); - } - if object_id(&bytes) != id { - bail!("historical carrier object content id mismatch") - } - Ok(Some(bytes)) - } - - fn restore_object_at(&self, commit: &str, kind: ObjectKind, id: &str) -> Result { - let Some(chunks) = self.chunk_entries_at(commit, kind, id)? else { - return Ok(false); + mut sink: impl FnMut(usize, &[u8]) -> Result<()>, + ) -> Result { + let chunks = match self.chunk_entries_at(commit, kind, id) { + Ok(Some(chunks)) => chunks, + Ok(None) => return Ok(false), + Err(error) if error.is::() => return Err(error), + Err(_) => return Ok(false), }; - let staging = tempfile::Builder::new() - .prefix("git-remote-e2ee-restore-") - .tempdir_in(self.root())?; - let object_dir = staging.path().join("object"); - fs::create_dir(&object_dir)?; let mut hasher = Sha256::new(); for (index, (path, _)) in chunks.iter().enumerate() { let output = carrier_git_command() @@ -1181,33 +1175,17 @@ impl GitStorage { .arg(self.root()) .args(["show", &format!("{commit}:{path}")]) .output()?; - ensure_git_success(&output, "git show historical carrier object")?; let length = output.stdout.len(); - if (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE) + if !output.status.success() + || (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE) || (index + 1 == chunks.len() && (length == 0 || length > CARRIER_CHUNK_SIZE)) { - bail!("historical carrier object chunk has invalid size") + return Ok(false); } hasher.update(&output.stdout); - let chunk_path = object_dir.join(format!("{index:08}")); - let mut file = File::create(chunk_path)?; - file.write_all(&output.stdout)?; - file.sync_all()?; - } - if hex::encode(hasher.finalize()) != id { - bail!("historical carrier object content id mismatch") - } - let target = self.object_directory(kind, id)?; - if target.exists() { - fs::remove_dir_all(&target)?; + sink(index, &output.stdout)?; } - fs::create_dir_all( - target - .parent() - .context("carrier object path has no parent")?, - )?; - fs::rename(object_dir, target)?; - Ok(true) + Ok(hex::encode(hasher.finalize()) == id) } fn publish_checkout_head( @@ -1420,10 +1398,11 @@ impl Storage for GitStorage { )) } let mut history = RecoveryHistory::default(); + // Every byte read counts, including corrupt copies that are skipped. let mut total_bytes = 0_u64; - for commit_id in commits { + for commit_id in &commits { let head_id = self - .head_bytes_at_commit(&commit_id)? + .head_bytes_at_commit(commit_id)? .as_deref() .and_then(|bytes| { std::str::from_utf8(bytes) @@ -1435,22 +1414,31 @@ impl Storage for GitStorage { if let Some(id) = &head_id && !history.manifests.contains_key(id) { - let remaining = max_manifest_bytes.saturating_sub(total_bytes); - if let Some(bytes) = - self.object_at_commit(&commit_id, ObjectKind::Manifest, id, remaining)? - { - total_bytes = total_bytes - .checked_add(bytes.len() as u64) - .context("carrier recovery manifest byte count overflow")?; - if total_bytes > max_manifest_bytes { - bail!(BudgetExhausted( - "carrier recovery manifest byte budget exhausted".to_owned() - )) + for copy in self.copy_commits(&commits, ObjectKind::Manifest, id)? { + let mut bytes = Vec::new(); + let valid = + self.read_copy_at(©, ObjectKind::Manifest, id, |_, chunk| { + total_bytes = total_bytes + .checked_add(chunk.len() as u64) + .context("carrier recovery manifest byte count overflow")?; + if total_bytes > max_manifest_bytes { + bail!(BudgetExhausted( + "carrier recovery manifest byte budget exhausted".to_owned() + )) + } + bytes.extend_from_slice(chunk); + Ok(()) + })?; + if valid { + history.manifests.insert(id.clone(), bytes); + break; } - history.manifests.insert(id.clone(), bytes); } } - history.commits.push(RecoveryCommit { commit_id, head_id }); + history.commits.push(RecoveryCommit { + commit_id: commit_id.clone(), + head_id, + }); } Ok(Some(history)) } @@ -1462,10 +1450,32 @@ impl Storage for GitStorage { "carrier recovery outer-commit budget exhausted".to_owned() )) } - for commit in commits { - if self.restore_object_at(&commit, kind, id)? { - return Ok(true); + for copy in self.copy_commits(&commits, kind, id)? { + let staging = tempfile::Builder::new() + .prefix("git-remote-e2ee-restore-") + .tempdir_in(self.root())?; + let object_dir = staging.path().join("object"); + fs::create_dir(&object_dir)?; + let valid = self.read_copy_at(©, kind, id, |index, chunk| { + let mut file = File::create(object_dir.join(format!("{index:08}")))?; + file.write_all(chunk)?; + file.sync_all()?; + Ok(()) + })?; + if !valid { + continue; } + let target = self.object_directory(kind, id)?; + if target.exists() { + fs::remove_dir_all(&target)?; + } + fs::create_dir_all( + target + .parent() + .context("carrier object path has no parent")?, + )?; + fs::rename(object_dir, target)?; + return Ok(true); } Ok(false) } diff --git a/tests/carrier_git.rs b/tests/carrier_git.rs index b9a272e..da63cfa 100644 --- a/tests/carrier_git.rs +++ b/tests/carrier_git.rs @@ -1958,3 +1958,72 @@ fn carrier_recovery_refuses_to_skip_a_historical_head_with_missing_objects() { let carrier_head = git(&carrier, &["show", "refs/heads/main:HEAD"], false); assert_eq!(carrier_head, junk_id); } + +#[test] +fn carrier_history_reads_skip_corrupt_copies_for_an_older_valid_one() { + let temporary = tempfile::tempdir().unwrap(); + let _cache = isolated_cache(&temporary); + let carrier = temporary.path().join("carrier.git"); + let source = temporary.path().join("source"); + init_source(&source, "content"); + + let owner = KeyFile::generate(); + init_carrier(temporary.path(), &carrier, &owner); + let repository = EncryptedRepository::new( + GitStorage::open(carrier.to_str().unwrap()).unwrap(), + owner.clone(), + ); + repository + .push_ref(&source, "refs/heads/main", false) + .unwrap(); + let (head_id, head) = repository.current_manifest().unwrap(); + let pack_id = head.new_packs[0].id.clone(); + let original = GitStorage::open(carrier.to_str().unwrap()).unwrap(); + let manifest_bytes = original.get_object(ObjectKind::Manifest, &head_id).unwrap(); + let pack_bytes = original.get_object(ObjectKind::Pack, &pack_id).unwrap(); + + edit_carrier( + &carrier, + &temporary.path().join("carrier-edit"), + |workdir| { + for (directory, id) in [("manifests", &head_id), ("packs", &pack_id)] { + let chunk = workdir + .join(directory) + .join(&id[..2]) + .join(id) + .join("00000000"); + fs::write(chunk, b"corrupt copy").unwrap(); + } + }, + ); + + let storage = GitStorage::open(carrier.to_str().unwrap()).unwrap(); + assert_ne!( + storage.get_object(ObjectKind::Pack, &pack_id).unwrap(), + pack_bytes + ); + let history = storage + .recovery_history(2048, 64 * 1024 * 1024) + .unwrap() + .unwrap(); + assert_eq!( + history.commits[0].head_id.as_deref(), + Some(head_id.as_str()) + ); + assert_eq!(history.manifests[&head_id], manifest_bytes); + + assert!( + storage + .restore_historical_object(ObjectKind::Pack, &pack_id) + .unwrap() + ); + assert_eq!( + storage.get_object(ObjectKind::Pack, &pack_id).unwrap(), + pack_bytes + ); + assert!( + !storage + .restore_historical_object(ObjectKind::Pack, &"0".repeat(64)) + .unwrap() + ); +} diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs index 62d7116..406aa56 100644 --- a/tests/end_to_end.rs +++ b/tests/end_to_end.rs @@ -445,7 +445,7 @@ fn returning_client_rejects_signed_ref_advance_without_its_pack() { } #[test] -fn missing_verified_frontier_tip_fails_closed() { +fn missing_verified_frontier_tip_forces_full_reverification() { let temporary = tempfile::tempdir().unwrap(); let source = temporary.path().join("source"); let destination = temporary.path().join("destination"); @@ -468,11 +468,12 @@ fn missing_verified_frontier_tip_fails_closed() { state["verified_refs"]["refs/heads/main"] = serde_json::Value::String("1".repeat(40)); fs::write(&state_path, serde_json::to_vec_pretty(&state).unwrap()).unwrap(); - let error = repository - .fetch_into(&destination, "e2ee") - .unwrap_err() - .to_string(); - assert!(error.contains("verified ref refs/heads/main is missing locally")); + let manifest = repository.fetch_into(&destination, "e2ee").unwrap(); + let state: serde_json::Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + assert_eq!( + state["verified_refs"]["refs/heads/main"], + manifest.refs["refs/heads/main"].as_str() + ); } #[test] diff --git a/tests/remote_helper.rs b/tests/remote_helper.rs index 1da7830..cf287d2 100644 --- a/tests/remote_helper.rs +++ b/tests/remote_helper.rs @@ -505,3 +505,88 @@ fn native_git_fetch_pins_branch_deletions_that_transfer_no_objects() { String::from_utf8_lossy(&output.stderr) ); } + +#[test] +fn native_git_fetch_reimports_packs_after_local_objects_are_pruned() { + let temporary = tempfile::tempdir().unwrap(); + let source = temporary.path().join("source"); + let destination = temporary.path().join("destination"); + let remote_path = temporary.path().join("remote"); + let key_path = temporary.path().join("repository.key.json"); + initialize_git(&source); + initialize_git(&destination); + let key = KeyFile::generate(); + key.write_new(&key_path).unwrap(); + EncryptedRepository::new(FilesystemStorage::new(&remote_path), key) + .initialize("owner") + .unwrap(); + configure_remote(&source, &remote_path, &key_path); + configure_remote(&destination, &remote_path, &key_path); + + let helper_git = |repo: &Path, args: &[&str]| { + let helper = Path::new(env!("CARGO_BIN_EXE_git-remote-e2ee")); + let mut paths = vec![helper.parent().unwrap().to_path_buf()]; + paths.extend(env::split_paths(&env::var_os("PATH").unwrap_or_default())); + let output = Command::new("git") + .arg("-C") + .arg(repo) + .args(args) + .env("PATH", env::join_paths(paths).unwrap()) + .env("GIT_REMOTE_E2EE_CACHE_DIR", temporary.path().join("cache")) + .env("XDG_STATE_HOME", temporary.path().join("state")) + .env("XDG_CONFIG_HOME", temporary.path().join("config")) + .output() + .unwrap(); + assert!( + output.status.success(), + "git {args:?}: stderr={}", + String::from_utf8_lossy(&output.stderr) + ); + }; + let mut heads = Vec::new(); + for (branch, content) in [("main", "first"), ("feature", "second")] { + git(&source, &["checkout", "-q", "-B", branch], false); + fs::write(source.join("note.md"), format!("{content}\n")).unwrap(); + git(&source, &["add", "note.md"], false); + git(&source, &["commit", "-q", "-m", content], false); + heads.push(git(&source, &["rev-parse", "HEAD"], false)); + helper_git(&source, &["push", "-q", "private", branch]); + } + + helper_git(&destination, &["fetch", "-q", "private"]); + git( + &destination, + &["update-ref", "-d", "refs/remotes/private/feature"], + false, + ); + git(&destination, &["gc", "-q", "--prune=now"], false); + assert!( + !git_output(&destination, &["cat-file", "-e", &heads[1]], false) + .status + .success(), + "gc pruned the feature commit" + ); + + helper_git(&destination, &["fetch", "-q", "private"]); + assert_eq!( + git( + &destination, + &["rev-parse", "refs/remotes/private/feature"], + false + ), + heads[1] + ); + git( + &destination, + &["fsck", "--connectivity-only", "--no-dangling"], + false, + ); + assert_eq!( + git( + &destination, + &["show", "refs/remotes/private/feature:note.md"], + false + ), + "second" + ); +}