diff --git a/.github/workflows/repository-gates.yml b/.github/workflows/repository-gates.yml new file mode 100644 index 00000000..b656c665 --- /dev/null +++ b/.github/workflows/repository-gates.yml @@ -0,0 +1,79 @@ +name: Repository gates + +on: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: repository-gates-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + linux: + name: Linux repository gates + runs-on: ubuntu-latest + timeout-minutes: 120 + env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: "0" + CARGO_BUILD_JOBS: "2" + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_PROFILE_TEST_DEBUG: "0" + RUST_BACKTRACE: "1" + NO_PROXY: localhost,127.0.0.1,::1,host.docker.internal + steps: + - uses: actions/checkout@v4 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y pkg-config libssl-dev protobuf-compiler + + - uses: dtolnay/rust-toolchain@nightly + with: + components: rustfmt + + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy + + - uses: Swatinem/rust-cache@v2 + with: + save-if: "true" + + - name: Check formatting + run: cargo +nightly fmt --all --check + + - name: Check all targets and features + run: cargo clippy --all-targets --all-features -- -D warnings + + - name: Build + run: cargo build + + - name: Build tests + run: cargo build --tests + + - name: Start isolated test dependencies + shell: bash + run: | + set -euo pipefail + source scripts/lib/mega2-it.sh + mega2_it_ensure_env_test + mega2_it_prepare_git_workdir + docker compose -p mega2-it -f docker/docker-compose.test.yml --profile git \ + up -d --build --wait postgres redis rustfs rustfs-init git-cli + + - name: Run repository tests with test environment + shell: bash + run: | + set -euo pipefail + source .env.test + export no_proxy="$NO_PROXY" + cargo test --all + + - name: Stop isolated test dependencies + if: always() + run: docker compose -p mega2-it -f docker/docker-compose.test.yml --profile git down -v diff --git a/src/callisto/mst2_verified_object.rs b/src/callisto/mst2_verified_object.rs index 851f5cd5..9933e936 100644 --- a/src/callisto/mst2_verified_object.rs +++ b/src/callisto/mst2_verified_object.rs @@ -2,9 +2,9 @@ //! //! One row per verified object: Git identity + raw content digest + 64-bit //! size + verification state. The natural key is -//! `(storage_domain, git_oid, object_kind)`; rows are insert-only with -//! ON CONFLICT DO NOTHING — a row only ever appears after the content was -//! fetched and hashed by this service (write-through verification). +//! `(storage_domain, git_oid, object_kind)`. Current-generation facts retain +//! first-write-wins semantics; valid legacy facts can be atomically upgraded +//! only after this service fetches and hashes the raw body again. use sea_orm::entity::prelude::*; use serde::{Deserialize, Serialize}; diff --git a/src/ceres/pack/import_repo.rs b/src/ceres/pack/import_repo.rs index 6b6bd6ce..b98cbb39 100644 --- a/src/ceres/pack/import_repo.rs +++ b/src/ceres/pack/import_repo.rs @@ -452,6 +452,10 @@ impl RepoHandler for ImportRepo { } } +#[allow( + clippy::double_must_use, + reason = "async_recursion adds must_use to the generated boxed Future" +)] #[async_recursion] pub(crate) async fn collect_git_blob_filepaths( storage: GitDbStorage, diff --git a/src/ceres/snapshot/error.rs b/src/ceres/snapshot/error.rs index 8d9cb316..4fe54274 100644 --- a/src/ceres/snapshot/error.rs +++ b/src/ceres/snapshot/error.rs @@ -34,6 +34,8 @@ pub enum SnapshotErrorCode { DigestMismatch, RangeNotSupported, SymlinkTraversal, + ObjectUnavailable, + IntegrityError, Internal, } @@ -60,6 +62,8 @@ impl SnapshotErrorCode { SnapshotErrorCode::DigestMismatch => "EXPECTED_DIGEST_MISMATCH", SnapshotErrorCode::RangeNotSupported => "RANGE_NOT_SUPPORTED", SnapshotErrorCode::SymlinkTraversal => "SYMLINK_TRAVERSAL", + SnapshotErrorCode::ObjectUnavailable => "OBJECT_UNAVAILABLE", + SnapshotErrorCode::IntegrityError => "INTEGRITY_ERROR", SnapshotErrorCode::Internal => "INTERNAL", } } @@ -79,6 +83,7 @@ impl SnapshotErrorCode { | SnapshotErrorCode::PathNotFound | SnapshotErrorCode::LeaseUnknown => 404, SnapshotErrorCode::SnapshotNotReady => 503, + SnapshotErrorCode::ObjectUnavailable => 503, SnapshotErrorCode::NotDirectory | SnapshotErrorCode::Conflict | SnapshotErrorCode::CursorStale @@ -87,6 +92,7 @@ impl SnapshotErrorCode { SnapshotErrorCode::UnsupportedEntry => 422, SnapshotErrorCode::LimitExceeded | SnapshotErrorCode::ProofBudgetExceeded => 413, SnapshotErrorCode::LeaseExpired | SnapshotErrorCode::SnapshotGone => 410, + SnapshotErrorCode::IntegrityError => 502, SnapshotErrorCode::Internal => 500, } } diff --git a/src/ceres/snapshot/mod.rs b/src/ceres/snapshot/mod.rs index 7ac7fc25..a7c44a61 100644 --- a/src/ceres/snapshot/mod.rs +++ b/src/ceres/snapshot/mod.rs @@ -1,12 +1,13 @@ //! MST/2 snapshot domain (spec 02/03): fixed views over pinned Git trees. //! -//! Slice scope (T02-a): native monorepo views only — no bindings, imports or -//! release policy yet (T04/T05). Everything here resolves against fixed OIDs; -//! no code path in this module may read current refs. +//! Serving remains native-only. The independent namespace index is an unwired +//! composition seam; identity, attestation and publication integration remain +//! separate gates. Fixed-source readers must never look up current refs. pub mod chunks; pub mod descriptor; pub mod error; pub mod frame_stream; +pub mod namespace; pub mod pages; pub mod publication; pub mod resolver; diff --git a/src/ceres/snapshot/namespace.rs b/src/ceres/snapshot/namespace.rs new file mode 100644 index 00000000..720a6171 --- /dev/null +++ b/src/ceres/snapshot/namespace.rs @@ -0,0 +1,405 @@ +//! Immutable component-prefix routing for an already attested namespace. +//! +//! `S` is a caller-owned fixed source handle, not a registry path or ref selector. +//! This module neither encodes NamespaceView identity nor grants authorization, +//! retention or release-write permission. It is not wired into serving/publication. + +use std::collections::{BTreeMap, BTreeSet}; + +use thiserror::Error; + +use super::{ + error::{SnapshotError, SnapshotErrorCode}, + resolver::FsKind, + view::validate_scope_relative_path, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BindingPolicy { + Mutable, + ImmutableRelease, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SourceKind { + Native, + Import, +} + +#[derive(Debug, Clone)] +pub struct FixedBinding { + pub mount_path: String, + pub source: S, + pub source_kind: SourceKind, + /// Canonical source-root-relative path; the source root is the empty string. + pub source_subpath: String, + pub policy: BindingPolicy, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NodeClass { + NativeTree, + NativeCheckoutRoot, + ImportRoot, + ImportTree, + Aggregate, +} + +#[derive(Debug, Clone)] +pub struct SourceNode { + pub fs_kind: FsKind, + pub oid: String, +} + +#[derive(Debug, Clone)] +pub struct SourceEntry { + pub name: String, + pub node: SourceNode, +} + +#[derive(Debug, Clone)] +pub enum SourceOutcome { + Found(SourceNode), + /// Absence established by reading a fixed parent tree, never an I/O failure. + AbsentProven, +} + +/// An adapter must read only the attested fixed root held by `source`, verify +/// membership, and preserve storage errors. Paths are relative to that root; +/// `scope_path` must not be prepended again. Attestation/auth/retention are external. +/// The adapter also enforces the attested scope plus source-path byte budget. +pub trait FixedSourceReader { + fn resolve(&mut self, source: &S, source_path: &str) -> Result; + fn list(&mut self, source: &S, source_path: &str) -> Result, SnapshotError>; +} + +#[derive(Debug)] +pub struct SourceRoute<'a, S> { + pub source: &'a S, + pub source_path: String, + pub source_kind: SourceKind, + pub mount_path: Option<&'a str>, + /// Selected binding provenance, not an inherited release protection or a + /// write permission. A persistent release registry must fence every writer. + pub policy: BindingPolicy, +} + +#[derive(Debug)] +pub struct NamespaceEntry<'a, S> { + pub name: String, + pub fs_kind: FsKind, + /// Synthetic ancestors have neither an object ID nor a source context. + pub oid: Option, + pub source_context: Option>, +} + +#[derive(Debug)] +pub struct NamespaceDirectory<'a, S> { + pub node_class: NodeClass, + pub source_context: Option>, + pub entries: Vec>, +} + +#[derive(Debug, Error)] +pub enum NamespaceError { + #[error(transparent)] + Snapshot(#[from] SnapshotError), + #[error("NAMESPACE_CONFLICT at {0}")] + Conflict(String), + #[error("duplicate mount path {0}")] + DuplicateMount(String), + #[error("binding source subpath is not a directory at {0}")] + SourceSubpathNotDirectory(String), + #[error("invalid source directory entry {0}")] + InvalidSourceEntry(String), +} + +#[derive(Debug, Default)] +struct PrefixNode { + binding: Option, + children: BTreeMap, +} + +/// Construct a new index for each fixed view; no binding/source mutators exist. +/// The trie restricts each request to its path and immediate indexed children. +#[derive(Debug)] +pub struct FixedNamespaceIndex { + native: S, + bindings: Vec>, + root: PrefixNode, + native_checkouts: BTreeSet, +} + +impl FixedNamespaceIndex { + pub fn new>( + native: S, + bindings: Vec>, + native_checkouts: BTreeSet, + reader: &mut R, + ) -> Result { + let mut root = PrefixNode::default(); + for (id, binding) in bindings.iter().enumerate() { + validate_scope_relative_path(&binding.mount_path)?; + validate_relative_path(&binding.source_subpath)?; + let mut node = &mut root; + for component in components(&binding.mount_path) { + node = node.children.entry(component.to_string()).or_default(); + } + if node.binding.replace(id).is_some() { + return Err(NamespaceError::DuplicateMount(binding.mount_path.clone())); + } + } + for checkout in &native_checkouts { + validate_scope_relative_path(checkout)?; + } + let index = Self { + native, + bindings, + root, + native_checkouts, + }; + index.validate_sources(reader)?; + Ok(index) + } + + pub fn route(&self, path: &str) -> Result, NamespaceError> { + self.route_inner(path, false) + } + + fn route_inner( + &self, + path: &str, + exclude_exact: bool, + ) -> Result, NamespaceError> { + validate_scope_relative_path(path)?; + let comps = components(path); + let mut prefix = &self.root; + let mut selected = prefix + .binding + .filter(|_| !(exclude_exact && comps.is_empty())); + let mut selected_depth = 0; + for (depth, component) in comps.iter().enumerate() { + let Some(child) = prefix.children.get(*component) else { + break; + }; + prefix = child; + if let Some(id) = prefix.binding + && !(exclude_exact && depth + 1 == comps.len()) + { + selected = Some(id); + selected_depth = depth + 1; + } + } + let route = if let Some(id) = selected { + let binding = &self.bindings[id]; + let suffix = comps[selected_depth..].join("/"); + let source_path = join_relative(&binding.source_subpath, &suffix); + SourceRoute { + source: &binding.source, + source_path, + source_kind: binding.source_kind, + mount_path: Some(&binding.mount_path), + policy: binding.policy, + } + } else { + SourceRoute { + source: &self.native, + source_path: comps.join("/"), + source_kind: SourceKind::Native, + mount_path: None, + policy: BindingPolicy::Mutable, + } + }; + validate_relative_path(&route.source_path)?; + Ok(route) + } + + fn prefix(&self, path: &str) -> Option<&PrefixNode> { + let mut node = &self.root; + for component in components(path) { + node = node.children.get(component)?; + } + Some(node) + } + + /// Validate all mount crossings once when constructing the index. The + /// binding at the endpoint is excluded so it cannot hide a source file. + fn validate_sources>( + &self, + reader: &mut R, + ) -> Result<(), NamespaceError> { + match reader.resolve(&self.native, "")? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => {} + _ => return Err(NamespaceError::Conflict("/".to_string())), + } + for binding in &self.bindings { + match reader.resolve(&binding.source, &binding.source_subpath)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => {} + _ => { + return Err(NamespaceError::SourceSubpathNotDirectory( + binding.mount_path.clone(), + )); + } + } + let comps = components(&binding.mount_path); + for length in 0..=comps.len() { + let path = format!("/{}", comps[..length].join("/")); + let route = self.route_inner(&path, length == comps.len())?; + if let SourceOutcome::Found(node) = + reader.resolve(route.source, &route.source_path)? + && node.fs_kind != FsKind::Directory + { + return Err(NamespaceError::Conflict(path)); + } + } + } + Ok(()) + } + + /// Return proven absence as None; failures remain errors. Exact mounts + /// replace the underlying source subtree, with only explicit descendants + /// added. No registry enumeration or implicit native/import union occurs. + pub fn list>( + &self, + path: &str, + reader: &mut R, + ) -> Result>, NamespaceError> { + let route = self.route(path)?; + let indexed = self.prefix(path); + let has_children = indexed.is_some_and(|node| !node.children.is_empty()); + let mut entries = BTreeMap::new(); + let mut aggregate = false; + let source_context = match reader.resolve(route.source, &route.source_path)? { + SourceOutcome::AbsentProven if !has_children => return Ok(None), + SourceOutcome::AbsentProven => { + aggregate = true; + None + } + SourceOutcome::Found(node) if node.fs_kind != FsKind::Directory => { + return Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "namespace path is not a directory", + ) + .into()); + } + SourceOutcome::Found(_) => { + for entry in reader.list(route.source, &route.source_path)? { + validate_entry_name(&entry.name)?; + let child_path = append_absolute(path, &entry.name); + let context = self.route(&child_path)?; + let name = entry.name; + if entries + .insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: entry.node.fs_kind, + oid: Some(entry.node.oid), + source_context: Some(context), + }, + ) + .is_some() + { + return Err(NamespaceError::InvalidSourceEntry(name)); + } + } + Some(route) + } + }; + if let Some(prefix) = indexed { + for (name, child) in &prefix.children { + let child_path = append_absolute(path, name); + if child.binding.is_some() { + let route = self.route(&child_path)?; + let node = match reader.resolve(route.source, &route.source_path)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => node, + _ => return Err(NamespaceError::SourceSubpathNotDirectory(child_path)), + }; + aggregate = true; + entries.insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: node.fs_kind, + oid: Some(node.oid), + source_context: Some(route), + }, + ); + } else if let Some(entry) = entries.get(name) { + if entry.fs_kind != FsKind::Directory { + return Err(NamespaceError::Conflict(child_path)); + } + } else { + aggregate = true; + entries.insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: FsKind::Directory, + oid: None, + source_context: None, + }, + ); + } + } + } + let node_class = if aggregate { + NodeClass::Aggregate + } else if let Some(route) = &source_context { + match route.source_kind { + SourceKind::Import if route.mount_path == Some(path) => NodeClass::ImportRoot, + SourceKind::Import => NodeClass::ImportTree, + SourceKind::Native if self.native_checkouts.contains(path) => { + NodeClass::NativeCheckoutRoot + } + SourceKind::Native => NodeClass::NativeTree, + } + } else { + NodeClass::Aggregate + }; + Ok(Some(NamespaceDirectory { + node_class, + source_context, + entries: entries.into_values().collect(), + })) + } +} + +fn components(path: &str) -> Vec<&str> { + if path == "/" { + Vec::new() + } else { + path[1..].split('/').collect() + } +} + +fn join_relative(parent: &str, suffix: &str) -> String { + match (parent.is_empty(), suffix.is_empty()) { + (true, _) => suffix.to_string(), + (_, true) => parent.to_string(), + _ => format!("{parent}/{suffix}"), + } +} + +fn append_absolute(parent: &str, name: &str) -> String { + if parent == "/" { + format!("/{name}") + } else { + format!("{parent}/{name}") + } +} + +fn validate_relative_path(path: &str) -> Result<(), SnapshotError> { + validate_scope_relative_path(&format!("/{path}")) +} + +fn validate_entry_name(name: &str) -> Result<(), NamespaceError> { + if name.contains('/') || validate_relative_path(name).is_err() || name.is_empty() { + return Err(NamespaceError::InvalidSourceEntry(name.to_string())); + } + Ok(()) +} + +#[cfg(test)] +#[path = "namespace/tests.rs"] +mod tests; diff --git a/src/ceres/snapshot/namespace/tests.rs b/src/ceres/snapshot/namespace/tests.rs new file mode 100644 index 00000000..6b2fa27e --- /dev/null +++ b/src/ceres/snapshot/namespace/tests.rs @@ -0,0 +1,838 @@ +use std::{ + collections::{BTreeMap, BTreeSet}, + io::Write, + path::Path, + process::{Command, Stdio}, +}; + +use tempfile::TempDir; + +use crate::ceres::snapshot::{ + error::{SnapshotError, SnapshotErrorCode}, + namespace::{ + BindingPolicy, FixedBinding, FixedNamespaceIndex, FixedSourceReader, NamespaceError, + NodeClass, SourceEntry, SourceKind, SourceNode, SourceOutcome, + }, + resolver::FsKind, +}; + +#[derive(Default)] +struct MemoryReader { + nodes: BTreeMap<(String, String), SourceNode>, + fail_path: Option, + fail_listing: bool, + resolves: usize, + lists: usize, +} + +impl MemoryReader { + fn add(&mut self, source: &str, path: &str, fs_kind: FsKind) { + self.nodes.insert( + (source.to_string(), path.to_string()), + SourceNode { + fs_kind, + oid: format!("{source}:{path}"), + }, + ); + } + + fn roots() -> Self { + let mut reader = Self::default(); + for source in ["native", "a20", "a21", "b30"] { + reader.add(source, "", FsKind::Directory); + } + reader + } +} + +impl FixedSourceReader for MemoryReader { + fn resolve(&mut self, source: &String, path: &str) -> Result { + self.resolves += 1; + if self.fail_path.as_deref() == Some(path) { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "injected missing backend object", + )); + } + Ok(self + .nodes + .get(&(source.clone(), path.to_string())) + .cloned() + .map_or(SourceOutcome::AbsentProven, SourceOutcome::Found)) + } + + fn list(&mut self, source: &String, path: &str) -> Result, SnapshotError> { + self.lists += 1; + if self.fail_listing { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "injected listing failure", + )); + } + let prefix = if path.is_empty() { + String::new() + } else { + format!("{path}/") + }; + Ok(self + .nodes + .iter() + .filter_map(|((owner, node_path), node)| { + if owner != source { + return None; + } + let suffix = node_path.strip_prefix(&prefix)?; + if suffix.is_empty() || suffix.contains('/') { + return None; + } + Some(SourceEntry { + name: suffix.to_string(), + node: node.clone(), + }) + }) + .collect()) + } +} + +fn binding(path: &str, source: &str, subpath: &str) -> FixedBinding { + FixedBinding { + mount_path: path.to_string(), + source: source.to_string(), + source_kind: SourceKind::Import, + source_subpath: subpath.to_string(), + policy: BindingPolicy::Mutable, + } +} + +#[test] +fn component_prefix_source_subpath_and_nested_routing_match_manual_table() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "src", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/x/lib", "a20", "src"), + binding("/x/lib/vendor/b", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + for (path, expected_source, expected_path) in [ + ("/", "native", ""), + ("/x/lib", "a20", "src"), + ("/x/lib/a", "a20", "src/a"), + ("/x/library/a", "native", "x/library/a"), + ("/x/lib/vendor", "a20", "src/vendor"), + ("/x/lib/vendor/b/z", "b30", "z"), + ("/x/lib/vendor/beta/z", "a20", "src/vendor/beta/z"), + ("/x/lib/é+\\name", "a20", "src/é+\\name"), + ] { + let route = index.route(path).unwrap(); + assert_eq!(route.source, expected_source, "{path}"); + assert_eq!(route.source_path, expected_path, "{path}"); + } +} + +#[test] +fn synthetic_ancestors_have_no_source_and_exact_mount_has_no_native_union() { + let mut reader = MemoryReader::roots(); + reader.add("native", "lib", FsKind::Directory); + reader.add("native", "lib/native-only", FsKind::Regular); + reader.add("a20", "src", FsKind::Directory); + reader.add("a20", "src/import-only", FsKind::Executable); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/lib", "a20", "src"), + binding("/missing/deep/mount", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let synthetic = index.list("/missing", &mut reader).unwrap().unwrap(); + assert_eq!(synthetic.node_class, NodeClass::Aggregate); + assert!(synthetic.source_context.is_none()); + assert_eq!(synthetic.entries[0].name, "deep"); + assert!(synthetic.entries[0].oid.is_none()); + assert!(synthetic.entries[0].source_context.is_none()); + let imported = index.list("/lib", &mut reader).unwrap().unwrap(); + assert_eq!(imported.node_class, NodeClass::ImportRoot); + assert_eq!(imported.entries.len(), 1); + assert_eq!(imported.entries[0].name, "import-only"); + assert_eq!(imported.entries[0].fs_kind, FsKind::Executable); + assert!(index.list("/absent", &mut reader).unwrap().is_none()); +} + +#[test] +fn node_class_depends_on_direct_composition_and_explicit_checkout() { + let mut reader = MemoryReader::roots(); + reader.add("native", "app", FsKind::Directory); + reader.add("native", "plain", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/app/vendor/lib", "a20", "")], + BTreeSet::from(["/app".to_string(), "/plain".to_string()]), + &mut reader, + ) + .unwrap(); + assert_eq!( + index.list("/", &mut reader).unwrap().unwrap().node_class, + NodeClass::NativeTree + ); + let app = index.list("/app", &mut reader).unwrap().unwrap(); + assert_eq!(app.node_class, NodeClass::Aggregate); + assert!(app.source_context.is_some()); + assert_eq!( + index + .list("/plain", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::NativeCheckoutRoot + ); +} + +#[test] +fn file_and_symlink_crossings_and_exact_mount_targets_are_conflicts() { + for kind in [FsKind::Regular, FsKind::Executable, FsKind::Symlink] { + for mount in ["/blocked", "/blocked/child"] { + let mut reader = MemoryReader::roots(); + reader.add("native", "blocked", kind); + let error = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding(mount, "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap_err(); + assert!(matches!(error, NamespaceError::Conflict(path) if path == "/blocked")); + } + } + let mut reader = MemoryReader::roots(); + reader.add("a20", "file", FsKind::Regular); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/import", "a20", "file")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::SourceSubpathNotDirectory(_)) + )); +} + +#[test] +fn nested_binding_cannot_hide_an_outer_source_file() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "blocked", FsKind::Symlink); + let result = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/import", "a20", ""), + binding("/import/blocked/nested", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ); + assert!(matches!(result, Err(NamespaceError::Conflict(path)) if path == "/import/blocked")); +} + +#[test] +fn removing_binding_restores_native_in_new_index_and_preserves_old_index() { + let mut reader = MemoryReader::roots(); + reader.add("native", "lib", FsKind::Directory); + reader.add("native", "lib/native-only", FsKind::Regular); + reader.add("a20", "old-only", FsKind::Regular); + reader.add("a21", "new-only", FsKind::Regular); + let old = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let new = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert_eq!( + old.list("/lib", &mut reader).unwrap().unwrap().entries[0].name, + "old-only" + ); + assert_eq!( + new.list("/lib", &mut reader).unwrap().unwrap().entries[0].name, + "native-only" + ); + assert_eq!(old.route("/lib/unvisited/deep").unwrap().source, "a20"); +} + +#[test] +fn duplicate_mounts_and_noncanonical_paths_are_rejected() { + let mut reader = MemoryReader::roots(); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/same", "a20", ""), binding("/same", "b30", "")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::DuplicateMount(_)) + )); + for path in ["", "relative", "/a/", "/a//b", "/a/..", "/a\0"] { + assert!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding(path, "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .is_err(), + "{path:?}" + ); + } + for subpath in ["/src", "src/", "src//a", "src/..", "src\0"] { + assert!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", subpath)], + BTreeSet::new(), + &mut reader, + ) + .is_err(), + "{subpath:?}" + ); + } +} + +#[test] +fn composed_source_path_limits_are_checked_after_prefix_replacement() { + let mut reader = MemoryReader::roots(); + let subpath = vec!["a"; 256].join("/"); + reader.add("a20", &subpath, FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", &subpath)], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(index.route("/lib").is_ok()); + assert!(index.route("/lib/extra").is_err()); + assert!(index.route(&format!("/{}", "a".repeat(256))).is_err()); + assert!( + index + .route(&format!("/{}", vec!["a"; 257].join("/"))) + .is_err() + ); + let max_bytes = vec!["a".repeat(255); 16].join("/"); + assert_eq!(max_bytes.len() + 1, 4096); + reader.add("a20", &max_bytes, FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", &max_bytes)], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(index.route("/lib").is_ok()); + assert!(index.route("/lib/x").is_err()); +} + +#[test] +fn import_root_class_yields_to_direct_binding_and_import_tree_class_is_retained() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "src", FsKind::Directory); + reader.add("a20", "src/plain", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/lib", "a20", "src"), + binding("/lib/nested", "b30", ""), + ], + BTreeSet::from(["/lib".to_string()]), + &mut reader, + ) + .unwrap(); + let root = index.list("/lib", &mut reader).unwrap().unwrap(); + assert_eq!(root.node_class, NodeClass::Aggregate); + assert_eq!(root.source_context.as_ref().unwrap().source, "a20"); + assert_eq!( + index + .list("/lib/plain", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::ImportTree + ); + assert_eq!( + index + .list("/lib/nested", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::ImportRoot + ); +} + +#[test] +fn storage_failure_is_preserved_in_construction_and_listing() { + let mut reader = MemoryReader::roots(); + let index = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + reader.fail_path = Some("missing".to_string()); + assert!(matches!( + index.list("/missing", &mut reader), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", "missing")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); + reader.fail_path = None; + reader.fail_listing = true; + assert!(matches!( + index.list("/", &mut reader), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); +} + +#[test] +fn root_binding_replaces_entire_native_source() { + let mut reader = MemoryReader::roots(); + reader.add("native", "native-only", FsKind::Regular); + reader.add("a20", "import-only", FsKind::Regular); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/", "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert_eq!(index.route("/").unwrap().source, "a20"); + assert_eq!(index.route("/anything").unwrap().source_path, "anything"); + let root = index.list("/", &mut reader).unwrap().unwrap(); + assert_eq!(root.node_class, NodeClass::ImportRoot); + assert_eq!(root.entries.len(), 1); + assert_eq!(root.entries[0].name, "import-only"); +} + +#[test] +fn invalid_and_duplicate_source_entries_are_rejected() { + struct MalformedReader(Vec); + impl FixedSourceReader for MalformedReader { + fn resolve(&mut self, _: &String, _: &str) -> Result { + Ok(SourceOutcome::Found(SourceNode { + fs_kind: FsKind::Directory, + oid: "root".to_string(), + })) + } + + fn list(&mut self, _: &String, _: &str) -> Result, SnapshotError> { + Ok(self + .0 + .iter() + .map(|name| SourceEntry { + name: name.clone(), + node: SourceNode { + fs_kind: FsKind::Regular, + oid: "blob".to_string(), + }, + }) + .collect()) + } + } + for names in [vec![""], vec![".."], vec!["a/b"], vec!["same", "same"]] { + let mut reader = MalformedReader(names.iter().map(|s| s.to_string()).collect()); + let index = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(matches!( + index.list("/", &mut reader), + Err(NamespaceError::InvalidSourceEntry(_)) + )); + } +} + +#[test] +fn unrelated_binding_count_does_not_expand_source_reads_for_a_listing() { + let mut reader = MemoryReader::roots(); + reader.add("native", "app", FsKind::Directory); + let bindings = (0..512) + .map(|i| binding(&format!("/elsewhere-{i}/lib"), "a20", "")) + .collect(); + let index = + FixedNamespaceIndex::new("native".to_string(), bindings, BTreeSet::new(), &mut reader) + .unwrap(); + reader.resolves = 0; + reader.lists = 0; + index.list("/app", &mut reader).unwrap().unwrap(); + assert_eq!((reader.resolves, reader.lists), (1, 1)); +} + +fn git(repo: &Path, args: &[&str], input: Option<&[u8]>) -> String { + let mut child = Command::new("git") + .current_dir(repo) + .env("GIT_AUTHOR_NAME", "MST2 Oracle") + .env("GIT_AUTHOR_EMAIL", "mst2-oracle@example.invalid") + .env("GIT_COMMITTER_NAME", "MST2 Oracle") + .env("GIT_COMMITTER_EMAIL", "mst2-oracle@example.invalid") + .env("GIT_AUTHOR_DATE", "2026-09-15T00:00:00Z") + .env("GIT_COMMITTER_DATE", "2026-09-15T00:00:00Z") + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + if let Some(input) = input { + child.stdin.take().unwrap().write_all(input).unwrap(); + } else { + drop(child.stdin.take()); + } + let output = child.wait_with_output().unwrap(); + assert!( + output.status.success(), + "{args:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +} + +fn make_tree(repo: &Path, files: &[(&str, &str)]) -> String { + let mut items = Vec::new(); + let mut directories: BTreeMap<&str, Vec<(&str, &str)>> = BTreeMap::new(); + for (path, bytes) in files { + if let Some((head, tail)) = path.split_once('/') { + directories.entry(head).or_default().push((tail, bytes)); + } else { + let oid = git( + repo, + &["hash-object", "-w", "--stdin"], + Some(bytes.as_bytes()), + ); + items.push(format!("100644 blob {}\t{path}\n", oid.trim())); + } + } + for (name, children) in directories { + let oid = make_tree(repo, &children); + items.push(format!("040000 tree {oid}\t{name}\n")); + } + git(repo, &["mktree"], Some(items.concat().as_bytes())) + .trim() + .to_string() +} + +#[derive(Debug, Clone)] +struct GitSource { + root: String, + commit: String, + scope_path: String, +} + +struct GitReader<'a> { + repo: &'a Path, +} + +impl FixedSourceReader for GitReader<'_> { + fn resolve(&mut self, source: &GitSource, path: &str) -> Result { + if path.is_empty() { + return Ok(SourceOutcome::Found(SourceNode { + fs_kind: FsKind::Directory, + oid: source.root.clone(), + })); + } + let mut tree = source.root.clone(); + let components: Vec<_> = path.split('/').collect(); + for (position, component) in components.iter().enumerate() { + let Some(entry) = self + .read_tree(&tree) + .into_iter() + .find(|entry| entry.name == *component) + else { + return Ok(SourceOutcome::AbsentProven); + }; + if position + 1 == components.len() { + return Ok(SourceOutcome::Found(entry.node)); + } + if entry.node.fs_kind != FsKind::Directory { + return Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "fixed Git path crossing", + )); + } + tree = entry.node.oid; + } + unreachable!() + } + + fn list(&mut self, source: &GitSource, path: &str) -> Result, SnapshotError> { + match self.resolve(source, path)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => { + Ok(self.read_tree(&node.oid)) + } + _ => Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "fixed Git listing", + )), + } + } +} + +impl GitReader<'_> { + fn read_tree(&self, oid: &str) -> Vec { + git(self.repo, &["ls-tree", "-z", oid], None) + .split('\0') + .filter(|row| !row.is_empty()) + .map(|row| { + let (metadata, name) = row.split_once('\t').unwrap(); + let fields: Vec<_> = metadata.split_whitespace().collect(); + SourceEntry { + name: name.to_string(), + node: SourceNode { + fs_kind: if fields[0] == "040000" { + FsKind::Directory + } else { + FsKind::Regular + }, + oid: fields[2].to_string(), + }, + } + }) + .collect() + } +} + +fn fixed_source(repo: &Path, files: &[(&str, &str)]) -> GitSource { + let root = make_tree(repo, files); + let commit = git( + repo, + &["commit-tree", &root], + Some(b"fixed oracle fixture\n"), + ) + .trim() + .to_string(); + GitSource { + root, + commit, + scope_path: "/".to_string(), + } +} + +#[test] +fn attested_scoped_root_is_not_prefixed_again_when_subpath_and_nested_mount_are_used() { + let repo = TempDir::new().unwrap(); + git( + repo.path(), + &["init", "--bare", "--object-format=sha1", "--quiet"], + None, + ); + let native = fixed_source(repo.path(), &[("native.txt", "native")]); + let full = fixed_source( + repo.path(), + &[ + ("scope/src/deep/leaf.txt", "scoped"), + ("other/outside.txt", "outside"), + ], + ); + let scoped = GitSource { + root: git( + repo.path(), + &["rev-parse", &format!("{}:scope", full.commit)], + None, + ) + .trim() + .to_string(), + commit: full.commit.clone(), + scope_path: "/scope".to_string(), + }; + let nested = fixed_source(repo.path(), &[("inside/nested.txt", "nested")]); + let mut reader = GitReader { repo: repo.path() }; + let index = FixedNamespaceIndex::new( + native, + vec![ + git_binding("/lib", &scoped, "src"), + git_binding("/lib/deep/nested", &nested, "inside"), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let route = index.route("/lib/deep/leaf.txt").unwrap(); + assert_eq!(route.source.scope_path, "/scope"); + assert_eq!(route.source_path, "src/deep/leaf.txt"); + let route = index.route("/lib/deep/nested/nested.txt").unwrap(); + assert_eq!(route.source_path, "inside/nested.txt"); + let mut files = BTreeMap::new(); + traverse(&index, &mut reader, "/", &mut files); + assert_eq!( + files, + BTreeMap::from([ + ("/native.txt".to_string(), b"native".to_vec()), + ("/lib/deep/leaf.txt".to_string(), b"scoped".to_vec()), + ( + "/lib/deep/nested/nested.txt".to_string(), + b"nested".to_vec() + ), + ]) + ); +} + +fn git_binding(path: &str, source: &GitSource, subpath: &str) -> FixedBinding { + FixedBinding { + mount_path: path.to_string(), + source: source.clone(), + source_kind: SourceKind::Import, + source_subpath: subpath.to_string(), + policy: BindingPolicy::Mutable, + } +} + +/// Independently flatten commits using Git, then compose a handwritten fixture. +/// This oracle does not call the production trie or source reader to select paths. +fn flatten_commit(repo: &Path, source: &GitSource) -> BTreeMap> { + git( + repo, + &["ls-tree", "-rz", "--full-tree", &source.commit], + None, + ) + .split('\0') + .filter(|row| !row.is_empty()) + .map(|row| { + let (metadata, path) = row.split_once('\t').unwrap(); + let oid = metadata.split_whitespace().nth(2).unwrap(); + ( + path.to_string(), + git(repo, &["cat-file", "blob", oid], None).into_bytes(), + ) + }) + .collect() +} + +fn traverse( + index: &FixedNamespaceIndex, + reader: &mut GitReader<'_>, + path: &str, + files: &mut BTreeMap>, +) { + for entry in index.list(path, reader).unwrap().unwrap().entries { + let child = if path == "/" { + format!("/{}", entry.name) + } else { + format!("{path}/{}", entry.name) + }; + if entry.fs_kind == FsKind::Directory { + traverse(index, reader, &child, files); + } else { + files.insert( + child, + git( + reader.repo, + &["cat-file", "blob", entry.oid.as_deref().unwrap()], + None, + ) + .into_bytes(), + ); + } + } +} + +#[test] +fn composed_listing_matches_independent_fixed_git_oracle_after_ref_and_path_reuse() { + let repo = TempDir::new().unwrap(); + git( + repo.path(), + &["init", "--bare", "--object-format=sha1", "--quiet"], + None, + ); + let native = fixed_source( + repo.path(), + &[ + ("project/app/native.txt", "native"), + ("lib/native-hidden.txt", "hidden"), + ("library/keep.txt", "neighbor"), + ], + ); + let a20 = fixed_source( + repo.path(), + &[ + ("src/deep/old.txt", "A20"), + ("src/vendor/b/hidden.txt", "outer-hidden"), + ("elsewhere/not-visible.txt", "outside-subpath"), + ], + ); + let a21 = fixed_source(repo.path(), &[("src/deep/new.txt", "A21")]); + let b30 = fixed_source(repo.path(), &[("nested.txt", "B30")]); + let bindings = vec![ + git_binding("/lib", &a20, "src"), + git_binding("/lib/vendor/b", &b30, ""), + git_binding("/missing/deep/import", &b30, ""), + ]; + let mut reader = GitReader { repo: repo.path() }; + let old = + FixedNamespaceIndex::new(native.clone(), bindings, BTreeSet::new(), &mut reader).unwrap(); + // Simulate advancing current refs and reassigning a registry path. The + // reader has only the captured fixed handle; no registry/ref read is possible. + git( + repo.path(), + &["update-ref", "refs/heads/import-current", &a21.commit], + None, + ); + let current_registry = BTreeMap::from([("/lib", a21.clone())]); + assert_ne!( + current_registry["/lib"].commit, + old.route("/lib/deep/old.txt").unwrap().source.commit + ); + let mut expected = BTreeMap::new(); + for (path, bytes) in flatten_commit(repo.path(), &native) { + if !path.starts_with("lib/") { + expected.insert(format!("/{path}"), bytes); + } + } + for (path, bytes) in flatten_commit(repo.path(), &a20) { + if let Some(suffix) = path.strip_prefix("src/") + && !suffix.starts_with("vendor/b/") + { + expected.insert(format!("/lib/{suffix}"), bytes); + } + } + for (path, bytes) in flatten_commit(repo.path(), &b30) { + expected.insert(format!("/lib/vendor/b/{path}"), bytes.clone()); + expected.insert(format!("/missing/deep/import/{path}"), bytes); + } + let mut actual = BTreeMap::new(); + traverse(&old, &mut reader, "/", &mut actual); + assert_eq!(actual, expected); + assert!(actual.contains_key("/lib/deep/old.txt")); + assert!(!actual.contains_key("/lib/deep/new.txt")); + let unbound = + FixedNamespaceIndex::new(native.clone(), Vec::new(), BTreeSet::new(), &mut reader).unwrap(); + let mut restored = BTreeMap::new(); + traverse(&unbound, &mut reader, "/", &mut restored); + let expected_native = flatten_commit(repo.path(), &native) + .into_iter() + .map(|(path, bytes)| (format!("/{path}"), bytes)) + .collect::>(); + assert_eq!(restored, expected_native); + let mut old_again = BTreeMap::new(); + traverse(&old, &mut reader, "/", &mut old_again); + assert_eq!(old_again, expected); +} diff --git a/src/ceres/snapshot/pages.rs b/src/ceres/snapshot/pages.rs index 74824a38..555ed913 100644 --- a/src/ceres/snapshot/pages.rs +++ b/src/ceres/snapshot/pages.rs @@ -15,13 +15,16 @@ use mst2_codec::metapage::{Entry, EntryKind, Page, page_id}; use sea_orm::ActiveValue::Set; use sha2::{Digest, Sha256}; -use crate::ceres::{ - api_service::ApiHandler, - snapshot::{ - error::{SnapshotError, SnapshotErrorCode}, - resolver::FsKind, - view::hex, +use crate::{ + ceres::{ + api_service::ApiHandler, + snapshot::{ + error::{SnapshotError, SnapshotErrorCode}, + resolver::FsKind, + view::hex, + }, }, + jupiter::storage::mono_storage::MST2_VERIFICATION_VERSION, }; /// A directory page plus everything the JSON layer needs. @@ -99,37 +102,43 @@ async fn build_directory_page_uncached( let dirents = crate::ceres::snapshot::resolver::direct_entries(&tree)?; // T03 write-through verification: consult verified records first, then - // fetch + hash the misses and persist them. A verification-record read - // failure degrades to recomputation (safe: strictly more verification). + // fetch + hash the misses and persist them. Invalid records and lookup + // failures are errors, never a missing verification fact. let blob_oids: Vec = dirents .iter() .filter(|(_, k, _)| *k != FsKind::Directory) .map(|(_, _, oid)| oid.clone()) .collect(); - let verified = match handler + let verified = handler .get_context() .mono_storage() .get_verified_blobs(blob_oids) .await - { - Ok(map) => map, - Err(e) => { - tracing::warn!(error = %e, "verified blob lookup failed; recomputing digests"); - HashMap::new() - } - }; + .map_err(|e| { + let code = if matches!( + e, + crate::common::errors::MegaError::ObjStorageInconsistent(_) + ) { + SnapshotErrorCode::IntegrityError + } else { + SnapshotErrorCode::Internal + }; + tracing::warn!(error = %e, "verified blob lookup failed"); + SnapshotError::new(code, "verified blob lookup failed") + })?; - let mut new_verified: Vec = Vec::new(); + let mut new_verified = HashMap::new(); let mut entries = Vec::with_capacity(dirents.len()); let mut codec_entries = Vec::with_capacity(dirents.len()); for (name, fs_kind, oid) in dirents { + let child_rel = if rel_path == "/" { + format!("/{name}") + } else { + format!("{rel_path}/{name}") + }; + crate::ceres::snapshot::view::validate_scope_relative_path(&child_rel)?; match fs_kind { FsKind::Directory => { - let child_rel = if rel_path == "/" { - format!("/{name}") - } else { - format!("{rel_path}/{name}") - }; let child_page = Box::pin(build_directory_page(handler, root_tree, &child_rel)).await?; codec_entries.push(Entry::dir(name.as_bytes(), child_page.page_id)); @@ -145,24 +154,37 @@ async fn build_directory_page_uncached( FsKind::Regular | FsKind::Executable | FsKind::Symlink => { let (size, digest) = if let Some(v) = verified.get(&oid) { // Verified record: 64-bit size + raw digest, no content read. - let d: [u8; 32] = v.raw_sha256.clone().try_into().unwrap_or([0u8; 32]); - (v.size as u64, d) + let d: [u8; 32] = v.raw_sha256.as_slice().try_into().map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "invalid verified content digest", + ) + })?; + let size = u64::try_from(v.size).map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "invalid verified content size", + ) + })?; + (size, d) } else { let raw = fetch_raw_blob(handler, &oid).await?; let mut h = Sha256::new(); h.update(&raw); let digest: [u8; 32] = h.finalize().into(); - new_verified.push(crate::callisto::mst2_verified_object::ActiveModel { - id: sea_orm::ActiveValue::NotSet, - storage_domain: Set("git".to_string()), - git_oid: Set(oid.clone()), - object_kind: Set("blob".to_string()), - raw_sha256: Set(digest.to_vec()), - size: Set(raw.len() as i64), - verification_version: Set(1), - state: Set("VERIFIED".to_string()), - created_at: Set(chrono_now()), - }); + new_verified.entry(oid.clone()).or_insert( + crate::callisto::mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(digest.to_vec()), + size: Set(raw.len() as i64), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono_now()), + }, + ); (raw.len() as u64, digest) }; let kind = match fs_kind { @@ -189,7 +211,7 @@ async fn build_directory_page_uncached( if let Err(e) = handler .get_context() .mono_storage() - .insert_verified_blobs(new_verified) + .insert_verified_blobs(new_verified.into_values().collect()) .await { tracing::warn!(error = %e, "verified blob persistence failed"); @@ -242,6 +264,7 @@ async fn fetch_tree( root_tree: &git_internal::internal::object::tree::Tree, rel_path: &str, ) -> Result { + crate::ceres::snapshot::view::validate_scope_relative_path(rel_path)?; if rel_path == "/" { return Ok(root_tree.clone()); } @@ -329,6 +352,7 @@ pub async fn resolve_abs( root_tree: &git_internal::internal::object::tree::Tree, abs_path: &str, ) -> Result { + crate::ceres::snapshot::view::validate_scope_relative_path(abs_path)?; if abs_path == "/" { return Ok(WalkOutcome::FoundDir); } @@ -379,7 +403,7 @@ pub async fn resolve_abs( unreachable!("loop returns on the last component") } -/// Raw blob bytes with the `blob \0` git header stripped. +/// GitService stores and returns raw file content, without an object header. pub async fn fetch_raw_blob( handler: &T, oid: &str, @@ -387,24 +411,19 @@ pub async fn fetch_raw_blob( let data = handler.get_raw_blob_by_hash(oid).await.map_err(|e| { // Missing/corrupt content must be an error, never an empty file // (spec 00 SYS-04). - SnapshotError::new( - SnapshotErrorCode::Internal, - format!("blob fetch failed for {oid}: {e}"), - ) + let code = match e { + crate::common::errors::MegaError::ObjStorageNotFound(_) => { + SnapshotErrorCode::ObjectUnavailable + } + crate::common::errors::MegaError::ObjStorageInconsistent(_) => { + SnapshotErrorCode::IntegrityError + } + _ => SnapshotErrorCode::Internal, + }; + tracing::warn!(error = %e, "fixed-view blob fetch failed"); + SnapshotError::new(code, "fixed-view content could not be read") })?; - Ok(strip_git_blob_header(&data)) -} - -fn strip_git_blob_header(data: &[u8]) -> Vec { - if let Some(pos) = data.iter().position(|&b| b == 0) - && data.starts_with(b"blob ") - && let Ok(len) = std::str::from_utf8(&data[5..pos]).map(|s| s.parse::()) - && let Ok(len) = len - && data.len() == pos + 1 + len - { - return data[pos + 1..].to_vec(); - } - data.to_vec() + Ok(data) } pub fn sha256_hex(data: &[u8]) -> String { @@ -426,3 +445,486 @@ fn chrono_now() -> chrono::DateTime { pub fn hex_of(id: &[u8; 32]) -> String { hex(id) } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use axum::{body::Body, http::Request}; + use bytes::Bytes; + use git_internal::internal::{ + metadata::EntryMeta, + object::{ + blob::Blob, + tree::{Tree, TreeItem, TreeItemMode}, + }, + }; + use sea_orm::{ActiveValue::Set, EntityTrait, IntoActiveModel}; + use tower::ServiceExt; + + use super::*; + use crate::{ + api::{ + MonoApiServiceState, oauth::api_store::BrowserSessionStore, router::snapshot_router, + }, + callisto::{mega_refs, mst2_verified_object}, + ceres::api_service::cache::GitObjectCache, + config::testing::isolated_config, + jupiter::{ + storage::base_storage::StorageConnector, + tests::{test_redis_manager, test_storage_with_config}, + utils::converter::IntoMegaModel, + }, + }; + + async fn state(temp: &std::path::Path) -> MonoApiServiceState { + let mut config = isolated_config(temp.join("config")); + config.mst2.enabled = true; + config.mst2.instance_uuid = Some(uuid::Uuid::new_v4().to_string()); + config.mst2.auth_token = Some("mst2-projection-test".to_string()); + let storage = test_storage_with_config(temp, config).await; + MonoApiServiceState { + entity_store: storage.entity_store.clone(), + storage, + session_store: BrowserSessionStore::Anonymous, + git_object_cache: Arc::new(GitObjectCache { + connection: test_redis_manager().await, + prefix: String::new(), + }), + listen_addr: "127.0.0.1:0".to_string(), + } + } + + async fn seed(state: &MonoApiServiceState, filename: &str, raw: &[u8]) -> String { + seed_path(state, &format!("/{filename}"), raw).await + } + + async fn seed_path(state: &MonoApiServiceState, path: &str, raw: &[u8]) -> String { + let blob = Blob::from_content_bytes(raw.to_vec()); + state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(raw)) + .await + .unwrap(); + let mut child_id = blob.id; + let mut child_mode = TreeItemMode::Blob; + let mut trees = Vec::new(); + for name in path[1..].split('/').rev() { + let tree = + Tree::from_tree_items(vec![TreeItem::new(child_mode, child_id, name.to_string())]) + .unwrap(); + child_id = tree.id; + child_mode = TreeItemMode::Tree; + trees.push(tree.into_mega_model(EntryMeta::new()).into_active_model()); + } + let mono = state.storage.mono_storage(); + mono.batch_save_model(trees).await.unwrap(); + mono.save_refs( + mega_refs::Model::new( + "/", + "refs/heads/main".to_string(), + blob.id.to_string(), + child_id.to_string(), + false, + ), + None, + ) + .await + .unwrap(); + blob.id.to_string() + } + + async fn seed_legacy(state: &MonoApiServiceState, oid: &str, raw: &[u8]) { + mst2_verified_object::Entity::insert(mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.to_string()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(Sha256::digest(raw).to_vec()), + size: Set(raw.len() as i64), + verification_version: Set(1), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }) + .exec(state.storage.mono_storage().get_connection()) + .await + .unwrap(); + } + + async fn snapshot_id(state: &MonoApiServiceState, scope: &str) -> String { + let main = state + .storage + .mono_storage() + .get_main_ref("/") + .await + .unwrap() + .unwrap(); + let handler = crate::ceres::api_service::mono_api_service::MonoApiService::from(state); + let tree = handler.get_tree_by_hash(&main.ref_tree_hash).await.unwrap(); + let built = build_directory_page(&handler, &tree, scope).await.unwrap(); + crate::ceres::snapshot::descriptor::build( + &state.storage.config().mst2, + &crate::ceres::snapshot::view::SnapshotView::from_commit( + &main.ref_commit_hash, + &main.ref_tree_hash, + ), + scope, + built.page_id, + ) + .unwrap() + .snapshot_id + } + + fn request(method: &str, uri: &str, lease: Option<&str>, body: Body) -> Request { + let mut request = Request::builder() + .method(method) + .uri(uri) + .header("authorization", "Bearer mst2-projection-test") + .header("content-type", "application/json"); + if let Some(lease) = lease { + request = request.header("x-mega-snapshot-lease", lease); + } + request.body(body).unwrap() + } + + #[tokio::test] + async fn mst2_raw_header_like_bytes_survive_projection_and_http_reads() { + assert_header_like_blob_http(false).await; + } + + #[tokio::test] + async fn mst2_legacy_header_like_fact_is_reverified_before_http_projection() { + assert_header_like_blob_http(true).await; + } + + async fn assert_header_like_blob_http(legacy: bool) { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let raw = b"blob 3\0abc"; + let filename = format!("binary-{}", uuid::Uuid::new_v4()); + let oid = seed(&state, &filename, raw).await; + if legacy { + seed_legacy(&state, &oid, b"abc").await; + } + let app = snapshot_router::routers(state.clone()).with_state(state.clone()); + let resolve = app + .clone() + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(resolve.status(), 200); + let resolved: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(resolve.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + let snapshot = snapshot_id(&state, "/").await; + let lease = resolved["lease_id"].as_str().unwrap(); + let digest = sha256_hex(raw); + let rows = state + .storage + .mono_storage() + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap(); + assert_eq!(rows[&oid].size, raw.len() as i64); + assert_eq!(rows[&oid].raw_sha256, Sha256::digest(raw).to_vec()); + assert_eq!(rows[&oid].verification_version, MST2_VERIFICATION_VERSION); + + let directory = app + .clone() + .oneshot(request( + "GET", + &format!("/snapshots/{snapshot}/directory"), + Some(lease), + Body::empty(), + )) + .await + .unwrap(); + assert_eq!(directory.status(), 200); + let directory: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(directory.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(directory["entries"][0]["size"], raw.len().to_string()); + assert_eq!( + directory["entries"][0]["content_digest"], + format!("sha256:{digest}") + ); + let uri = + format!("/snapshots/{snapshot}/blob?path=/{filename}&expected_digest=sha256:{digest}"); + let get = app + .clone() + .oneshot(request("GET", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(get.status(), 200); + assert_eq!( + axum::body::to_bytes(get.into_body(), 128) + .await + .unwrap() + .as_ref(), + raw + ); + let head = app + .clone() + .oneshot(request("HEAD", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(head.status(), 200); + assert_eq!(head.headers()["content-length"], raw.len().to_string()); + assert!( + axum::body::to_bytes(head.into_body(), 128) + .await + .unwrap() + .is_empty() + ); + let again = app + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(again.status(), 200); + let again: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(again.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(again["descriptor"], resolved["descriptor"]); + crate::ceres::snapshot::runtime::runtime() + .release_lease(again["lease_id"].as_str().unwrap()); + crate::ceres::snapshot::runtime::runtime().release_lease(lease); + } + + #[tokio::test] + async fn mst2_invalid_verified_record_is_not_recomputed_into_a_successful_snapshot() { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let filename = format!("corrupt-{}", uuid::Uuid::new_v4()); + let raw = b"valid stored content"; + let oid = seed(&state, &filename, raw).await; + mst2_verified_object::Entity::insert(mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![7; 31]), + size: Set(raw.len() as i64), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }) + .exec(state.storage.mono_storage().get_connection()) + .await + .unwrap(); + let response = snapshot_router::routers(state.clone()) + .with_state(state) + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(response.status(), 502); + let body: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["error"]["code"], "INTEGRITY_ERROR"); + } + + #[tokio::test] + async fn mst2_legacy_fact_cannot_upgrade_when_raw_storage_is_unavailable() { + use sea_orm::{ColumnTrait, QueryFilter}; + + use crate::orbit_api::object_storage::{ObjectKey, ObjectNamespace}; + + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let filename = format!("missing-{}", uuid::Uuid::new_v4()); + let oid = seed(&state, &filename, b"blob 3\0abc").await; + seed_legacy(&state, &oid, b"abc").await; + state + .storage + .git_service + .obj_storage + .inner + .delete(&ObjectKey { + namespace: ObjectNamespace::Git, + key: oid.clone(), + }) + .await + .unwrap(); + let response = snapshot_router::routers(state.clone()) + .with_state(state.clone()) + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(response.status(), 503); + let body: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["error"]["code"], "OBJECT_UNAVAILABLE"); + let saved = mst2_verified_object::Entity::find() + .filter(mst2_verified_object::Column::GitOid.eq(oid)) + .one(state.storage.mono_storage().get_connection()) + .await + .unwrap() + .unwrap(); + assert_eq!(saved.verification_version, 1); + assert_eq!(saved.size, 3); + assert_eq!(saved.raw_sha256, Sha256::digest(b"abc").to_vec()); + } + + #[tokio::test] + async fn mst2_shared_blob_paths_upgrade_one_verification_record() { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let raw = b"blob 3\0abc"; + let oid = state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(raw)) + .await + .unwrap(); + seed_legacy(&state, &oid, b"abc").await; + let blob = Blob::from_content_bytes(raw.to_vec()); + let tree = Tree::from_tree_items(vec![ + TreeItem::new( + TreeItemMode::Blob, + blob.id, + format!("a-{}", uuid::Uuid::new_v4()), + ), + TreeItem::new( + TreeItemMode::Blob, + blob.id, + format!("b-{}", uuid::Uuid::new_v4()), + ), + ]) + .unwrap(); + let handler = crate::ceres::api_service::mono_api_service::MonoApiService::from(&state); + let built = build_directory_page_uncached(&handler, &tree, "/") + .await + .unwrap(); + assert_eq!(built.entries.len(), 2); + assert!(built.entries.iter().all(|entry| entry.size == Some(10) + && entry.content_digest == Some(Sha256::digest(raw).into()))); + let rows = state + .storage + .mono_storage() + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[&oid].size, 10); + assert_eq!(rows[&oid].verification_version, MST2_VERIFICATION_VERSION); + } + + #[tokio::test] + async fn mst2_projection_checks_full_leaf_path_before_resolve_and_http_reads() { + for (component_boundary, valid) in + [(true, true), (true, false), (false, true), (false, false)] + { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let marker = uuid::Uuid::new_v4().to_string(); + let mut components = if component_boundary { + let mut parts = vec!["d".to_string(); if valid { 255 } else { 256 }]; + parts[0] = marker; + parts.push("file".to_string()); + parts + } else { + let mut parts = vec!["d".repeat(255); 15]; + parts[0].replace_range(..marker.len(), &marker); + parts.push("d".repeat(127)); + parts.push("f".repeat(if valid { 127 } else { 128 })); + parts + }; + let filename = components.pop().unwrap(); + let scope = format!("/{}", components.join("/")); + let full_path = format!("{scope}/{filename}"); + if component_boundary { + assert_eq!( + full_path[1..].split('/').count(), + if valid { 256 } else { 257 } + ); + } else { + assert_eq!(full_path.len(), if valid { 4096 } else { 4097 }); + } + let raw = b"boundary file bytes"; + seed_path(&state, &full_path, raw).await; + let app = snapshot_router::routers(state.clone()).with_state(state.clone()); + let resolve = app + .clone() + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from( + serde_json::json!({"target":{"kind":"latest"}, "scope":scope}).to_string(), + ), + )) + .await + .unwrap(); + let status = resolve.status(); + let resolved: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(resolve.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + if !valid { + assert_eq!(status, 400); + assert_eq!(resolved["error"]["code"], "SCOPE_INVALID"); + assert!(resolved.get("lease_id").is_none()); + continue; + } + assert_eq!(status, 200); + let snapshot = snapshot_id(&state, &scope).await; + let lease = resolved["lease_id"].as_str().unwrap(); + let uri = format!( + "/snapshots/{snapshot}/blob?path=/{filename}&expected_digest=sha256:{}", + sha256_hex(raw) + ); + let get = app + .oneshot(request("GET", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(get.status(), 200); + assert_eq!( + axum::body::to_bytes(get.into_body(), 128) + .await + .unwrap() + .as_ref(), + raw + ); + crate::ceres::snapshot::runtime::runtime().release_lease(lease); + } + } +} diff --git a/src/ceres/snapshot/runtime.rs b/src/ceres/snapshot/runtime.rs index 03cafef6..770c4725 100644 --- a/src/ceres/snapshot/runtime.rs +++ b/src/ceres/snapshot/runtime.rs @@ -349,7 +349,15 @@ mod tests { use crate::ceres::snapshot::retention::{ NodeState, RetainedKind, RetentionRoot, RetentionStore as _, }; - let r = runtime(); + let r = Mst2Runtime { + hmac_key: blake3_key(), + contexts: Mutex::new(HashMap::new()), + leases: Mutex::new(HashMap::new()), + retention: crate::ceres::snapshot::retention::RetentionCoordinator::new( + crate::ceres::snapshot::retention::mem::InMemoryRetentionStore::default(), + ), + tip_state: Mutex::new((String::new(), 0u64)), + }; // Retain the snapshot's derived nodes directly, covered by the lease // root that insert_context created. let lease_root = RetentionRoot::Lease("gc-lease-1".to_string()); diff --git a/src/ceres/snapshot/view.rs b/src/ceres/snapshot/view.rs index 5f81cba5..88d1352c 100644 --- a/src/ceres/snapshot/view.rs +++ b/src/ceres/snapshot/view.rs @@ -62,15 +62,20 @@ pub fn validate_scope_relative_path(path: &str) -> Result<(), SnapshotError> { if path == "/" { return Ok(()); } + let mut component_count = 0; for comp in path[1..].split('/') { + component_count += 1; + if component_count > 256 { + return Err(invalid("path has more than 256 components")); + } if comp.is_empty() { return Err(invalid("empty path component")); } if comp == "." || comp == ".." { return Err(invalid("dot component")); } - if comp.len() > 256 { - return Err(invalid("path component over 256 bytes")); + if comp.len() > 255 { + return Err(invalid("path component over 255 bytes")); } } Ok(()) @@ -99,4 +104,14 @@ mod tests { assert!(validate_scope_relative_path("/a/").is_err()); assert!(validate_scope_relative_path("/a\0").is_err()); } + + #[test] + fn scope_relative_path_enforces_byte_and_component_boundaries() { + assert!(validate_scope_relative_path(&format!("/{}", "a".repeat(255))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", "a".repeat(256))).is_err()); + assert!(validate_scope_relative_path(&format!("/{}a", "é".repeat(127))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", "é".repeat(128))).is_err()); + assert!(validate_scope_relative_path(&format!("/{}", vec!["a"; 256].join("/"))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", vec!["a"; 257].join("/"))).is_err()); + } } diff --git a/src/jupiter/storage/mono_storage.rs b/src/jupiter/storage/mono_storage.rs index 550d8ced..dc140b2e 100644 --- a/src/jupiter/storage/mono_storage.rs +++ b/src/jupiter/storage/mono_storage.rs @@ -47,6 +47,10 @@ pub struct MonoStorage { pub base: BaseStorage, } +/// Generation 2 hashes the raw file body without guessing a Git envelope. +pub const MST2_VERIFICATION_VERSION: i32 = 2; +const MST2_MAX_FILE_SIZE: i64 = 8_796_093_022_208; + impl Deref for MonoStorage { type Target = BaseStorage; fn deref(&self) -> &Self::Target { @@ -1730,12 +1734,29 @@ impl MonoStorage { .filter(mst2_verified_object::Column::GitOid.is_in(oids)) .all(self.get_connection()) .await?; - Ok(rows.into_iter().map(|m| (m.git_oid.clone(), m)).collect()) + for row in &rows { + if row.state != "VERIFIED" + || ![1, MST2_VERIFICATION_VERSION].contains(&row.verification_version) + || !(0..=MST2_MAX_FILE_SIZE).contains(&row.size) + || row.raw_sha256.len() != 32 + { + return Err(MegaError::ObjStorageInconsistent( + "invalid MST/2 verified blob record".to_string(), + )); + } + } + // Generation 1 may have stripped header-like file bytes. Reverify + // valid legacy rows from object storage before trusting their facts. + Ok(rows + .into_iter() + .filter(|m| m.verification_version == MST2_VERIFICATION_VERSION) + .map(|m| (m.git_oid.clone(), m)) + .collect()) } /// Write-through verification: rows are only written after the content /// was fetched and hashed. Conflicts (concurrent verification of the same - /// oid) resolve to the existing row. + /// oid) keep current facts. Only valid generation-1 facts may be upgraded. pub async fn insert_verified_blobs( &self, rows: Vec, @@ -1743,6 +1764,23 @@ impl MonoStorage { if rows.is_empty() { return Ok(()); } + for row in &rows { + if row.state.try_as_ref().map(String::as_str) != Some("VERIFIED") + || row.verification_version.try_as_ref() != Some(&MST2_VERIFICATION_VERSION) + || !row + .size + .try_as_ref() + .is_some_and(|size| (0..=MST2_MAX_FILE_SIZE).contains(size)) + || !row + .raw_sha256 + .try_as_ref() + .is_some_and(|digest| digest.len() == 32) + { + return Err(MegaError::ObjStorageInconsistent( + "invalid MST/2 verified blob write".to_string(), + )); + } + } match mst2_verified_object::Entity::insert_many(rows) .on_conflict( OnConflict::columns([ @@ -1750,15 +1788,43 @@ impl MonoStorage { mst2_verified_object::Column::GitOid, mst2_verified_object::Column::ObjectKind, ]) - .do_nothing() + .update_columns([ + mst2_verified_object::Column::RawSha256, + mst2_verified_object::Column::Size, + mst2_verified_object::Column::VerificationVersion, + mst2_verified_object::Column::CreatedAt, + ]) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::VerificationVersion, + )) + .eq(1), + ) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::State, + )) + .eq("VERIFIED"), + ) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::Size, + )) + .between(0, MST2_MAX_FILE_SIZE), + ) + .action_and_where(Expr::cust( + "length(\"mst2_verified_object\".\"raw_sha256\") = 32", + )) .to_owned(), ) .exec(self.get_connection()) .await { - // ON CONFLICT DO NOTHING returns RecordNotInserted when every - // row was a duplicate; that is the intended first-write-wins - // outcome (same convention as next_insert_retry). + // A skipped conditional conflict returns RecordNotInserted; + // current-generation facts retain first-write-wins semantics. Ok(_) | Err(DbErr::RecordNotInserted) => Ok(()), Err(e) => Err(e.into()), } @@ -3339,7 +3405,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(digest.clone()), size: Set(4_294_967_296), // > i32::MAX: the size64 gap this table closes - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3364,7 +3430,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(vec![9u8; 32]), size: Set(1), - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3381,8 +3447,7 @@ mod tests { #[tokio::test] async fn mst2_verified_blob_insert_real_error_is_not_swallowed() { - // The ON CONFLICT DO NOTHING sentinel (RecordNotInserted on an - // all-duplicate batch) is treated as idempotent success; a genuine + // A skipped conditional conflict is idempotent success; a genuine // database error must still surface, never be masked as Ok. let temp = tempfile::TempDir::new().unwrap(); let storage = test_storage(temp.path()).await; @@ -3402,7 +3467,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(vec![2u8; 32]), size: Set(2), - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3412,4 +3477,169 @@ mod tests { "a real DB error must propagate, not be swallowed by the ON CONFLICT sentinel match" ); } + + #[tokio::test] + async fn mst2_verified_blob_invalid_records_reject_the_entire_lookup() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + let valid_oid = "f".repeat(40); + let row = |oid: String, state: &str, version, size, digest: Vec| { + mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(digest), + size: Set(size), + verification_version: Set(version), + state: Set(state.to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + } + }; + mono.insert_verified_blobs(vec![row( + valid_oid.clone(), + "VERIFIED", + MST2_VERIFICATION_VERSION, + 0, + vec![7; 32], + )]) + .await + .unwrap(); + for (index, (state, version, size, digest_len)) in [ + ("PENDING", 1, 1, 32), + ("VERIFIED", 0, 1, 32), + ("VERIFIED", 3, 1, 32), + ("VERIFIED", 1, -1, 32), + ("VERIFIED", 1, 8_796_093_022_209, 32), + ("VERIFIED", 1, 1, 31), + ("VERIFIED", 1, 1, 33), + ("VERIFIED", MST2_VERIFICATION_VERSION, -1, 32), + ("VERIFIED", MST2_VERIFICATION_VERSION, 8_796_093_022_209, 32), + ("VERIFIED", MST2_VERIFICATION_VERSION, 1, 31), + ("VERIFIED", MST2_VERIFICATION_VERSION, 1, 33), + ] + .into_iter() + .enumerate() + { + let bad_oid = format!("{index:040x}"); + mst2_verified_object::Entity::insert(row( + bad_oid.clone(), + state, + version, + size, + vec![8; digest_len], + )) + .exec(mono.get_connection()) + .await + .unwrap(); + assert!(matches!( + mono.get_verified_blobs(vec![valid_oid.clone(), bad_oid]) + .await, + Err(MegaError::ObjStorageInconsistent(_)) + )); + assert_eq!( + mono.get_verified_blobs(vec![valid_oid.clone()]) + .await + .unwrap() + .len(), + 1 + ); + } + } + + #[tokio::test] + async fn mst2_verified_blob_accepts_the_eight_tib_boundary() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + let oid = "e".repeat(40); + mono.insert_verified_blobs(vec![mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![9; 32]), + size: Set(8_796_093_022_208), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }]) + .await + .unwrap(); + let rows = mono.get_verified_blobs(vec![oid.clone()]).await.unwrap(); + assert_eq!(rows[&oid].size, 8_796_093_022_208); + } + + #[tokio::test] + async fn mst2_verified_blob_upgrade_only_replaces_valid_legacy_facts() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + for (index, (version, state, size, digest_len)) in [ + (1, "VERIFIED", 3, 32), + (1, "PENDING", 3, 32), + (1, "VERIFIED", -1, 32), + (1, "VERIFIED", MST2_MAX_FILE_SIZE + 1, 32), + (1, "VERIFIED", 3, 31), + (MST2_VERIFICATION_VERSION, "VERIFIED", 3, 32), + (3, "VERIFIED", 3, 32), + ] + .into_iter() + .enumerate() + { + let oid = format!("{index:040x}"); + let legacy = mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![7; digest_len]), + size: Set(size), + verification_version: Set(version), + state: Set(state.to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }; + let old = mst2_verified_object::Entity::insert(legacy) + .exec_with_returning(mono.get_connection()) + .await + .unwrap(); + if index == 0 { + assert!( + mono.get_verified_blobs(vec![oid.clone()]) + .await + .unwrap() + .is_empty() + ); + } + mono.insert_verified_blobs(vec![mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![9; 32]), + size: Set(10), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }]) + .await + .unwrap(); + let saved = mst2_verified_object::Entity::find_by_id(old.id) + .one(mono.get_connection()) + .await + .unwrap() + .unwrap(); + if index == 0 { + assert_eq!(saved.verification_version, MST2_VERIFICATION_VERSION); + assert_eq!(saved.size, 10); + assert_eq!(saved.raw_sha256, vec![9; 32]); + } else { + assert_eq!( + saved, old, + "invalid, unknown and current rows are never replaced" + ); + } + } + } } diff --git a/tests/integration_api_write_trunk.rs b/tests/integration_api_write_trunk.rs index 938f5a6e..569340ce 100644 --- a/tests/integration_api_write_trunk.rs +++ b/tests/integration_api_write_trunk.rs @@ -18,7 +18,7 @@ mod git_cli; use std::{ fs, io::Read, - net::{TcpListener, TcpStream}, + net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -235,6 +235,7 @@ struct ServiceProcess { impl ServiceProcess { fn spawn(mut command: Command) -> Self { let child = command.spawn().expect("spawn mega2 service"); + git_cli::record_service_pid(child.id()); Self { child, reaped: false, @@ -711,11 +712,9 @@ fn with_runtime(future: impl std::future::Future) -> T { } fn reserve_free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind ephemeral port") - .local_addr() - .expect("local addr") - .port() + let port = git_cli::reserve_ephemeral_port(); + git_cli::record_allocated_port(port); + port } fn create_log_file(path: &Path) -> fs::File {