diff --git a/crates/crab-auth-server/README.md b/crates/crab-auth-server/README.md index 950e7ea74..05d1223a8 100644 --- a/crates/crab-auth-server/README.md +++ b/crates/crab-auth-server/README.md @@ -35,6 +35,20 @@ ref updates, staged Xet/Git objects, manifests, indexes, receipts, and active-active options. Its `Prepare`, `Verify`, and `Commit` commands make the publish boundary explicit; commit requires the verified plan digest. +Capsule receive captures source and filtered Git/catalog views separately. +Before source materialization, scoped candidates are verified using only the +filtered view and incoming packs. Newly reachable Git objects must be visible +under that capture or supplied by the candidate; retained physical objects do +not grant access. Both receive paths share the graph/visibility verifier. +Receive then queries pointer files and candidate shard/xorb identities under +each capture and applies the candidate delta. +File-only deltas can reference historical filtered artifacts; source-first +file selection and immutable descriptor conflicts remain enforced. Prepare +reads only ref authority. Git workspace intake combines captured packs and +candidate packs under the existing byte and external-delta checks, including +checkpoint-backed views. Physical shard/xorb verification and copy-before-ref +publication remain unchanged. + `crab-auth-view` materializes a filtered view using repeated `--read-path` and `--deny-path` rules, a scope hash, and a source repository URL. It builds the view's Git workspace and Crab metadata and reports the resulting prefixes, diff --git a/crates/crab-auth-server/src/receive/capsule.rs b/crates/crab-auth-server/src/receive/capsule.rs index f8bfc37cf..3a6475ac3 100644 --- a/crates/crab-auth-server/src/receive/capsule.rs +++ b/crates/crab-auth-server/src/receive/capsule.rs @@ -7,7 +7,8 @@ use crab_coordination::write_coordinator::{CommitOutcome, CoordinatedRefUpdate}; use crab_metadata::capsule_protocol::{Capsule, CapsuleRun, CapsuleSection, CapsuleSectionKind}; use super::capsule_dependencies::{ - DependencyCopy, promote_dependency_copies, source_pointer_delta, verify_pointer_dependencies, + DependencyCopy, promote_dependency_copies, select_pointer_catalog, source_pointer_delta, + verify_pointer_dependencies, }; use super::git_workspace::{materialize_capsule_source_push, verify_capsule_git_candidate}; use super::{ @@ -331,8 +332,10 @@ pub(super) async fn verify_capsule_candidate( .source_root_digest .as_deref() .ok_or_else(|| invalid("capsule prepare record is missing its root digest"))?; - let source_view = crab_read::capsule_protocol::open_view( + let source_root = crab_metadata::capsule_protocol::load_root(ctx.router()).await?; + let source_view = crab_read::capsule_protocol::open_git_view_from_root( ctx.router(), + source_root, crab_read::capsule_protocol::CapsuleReadLimits { max_capsule_bytes: MAX_PROTECTED_CAPSULE_BYTES, max_frontier_bytes: MAX_PROTECTED_CAPSULE_BYTES, @@ -354,16 +357,13 @@ pub(super) async fn verify_capsule_candidate( .ok_or_else(|| invalid("protected capsule run is empty"))?; validate_capsule_transaction(plan, candidate)?; + let candidate_delta = candidate.pointer_catalog_delta()?.unwrap_or_default(); let (changed_paths, publication, replication_objects, dependency_copies) = match prepare.view_scope.as_ref() { None => { if prepared_root != plan.base_root_digest { return Err(conflict("capsule base root differs from prepare record")); } - let mut catalog = source_view.pointer_catalog().await?; - if let Some(delta) = candidate.pointer_catalog_delta()? { - catalog.apply(&delta)?; - } let (changed_paths, pointers) = verify_capsule_git_candidate( ctx.store(), ctx.router(), @@ -373,6 +373,15 @@ pub(super) async fn verify_capsule_candidate( MAX_PROTECTED_CAPSULE_BYTES, ) .await?; + let mut catalog = select_pointer_catalog( + &source_view, + ctx.router(), + &pointers, + &candidate_delta, + MAX_PROTECTED_CAPSULE_BYTES, + ) + .await?; + catalog.apply(&candidate_delta)?; let (objects, copies) = verify_pointer_dependencies(ctx, plan, &catalog, &pointers, None).await?; (changed_paths, candidate.clone(), objects, copies) @@ -383,8 +392,11 @@ pub(super) async fn verify_capsule_candidate( scope.repo_prefix.clone(), scope.global_prefix.clone(), ); - let filtered_view = crab_read::capsule_protocol::open_view( + let filtered_root = + crab_metadata::capsule_protocol::load_root(&filtered_router).await?; + let filtered_view = crab_read::capsule_protocol::open_git_view_from_root( &filtered_router, + filtered_root, crab_read::capsule_protocol::CapsuleReadLimits { max_capsule_bytes: MAX_PROTECTED_CAPSULE_BYTES, max_frontier_bytes: MAX_PROTECTED_CAPSULE_BYTES, @@ -395,21 +407,36 @@ pub(super) async fn verify_capsule_candidate( return Err(conflict("filtered capsule root changed since prepare")); } validate_ref_heads(&prepare.view_ref_updates, filtered_view.refs())?; - let candidate_delta = candidate.pointer_catalog_delta()?.unwrap_or_default(); - let mut candidate_catalog = filtered_view.pointer_catalog().await?; - candidate_catalog.apply(&candidate_delta)?; let materialized = materialize_capsule_source_push( ctx.router(), &source_view, &filtered_view, + &filtered_router, candidate, &plan.ref_updates, &prepare.source_ref_updates, - &plan.transaction_id, MAX_PROTECTED_CAPSULE_BYTES, ) .await?; - let source_catalog = source_view.pointer_catalog().await?; + // Source and filtered captures have different authority and namespaces. + // Select each independently; source-first file policy is applied below. + let mut candidate_catalog = select_pointer_catalog( + &filtered_view, + &filtered_router, + &materialized.pointers, + &candidate_delta, + MAX_PROTECTED_CAPSULE_BYTES, + ) + .await?; + candidate_catalog.apply(&candidate_delta)?; + let source_catalog = select_pointer_catalog( + &source_view, + ctx.router(), + &materialized.pointers, + &candidate_catalog, + MAX_PROTECTED_CAPSULE_BYTES, + ) + .await?; let delta = source_pointer_delta( &source_catalog, &candidate_catalog, diff --git a/crates/crab-auth-server/src/receive/capsule_dependencies.rs b/crates/crab-auth-server/src/receive/capsule_dependencies.rs index a8f6697f0..a3d692b3e 100644 --- a/crates/crab-auth-server/src/receive/capsule_dependencies.rs +++ b/crates/crab-auth-server/src/receive/capsule_dependencies.rs @@ -22,6 +22,56 @@ pub(super) struct DependencyCopy { digest: String, } +pub(super) async fn select_pointer_catalog( + view: &crab_read::capsule_protocol::CapsuleRepositoryView, + layout: &crab_storage::StoreLayout, + pointers: &ReachablePointerScan, + candidate: &PointerCatalog, + maximum: u64, +) -> Result { + let files = pointers + .crab_pointers + .iter() + .map(|pointer| MerkleHash::from(pointer.file_hash).hex()) + .chain(candidate.files().keys().cloned()) + .collect::>() + .into_iter() + .collect::>(); + let shards = candidate + .shards() + .keys() + .cloned() + .chain( + candidate + .files() + .values() + .map(|file| file.shard_hash().to_owned()), + ) + .collect::>() + .into_iter() + .collect::>(); + let xorbs = candidate + .xorbs() + .keys() + .cloned() + .chain( + candidate + .shards() + .values() + .flat_map(|shard| shard.xorb_hashes().iter().cloned()), + ) + .collect::>() + .into_iter() + .collect::>(); + // Delta references may be owned by history rather than carried by the delta. + // Query every candidate identity too, so selected reads retain collision checks. + view.pointer_catalog_lookup(layout, maximum)? + .dependencies(layout.store(), &files, &shards, &xorbs) + .await + .map_err(crab_read::ReadError::from) + .map_err(Into::into) +} + pub(super) fn source_pointer_delta( source: &PointerCatalog, candidate: &PointerCatalog, diff --git a/crates/crab-auth-server/src/receive/git_workspace.rs b/crates/crab-auth-server/src/receive/git_workspace.rs index 6a6036163..245180864 100644 --- a/crates/crab-auth-server/src/receive/git_workspace.rs +++ b/crates/crab-auth-server/src/receive/git_workspace.rs @@ -1,6 +1,6 @@ //! Git workspace orchestration for protected-push receive. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::fs; use std::io::Write; use std::path::Path; @@ -94,17 +94,30 @@ pub(super) async fn verify_capsule_git_candidate( let temp = tempfile::tempdir()?; let git_dir = temp.path().join("source.git"); run_git(["init", "--bare", path_str(&git_dir)?], None)?; - crab_read::capsule_protocol::install_git_packs_with_candidates( + crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( view, std::slice::from_ref(candidate), + router, &git_dir, max_input_bytes, + None, + &tokio_util::sync::CancellationToken::new(), ) .await?; let workspace = GitReceiveWorkspace::new(store, router, router.repo_prefix()); let paths = workspace.compute_changed_paths_in(&git_dir, ref_updates, &BTreeMap::new(), false)?; + let refs = verify_capsule_candidate_refs(&git_dir, view, candidate).await?; + let pointers = scan_reachable_pointers_from_refs(&git_dir, &refs)?; + Ok((paths, pointers)) +} + +async fn verify_capsule_candidate_refs( + git_dir: &Path, + view: &crab_read::capsule_protocol::CapsuleRepositoryView, + candidate: &Capsule, +) -> Result> { let transaction = candidate.transaction()?; let mut refs = view.refs().clone(); let mut peeled = view.peeled_refs().clone(); @@ -127,25 +140,57 @@ pub(super) async fn verify_capsule_git_candidate( } } let refs = refs.into_iter().collect::>(); - let closures = crab_git::walk_reachable_by_ref_bounded( - &git_dir, + let reachable = crab_git::walk_reachable_by_ref_bounded( + git_dir, &refs, &peeled, usize::try_from(crab_metadata::git_visibility::MAX_GIT_VISIBILITY_OBJECTS) .map_err(|_| invalid("Git visibility object limit does not fit usize"))?, ) - .map_err(|source| AuthServerError::GitVisibilityWalk { source })? - .into_iter() - .map(|(name, reachable)| (name, reachable_object_ids(&reachable))) - .collect::>(); + .map_err(|source| AuthServerError::GitVisibilityWalk { source })?; + // Pack presence is not read authority. Reintroducing an invisible object + // requires its body in the verified candidate, even if a retained base pack + // happens to contain it. + { + let visible = view.git_visibility_index().await?; + let mut allowed = visible + .objects_for_refs(view.refs().keys().map(String::as_str)) + .into_iter() + .collect::>(); + for pack in candidate.git_packs() { + let index = candidate.section_bytes(pack.index_section())?; + let (oids, _) = crab_git::pack_locator::sorted_object_ids_from_index_bytes(&index) + .map_err(std::io::Error::other)?; + for oid in oids { + let oid = <[u8; 20]>::try_from(oid.as_bytes()).map_err(std::io::Error::other)?; + allowed.insert(oid); + } + } + if reachable.values().any(|objects| { + objects + .commits + .iter() + .chain(&objects.trees) + .chain(&objects.blobs) + .chain(&objects.tags) + .any(|oid| !allowed.contains(oid)) + }) { + return Err(invalid( + "candidate Git object is neither visible in the captured view nor supplied by its pack", + )); + } + } + let closures = reachable + .into_iter() + .map(|(name, objects)| (name, reachable_object_ids(&objects))) + .collect::>(); let declared = view.candidate_git_visibility(candidate).await?; if closures != declared { return Err(invalid( "candidate capsule Git visibility differs from its reachable object closure", )); } - let pointers = scan_reachable_pointers_from_refs(&git_dir, &refs)?; - Ok((paths, pointers)) + Ok(refs) } pub(super) struct MaterializedCapsuleSourcePush { @@ -160,10 +205,10 @@ pub(super) async fn materialize_capsule_source_push( source_router: &StoreLayout, source_view: &crab_read::capsule_protocol::CapsuleRepositoryView, filtered_view: &crab_read::capsule_protocol::CapsuleRepositoryView, + filtered_router: &StoreLayout, candidate: &Capsule, view_updates: &[PushRefUpdate], source_updates: &[PushRefUpdate], - candidate_transaction_id: &str, max_input_bytes: u64, ) -> Result { if view_updates.len() != source_updates.len() { @@ -172,18 +217,14 @@ pub(super) async fn materialize_capsule_source_push( let temp = tempfile::tempdir()?; let git_dir = temp.path().join("source.git"); run_git(["init", "--bare", path_str(&git_dir)?], None)?; - crab_read::capsule_protocol::install_git_packs_with_candidates( - source_view, - &[], - &git_dir, - max_input_bytes, - ) - .await?; - crab_read::capsule_protocol::install_git_packs_with_candidates( + crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( filtered_view, std::slice::from_ref(candidate), + filtered_router, &git_dir, max_input_bytes, + None, + &tokio_util::sync::CancellationToken::new(), ) .await?; @@ -194,6 +235,19 @@ pub(super) async fn materialize_capsule_source_push( ); let changed_paths = workspace.compute_changed_paths_in(&git_dir, view_updates, filtered_view.refs(), true)?; + // Prove the candidate using only filtered and incoming objects before the + // private source ODB becomes available for source-commit synthesis. + let _ = verify_capsule_candidate_refs(&git_dir, filtered_view, candidate).await?; + crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( + source_view, + &[], + source_router, + &git_dir, + max_input_bytes, + None, + &tokio_util::sync::CancellationToken::new(), + ) + .await?; let mut final_updates = Vec::with_capacity(source_updates.len()); let mut synthesized_tips = Vec::with_capacity(source_updates.len()); for (view_update, source_update) in view_updates.iter().zip(source_updates) { @@ -251,7 +305,7 @@ pub(super) async fn materialize_capsule_source_push( )?; let transaction = CapsuleTransaction::for_protected_source( source_view.root().digest(), - candidate_transaction_id, + candidate.transaction_id(), final_updates .iter() .map(|update| { diff --git a/crates/crab-auth-server/src/receive/session.rs b/crates/crab-auth-server/src/receive/session.rs index 543bb1b68..a5b73675b 100644 --- a/crates/crab-auth-server/src/receive/session.rs +++ b/crates/crab-auth-server/src/receive/session.rs @@ -224,20 +224,13 @@ impl ReceiveContext { }, other => other.into(), })?; - let view = crab_read::capsule_protocol::open_view_from_root_with_control( - &self.router, - root, - crab_read::capsule_protocol::CapsuleReadLimits { - max_capsule_bytes: 2 * 1024 * 1024 * 1024, - max_frontier_bytes: 2 * 1024 * 1024 * 1024, - }, - ) - .await?; + let view = crab_read::capsule_protocol::open_ref_view_from_root(&self.router, root) + .await?; build_capsule_prepare_record( &self.repo_prefix, &self.push_id, - view.root().root().generation(), - view.root().digest(), + view.root_snapshot().record().root().generation(), + view.root_snapshot().record().digest(), view.refs(), view_ref_updates, view_scope, diff --git a/crates/crab-auth-server/src/receive/workflow.rs b/crates/crab-auth-server/src/receive/workflow.rs index 818e2ac31..e25927e46 100644 --- a/crates/crab-auth-server/src/receive/workflow.rs +++ b/crates/crab-auth-server/src/receive/workflow.rs @@ -1272,8 +1272,11 @@ mod tests { Some(source.path()), )?; run_git(["config", "user.name", "Alice"], Some(source.path()))?; + let xet_content = b"unrestricted protected Xet content\n"; + let xet = xet_fixture(xet_content)?; std::fs::write(source.path().join("tracked.txt"), b"protected capsule\n")?; - run_git(["add", "tracked.txt"], Some(source.path()))?; + std::fs::write(source.path().join("model.bin"), xet.pointer.serialize())?; + run_git(["add", "tracked.txt", "model.bin"], Some(source.path()))?; run_git(["commit", "-m", "capsule candidate"], Some(source.path()))?; let tip = run_git_capture(["-C", path_str(source.path())?, "rev-parse", "HEAD"])? .trim() @@ -1322,15 +1325,28 @@ mod tests { let capsule = Capsule::build( &transaction, vec![pack], - vec![CapsuleSection::new( - CapsuleSectionKind::VisibilityDelta, - visibility.encode()?, - )], + vec![ + CapsuleSection::new(CapsuleSectionKind::VisibilityDelta, visibility.encode()?), + CapsuleSection::new( + CapsuleSectionKind::CatalogDelta, + xet.catalog.encode_delta()?, + ), + ], )?; let run = CapsuleRun::leaf(capsule)?; let run_key = ctx.router().capsule_path(run.hash()).to_string(); let run_object = staged_object(run_key, run.bytes()); put_staged(&ctx, &run_object, run.bytes().to_vec()).await?; + let xorb_object = staged_object( + ctx.router().xorb_path(&xet.xorb_hash).to_string(), + &xet.xorb_bytes, + ); + put_staged(&ctx, &xorb_object, xet.xorb_bytes.to_vec()).await?; + let shard_object = staged_object( + ctx.router().shard_path(&xet.shard_hash).to_string(), + &xet.shard_bytes, + ); + put_staged(&ctx, &shard_object, xet.shard_bytes.clone()).await?; let plan = super::super::ProtectedCapsulePushPlan { schema_version: crab_remote::protected::PROTECTED_CAPSULE_PUSH_PLAN_SCHEMA_VERSION, repo_prefix: ctx.repo_prefix().to_owned(), @@ -1341,15 +1357,24 @@ mod tests { run_hash: run.hash().to_owned(), run_size: run.bytes().len() as u64, ref_updates: vec![update], - staged_objects: vec![run_object], + staged_objects: vec![run_object, xorb_object, shard_object], }; write_plan(&ctx, &plan).await?; let verified = verify_receive(&ctx).await?; assert_eq!(verified.ref_updates, plan.ref_updates); - assert_eq!(verified.verified_changed_paths, vec!["tracked.txt"]); - assert_eq!(verified.verified_staged_bytes, plan.run_size); + assert_eq!( + verified.verified_changed_paths, + vec!["model.bin", "tracked.txt"] + ); + assert_eq!( + verified.verified_staged_bytes, + plan.staged_objects + .iter() + .map(|object| object.size) + .sum::() + ); let response = commit_receive(&ctx, "crab://bucket/org/repo", &verified.plan_digest, None).await?; assert_eq!(response.ref_updates, plan.ref_updates); @@ -1366,6 +1391,21 @@ mod tests { view.visible_ref_transactions().get("refs/heads/main"), Some(&plan.transaction_id) ); + let catalog = view + .pointer_catalog_lookup(ctx.router(), 1024 * 1024)? + .files( + ctx.store(), + &[MerkleHash::from(xet.pointer.file_hash).hex()], + ) + .await?; + assert_eq!(catalog, xet.catalog); + crab_read::verify_capsule_pointer_catalog_objects(ctx.router(), &catalog).await?; + let (body, _) = ctx + .store() + .get_with_etag(&ctx.router().xorb_path(&xet.xorb_hash)) + .await?; + let parsed = crab_xet::xorb::parser::XorbParser::parse(body)?; + assert_eq!(parsed.get_chunk(0)?.data.as_ref(), xet_content); let receipt = crab_metadata::capsule_protocol::resolve_capsule_plan_receipt( ctx.store(), ctx.router(), @@ -1383,7 +1423,165 @@ mod tests { #[tokio::test] async fn protected_capsule_view_push_preserves_hidden_paths_and_dependencies() -> Result<()> { + for count in [100, 10_000] { + protected_capsule_view_push(false, count, None).await?; + } + Ok(()) + } + + #[tokio::test] + async fn checkpointed_capsule_view_push_preserves_hidden_paths_and_dependencies() -> Result<()> + { + for count in [100, 10_000] { + protected_capsule_view_push(true, count, None).await?; + } + Ok(()) + } + + #[tokio::test] + async fn scoped_capsule_candidate_cannot_borrow_hidden_source_git_objects() -> Result<()> { + protected_capsule_view_push(false, 100, Some("source")).await + } + + #[tokio::test] + async fn scoped_capsule_candidate_cannot_borrow_invisible_retained_git_objects() -> Result<()> { + protected_capsule_view_push(true, 100, Some("filtered")).await + } + + #[derive(Default)] + struct CatalogReads(AtomicU64, AtomicU64); + + impl crab_storage::StorageObserver for CatalogReads { + fn started(&self, _: crab_storage::StorageOperation) {} + + fn finished(&self, observation: crab_storage::StorageObservation) { + if matches!( + observation.operation, + crab_storage::StorageOperation::Get | crab_storage::StorageOperation::Range + ) { + self.0.fetch_add(1, Ordering::SeqCst); + self.1.fetch_add(observation.bytes_read, Ordering::SeqCst); + } + } + } + + async fn historical_catalog( + layout: &crab_storage::StoreLayout, + count: usize, + ) -> Result { + let mut builder = XorbBuilder::new(); + let mut files = Vec::new(); + for index in 0..count { + let data = Bytes::from(format!("unrelated historical payload {index:016x}\n")); + let file = MerkleHash::from(*blake3::hash(&data).as_bytes()); + let chunk = crab_xet::xorb::format::Chunk::new(data.clone()); + builder.push(&chunk, RunId(index as u64))?; + files.push((file, chunk.hash, data.len() as u64)); + } + let xorbs = builder.finalize()?; + let placements = xorbs + .iter() + .flat_map(|xorb| xorb.placements.iter()) + .map(|placement| (placement.chunk_hash, placement.clone())) + .collect::(); + let dependencies = xorbs + .iter() + .map(|xorb| { + Ok(( + xorb.hash, + Arc::new(xorb_info_from_placements(xorb.hash, &xorb.placements)?), + )) + }) + .collect::>>()?; + let mut shards = PushShardSession::new(); + let mut file_shards = Vec::new(); + let mut shard_xorbs = BTreeMap::>::new(); + for (hash, chunk, size) in files { + let dependency = &dependencies[&placements[&chunk].xorb_hash]; + let index = shards.add_file_bundle( + file_info_from_placements(hash, &[chunk], &placements)?, + std::slice::from_ref(dependency), + )?; + file_shards.push((hash, size, index)); + shard_xorbs + .entry(index) + .or_default() + .insert(placements[&chunk].xorb_hash.hex()); + } + let shards = shards.finalize()?; + let mut catalog = PointerCatalog::new(); + for xorb in xorbs { + let mut chunks = xorb.placements.iter().collect::>(); + chunks.sort_by_key(|chunk| chunk.chunk_index); + catalog.insert_xorb( + xorb.hash.hex(), + XorbCatalogEntry::new( + xorb.bytes.len() as u64, + blake3_hex(&xorb.bytes), + chunks + .into_iter() + .map(|chunk| { + XorbChunkEntry::new(chunk.chunk_hash.hex(), chunk.uncompressed_size) + }) + .collect(), + ), + )?; + layout + .store() + .put_exact(&layout.xorb_path(&xorb.hash), xorb.bytes) + .await?; + } + for (index, (bytes, hash)) in shards.iter().enumerate() { + catalog.insert_shard( + hash.hex(), + ShardCatalogEntry::new( + bytes.len() as u64, + shard_xorbs[&index].iter().cloned().collect(), + ), + )?; + layout + .store() + .put_exact(&layout.shard_path(hash), Bytes::from(bytes.clone())) + .await?; + } + for (hash, size, index) in file_shards { + catalog.insert_file( + hash.hex(), + FileCatalogEntry::new(size, shards[index].1.hex()), + )?; + } + Ok(catalog) + } + + async fn protected_capsule_view_push( + checkpointed: bool, + count: usize, + borrowed_from: Option<&str>, + ) -> Result<()> { let ctx = capsule_context().await?; + let counts = Arc::new(CatalogReads::default()); + let ctx = ReceiveContext::from_store( + ctx.store().clone().with_storage_observer(counts.clone()), + ctx.repo_prefix().to_owned(), + ctx.push_id().to_owned(), + ); + let history = historical_catalog(ctx.router(), count).await?; + let hidden_content = b"source-only Xet content\n"; + let hidden = xet_fixture(hidden_content)?; + ctx.store() + .put_exact( + &ctx.router().xorb_path(&hidden.xorb_hash), + hidden.xorb_bytes.clone(), + ) + .await?; + ctx.store() + .put_exact( + &ctx.router().shard_path(&hidden.shard_hash), + Bytes::from(hidden.shard_bytes.clone()), + ) + .await?; + let mut source_catalog = history.clone(); + source_catalog.apply(&hidden.catalog)?; let temp = tempfile::tempdir()?; let source = temp.path().join("source"); run_git(["init", "--initial-branch=main", path_str(&source)?], None)?; @@ -1393,11 +1591,20 @@ mod tests { std::fs::create_dir_all(source.join("secret"))?; std::fs::write(source.join("src/app.txt"), b"allowed v1\n")?; std::fs::write(source.join("secret/key.txt"), b"classified\n")?; + std::fs::write(source.join("secret/model.bin"), hidden.pointer.serialize())?; run_git(["add", "."], Some(&source))?; run_git(["commit", "-m", "source base"], Some(&source))?; let source_old = run_git_capture(["-C", path_str(&source)?, "rev-parse", "HEAD"])? .trim() .to_owned(); + let hidden_oid = run_git_capture([ + "-C", + path_str(&source)?, + "rev-parse", + &format!("{source_old}:secret/key.txt"), + ])? + .trim() + .to_owned(); let source_pack_bytes = git_capture_with_input( ["pack-objects", "--stdout", "--revs"], &source, @@ -1428,10 +1635,16 @@ mod tests { let source_capsule = Capsule::build( &source_transaction, vec![source_pack], - vec![CapsuleSection::new( - CapsuleSectionKind::VisibilityDelta, - source_visibility.encode()?, - )], + vec![ + CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + source_visibility.encode()?, + ), + CapsuleSection::new( + CapsuleSectionKind::CatalogDelta, + source_catalog.encode_delta()?, + ), + ], )?; crab_write::capsule_protocol::publish( ctx.router(), @@ -1470,15 +1683,29 @@ mod tests { "refs/heads/main", ) .await?; + assert_eq!(historical_catalog(&view_router, count).await?, history); + let view_base_revisions = if borrowed_from == Some("filtered") { + let oid = git_capture_with_input( + ["hash-object", "-w", "--stdin"], + &filtered, + b"classified\n", + )?; + assert_eq!(String::from_utf8(oid).unwrap().trim(), hidden_oid); + // Physical retention is permitted; the base visibility proof below + // deliberately grants no ref access to this extra blob. + format!("{view_old}\n{hidden_oid}\n") + } else { + format!("{view_old}\n") + }; let view_base_pack_bytes = git_capture_with_input( ["pack-objects", "--stdout", "--revs"], &filtered, - format!("{view_old}\n").as_bytes(), + view_base_revisions.as_bytes(), )?; let view_base_pack = capsule_git_pack( &filtered.join(".git"), &view_base_pack_bytes, - git_object_count(&filtered, &format!("{view_old}\n"))?, + git_object_count(&filtered, &view_base_revisions)?, )?; let view_base_transaction = CapsuleTransaction::new( view_root.record().digest(), @@ -1496,22 +1723,6 @@ mod tests { &view_old, &format!("{view_old}\n"), )?; - let view_base_capsule = Capsule::build( - &view_base_transaction, - vec![view_base_pack], - vec![CapsuleSection::new( - CapsuleSectionKind::VisibilityDelta, - view_base_visibility.encode()?, - )], - )?; - crab_write::capsule_protocol::publish( - &view_router, - view_root, - &view_base_transaction, - &view_base_capsule, - ) - .await?; - let xet_content = b"protected view Xet content\n"; let xet = xet_fixture(xet_content)?; ctx.store() @@ -1526,6 +1737,49 @@ mod tests { Bytes::from(xet.shard_bytes.clone()), ) .await?; + let mut filtered_catalog = history.clone(); + for (hash, entry) in xet.catalog.xorbs() { + filtered_catalog.insert_xorb(hash.clone(), entry.clone())?; + } + for (hash, entry) in xet.catalog.shards() { + filtered_catalog.insert_shard(hash.clone(), entry.clone())?; + } + let view_base_capsule = Capsule::build( + &view_base_transaction, + vec![view_base_pack], + vec![ + CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + view_base_visibility.encode()?, + ), + CapsuleSection::new( + CapsuleSectionKind::CatalogDelta, + filtered_catalog.encode_delta()?, + ), + ], + )?; + crab_write::capsule_protocol::publish( + &view_router, + view_root, + &view_base_transaction, + &view_base_capsule, + ) + .await?; + + if checkpointed { + for layout in [ctx.router(), &view_router] { + let outcome = crab_remote::checkpoint::publish_capsule_checkpoint( + layout, + 0, + 2 * 1024 * 1024 * 1024, + &CancellationToken::new(), + ) + .await + .map_err(std::io::Error::other)?; + assert!(outcome.published); + } + } + let lfs_content = b"protected view LFS content\n"; let lfs_oid = <[u8; 32]>::from(Sha256::digest(lfs_content)); let lfs_pointer = crab_git::lfs_pointer::LfsPointer { @@ -1540,6 +1794,10 @@ mod tests { ["add", "src/app.txt", "src/model.bin", "src/asset.lfs"], Some(&filtered), )?; + if borrowed_from.is_some() { + std::fs::write(filtered.join("src/leak.txt"), b"classified\n")?; + run_git(["add", "src/leak.txt"], Some(&filtered))?; + } run_git(["commit", "-m", "allowed update"], Some(&filtered))?; let view_new = run_git_capture(["-C", path_str(&filtered)?, "rev-parse", "HEAD"])? .trim() @@ -1568,15 +1826,23 @@ mod tests { }, ) .await?; + let candidate_revisions = if borrowed_from.is_some() { + // The candidate declares this source-only blob reachable but omits + // its body from the incoming pack. Source materialization must not + // satisfy that missing dependency from its broader private ODB. + format!("{view_new}\n^{view_old}\n^{hidden_oid}\n") + } else { + format!("{view_new}\n^{view_old}\n") + }; let candidate_pack_bytes = git_capture_with_input( ["pack-objects", "--stdout", "--revs"], &filtered, - format!("{view_new}\n^{view_old}\n").as_bytes(), + candidate_revisions.as_bytes(), )?; let candidate_pack = capsule_git_pack( &filtered.join(".git"), &candidate_pack_bytes, - git_object_count(&filtered, &format!("{view_new}\n^{view_old}\n"))?, + git_object_count(&filtered, &candidate_revisions)?, )?; let candidate_transaction = CapsuleTransaction::new( view.root().digest(), @@ -1594,6 +1860,10 @@ mod tests { &view_new, &format!("{view_new}\n"), )?; + let mut candidate_catalog = PointerCatalog::new(); + for (hash, entry) in xet.catalog.files() { + candidate_catalog.insert_file(hash.clone(), entry.clone())?; + } let candidate = Capsule::build( &candidate_transaction, vec![candidate_pack], @@ -1604,7 +1874,7 @@ mod tests { ), CapsuleSection::new( CapsuleSectionKind::CatalogDelta, - xet.catalog.encode_delta()?, + candidate_catalog.encode_delta()?, ), ], )?; @@ -1633,7 +1903,25 @@ mod tests { }; write_plan(&ctx, &plan).await?; - let verified = verify_receive(&ctx).await?; + counts.0.store(0, Ordering::SeqCst); + counts.1.store(0, Ordering::SeqCst); + let result = verify_receive(&ctx).await; + if borrowed_from.is_some() { + assert!( + result.is_err(), + "scoped candidate borrowed a hidden {borrowed_from:?} blob without supplying its body" + ); + let refs = crab_read::capsule_protocol::read_visible_refs(ctx.router()).await?; + assert_eq!(refs["refs/heads/main"], source_old); + return Ok(()); + } + let verified = result?; + assert!( + counts.0.load(Ordering::SeqCst) <= 64 && counts.1.load(Ordering::SeqCst) <= 512 * 1024, + "fixed protected verification exceeded read bounds: count={count}, checkpoint={checkpointed}, reads={}, bytes={}", + counts.0.load(Ordering::SeqCst), + counts.1.load(Ordering::SeqCst), + ); assert_eq!( verified.verified_changed_paths, vec!["src/app.txt", "src/asset.lfs", "src/model.bin"] @@ -1656,11 +1944,13 @@ mod tests { assert_ne!(final_oid, &view_new); let final_git = temp.path().join("final.git"); run_git(["init", "--bare", path_str(&final_git)?], None)?; - crab_read::capsule_protocol::install_git_packs_with_candidates( + crab_read::capsule_protocol::install_git_packs_from_store( &source_view, - &[], + ctx.router(), &final_git, u64::MAX, + None, + &CancellationToken::new(), ) .await?; assert_eq!( @@ -1681,6 +1971,13 @@ mod tests { ])?, "classified\n" ); + let hidden_pointer = run_git_capture([ + "--git-dir", + path_str(&final_git)?, + "show", + &format!("{final_oid}:secret/model.bin"), + ])?; + assert_eq!(Pointer::parse(hidden_pointer.as_bytes())?, hidden.pointer); let final_pointer = run_git_capture([ "--git-dir", path_str(&final_git)?, @@ -1722,17 +2019,26 @@ mod tests { ctx.store(), ctx.repo_prefix(), ); - let hydrated = temp.path().join("hydrated-model.bin"); - hydrator - .reconstruct_to_writer_with_cancel( - &xet.pointer, - std::fs::File::create(&hydrated)?, - Some(&lookup), - &CancellationToken::new(), - ) - .await?; + for (name, pointer, expected) in [ + ("hydrated-model.bin", &xet.pointer, xet_content.as_slice()), + ( + "hydrated-hidden.bin", + &hidden.pointer, + hidden_content.as_slice(), + ), + ] { + let hydrated = temp.path().join(name); + hydrator + .reconstruct_to_writer_with_cancel( + pointer, + std::fs::File::create(&hydrated)?, + Some(&lookup), + &CancellationToken::new(), + ) + .await?; + assert_eq!(std::fs::read(hydrated)?, expected); + } lookup.close().await?; - assert_eq!(std::fs::read(hydrated)?, xet_content); let (stored_lfs, _) = ctx .store() .get_with_etag_bounded(&ObjectPath::from(lfs_key), lfs_content.len() as u64) diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index affd4f070..7c7e003d2 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -66,7 +66,8 @@ capsule checkpoint formats; unshipped format 6 is rejected without a fallback. Captured run-v4 roots provide the same selected lookup over exact logical capsule occurrences; catalog bodies remain cold until their keys are selected. CLI and shared capsule-publication preparation use these captured queries. -Protected receive and hydration still require separate consumer integration. +Protected receive selects from separately authorized source/filtered captures; +hydration still requires consumer integration. External pointer catalogs use `CRBPCP01` authenticated pages. A bounded descriptor commits file, shard, xorb and reverse-chunk directories; nested shard closures and diff --git a/crates/crab-read/README.md b/crates/crab-read/README.md index 6fe09c3ad..002671bd6 100644 --- a/crates/crab-read/README.md +++ b/crates/crab-read/README.md @@ -61,6 +61,14 @@ work. `git_visibility_index`, `candidate_git_visibility` and `tip_bound_transitions` asynchronously materialize captured controls only when bulk inspection needs them. They do not recapture refs or switch origins. +`install_git_packs_with_candidates_from_store` combines a captured view's +origin-backed packs with verified candidate capsules. Candidates participate +in the same pack intake limit and external-delta ordering, so protected receive +can use a paged view without materializing its catalog. The caller binds each +candidate to the captured root; installation does not grant visibility. +The existing candidate-free installer delegates to this implementation and +retains its cold-clone proof and cache behavior. + `ShardHydrator` provides memory, file, and half-open byte-range reconstruction. Before full-file reconstruction it checks that the pointer's shard terms are covered; after reconstruction it verifies the requested file hash. A shared diff --git a/crates/crab-read/src/capsule_protocol.rs b/crates/crab-read/src/capsule_protocol.rs index edad84b9d..f52bb40d4 100644 --- a/crates/crab-read/src/capsule_protocol.rs +++ b/crates/crab-read/src/capsule_protocol.rs @@ -2150,24 +2150,59 @@ pub async fn install_git_packs_from_store( cache: Option<&crab_cache_store::CachingStore>, cancel: &CancellationToken, ) -> Result { - if cancel.is_cancelled() { - return Err(ReadError::Cancelled); - } - if let Some(installed) = install_layered_cold_clone_packs_from_store( + install_git_packs_with_candidates_from_store( view, + &[], router, git_dir, max_input_bytes, cache, cancel, ) - .await? + .await +} + +/// Install a captured origin-backed view and verified candidate capsules together. +/// +/// Candidates must be bound to the captured root by the caller. Base and candidate +/// packs share intake limits and external-delta ordering; no catalog bodies are read. +pub async fn install_git_packs_with_candidates_from_store( + view: &CapsuleRepositoryView, + candidates: &[Capsule], + router: &StoreLayout, + git_dir: &Path, + max_input_bytes: u64, + cache: Option<&crab_cache_store::CachingStore>, + cancel: &CancellationToken, +) -> Result { + if cancel.is_cancelled() { + return Err(ReadError::Cancelled); + } + // The cold-clone proof covers only the captured view. Candidate packs must + // join the same intake and delta-base ordering before anything is installed. + if candidates.is_empty() + && let Some(installed) = install_layered_cold_clone_packs_from_store( + view, + router, + git_dir, + max_input_bytes, + cache, + cancel, + ) + .await? { return Ok(installed); } if view.layered_checkpoint().is_some() || (view.capsules.is_empty() && !view.capsule_run_sources.is_empty()) + || !candidates.is_empty() { + let capsules = view + .capsules + .iter() + .chain(candidates) + .cloned() + .collect::>(); // Metadata views expose authenticated source ranges before the first // checkpoint too. Installing only resident capsules would drop their packs. let sources = view @@ -2176,7 +2211,7 @@ pub async fn install_git_packs_from_store( let paths = install_layered_git_packs_from_store_selected_with_sources( sources, &view.capsule_run_sources, - &view.capsules, + &capsules, router, git_dir, max_input_bytes, diff --git a/crates/crab-remote/tests/checkpoint/external_delta.rs b/crates/crab-remote/tests/checkpoint/external_delta.rs index c50bc3681..63960bbca 100644 --- a/crates/crab-remote/tests/checkpoint/external_delta.rs +++ b/crates/crab-remote/tests/checkpoint/external_delta.rs @@ -10,6 +10,19 @@ async fn publish_dependent_blob( base: &[u8], target: &[u8], ) { + let (root, capsule) = dependent_blob_capsule(layout, name, base, target).await; + let transaction = capsule.transaction().unwrap(); + crab_write::capsule_protocol::publish(layout, root, &transaction, &capsule) + .await + .unwrap(); +} + +async fn dependent_blob_capsule( + layout: &StoreLayout, + name: &str, + base: &[u8], + target: &[u8], +) -> (crab_metadata::capsule_protocol::RootSnapshot, Capsule) { let kind = gix_object::Kind::Blob; let base_oid = crab_remote::objects::object_id(kind, base).unwrap(); let target_oid = crab_remote::objects::object_id(kind, target).unwrap(); @@ -82,9 +95,111 @@ async fn publish_dependent_blob( )], ) .unwrap(); - crab_write::capsule_protocol::publish(layout, root, &transaction, &capsule) + (root, capsule) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn candidate_install_shares_intake_and_resolves_captured_external_bases() { + for checkpointed in [false, true] { + let (layout, observations) = empty_fixture().await; + let base = (0..4096_u32) + .flat_map(|ordinal| *blake3::hash(&ordinal.to_le_bytes()).as_bytes()) + .collect::>(); + let mut target = base.clone(); + for byte in &mut target[..64 * 1024] { + *byte ^= 0xa5; + } + publish_blobs(&layout, &[("base", &base)]).await; + if checkpointed { + checkpoint(&layout).await; + } + let root = crab_write::capsule_protocol::open_root(&layout) + .await + .unwrap(); + let captured = crab_read::capsule_protocol::open_git_view_from_root(&layout, root, LIMITS) + .await + .unwrap(); + let (root, candidate) = dependent_blob_capsule(&layout, "candidate", &base, &target).await; + assert_eq!(root.record().digest(), captured.root().digest()); + let candidate_size = candidate + .section_bytes(candidate.git_packs()[0].pack_section()) + .unwrap() + .len() as u64; + let base_sizes = captured + .layered_checkpoint() + .into_iter() + .flat_map(|checkpoint| checkpoint.sources()) + .chain(captured.capsule_run_sources()) + .flat_map(|source| source.members()) + .map(|member| (member.pack().blake3().to_owned(), member.pack().length())) + .collect::>(); + let base_size = base_sizes.values().sum::(); + let budget = 160 * 1024; + assert!( + base_size < budget && candidate_size < budget && base_size + candidate_size > budget + ); + let bounded = tempfile::tempdir().unwrap(); + crab_git::initialize_bare_git_dir(bounded.path()).unwrap(); + let result = crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( + &captured, + std::slice::from_ref(&candidate), + &layout, + bounded.path(), + budget, + None, + &CancellationToken::new(), + ) + .await; + assert!(matches!( + result, + Err(crab_read::ReadError::CapsuleReadLimit { + resource: "Git pack intake", + .. + }) + )); + let directory = tempfile::tempdir().unwrap(); + crab_git::initialize_bare_git_dir(directory.path()).unwrap(); + let cancelled = CancellationToken::new(); + cancelled.cancel(); + observations.0.lock().unwrap().clear(); + let result = crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( + &captured, + std::slice::from_ref(&candidate), + &layout, + directory.path(), + LIMIT, + None, + &cancelled, + ) + .await; + assert!(matches!(result, Err(crab_read::ReadError::Cancelled))); + assert!(observations.0.lock().unwrap().is_empty()); + let installed = crab_read::capsule_protocol::install_git_packs_with_candidates_from_store( + &captured, + std::slice::from_ref(&candidate), + &layout, + directory.path(), + LIMIT, + None, + &CancellationToken::new(), + ) .await .unwrap(); + assert!(!installed.complete_visibility); + let oid = crab_remote::objects::object_id(gix_object::Kind::Blob, &target).unwrap(); + let output = std::process::Command::new("git") + .arg("--git-dir") + .arg(directory.path()) + .args(["cat-file", "blob", &oid.to_string()]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!(output.stdout, target); + } } #[tokio::test(flavor = "multi_thread", worker_threads = 2)]