From 55615aae3f192cde0152a55aa05db21f591dac9e Mon Sep 17 00:00:00 2001 From: hmirin Date: Thu, 29 Oct 2026 12:56:54 +0900 Subject: [PATCH] Read untrusted storage through one bounded, regular-file-only path Size and file-type checks were applied per call site, so several reads escaped them: FORMAT and packs used a plain File::open (a FIFO hung every helper call), HEAD was checked by path before opening (a swap could still hang), carrier chunks were read whole with git show before their size was checked, historical restore had no byte budget, and writers could publish objects larger than readers accept. - src/bounded.rs opens storage files with O_NOFOLLOW | O_NONBLOCK, checks the opened handle is a regular file, and reads small files with a bound. Every directory-backend read goes through it. - An oversized or non-regular HEAD is a typed StoredHead observation: ordinary reads report an invalid head and recovery treats it as Invalid, instead of failing with a budget error. - Carrier chunks are read through cat-file --batch with the type and size checked from the header before the body; ls-tree output is bounded; historical restore charges every byte read, including rejected copies. - Object size limits are shared constants, and both backends refuse to stage a manifest or pack beyond them, before the compare-and-swap. Adds rustix (unix only) for the open flags. --- Cargo.lock | 2 + Cargo.toml | 6 + src/bounded.rs | 104 ++++ src/lib.rs | 1 + src/repository.rs | 26 +- src/storage.rs | 920 +++++++++++++++++++++------------- tests/carrier_chunk_bounds.rs | 121 +++++ tests/carrier_git.rs | 71 ++- tests/end_to_end.rs | 33 ++ 9 files changed, 928 insertions(+), 356 deletions(-) create mode 100644 src/bounded.rs create mode 100644 tests/carrier_chunk_bounds.rs diff --git a/Cargo.lock b/Cargo.lock index 8aac434..07dc502 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -529,9 +529,11 @@ dependencies = [ "hex", "hkdf", "hpke", + "libc", "rand 0.8.7", "rand 0.9.5", "rand_chacha 0.9.0", + "rustix", "serde", "serde_json", "sha2", diff --git a/Cargo.toml b/Cargo.toml index e4a59e1..316025a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,3 +39,9 @@ subtle = "2.6" thiserror = "2.0" tempfile = "3.14" zeroize = { version = "1.8", features = ["derive"] } + +[target.'cfg(unix)'.dependencies] +rustix = { version = "1.1", features = ["fs"] } + +[target.'cfg(unix)'.dev-dependencies] +libc = "0.2" diff --git a/src/bounded.rs b/src/bounded.rs new file mode 100644 index 0000000..67be690 --- /dev/null +++ b/src/bounded.rs @@ -0,0 +1,104 @@ +//! The one way to open a file that untrusted storage controls. Every +//! directory-backend read goes through `open_regular`, so the file-type check +//! happens on the opened handle and size limits apply before any allocation. + +use std::fs::File; +use std::io::{self, Read}; +use std::path::Path; + +pub(crate) enum Opened { + Absent, + /// A regular file, with its length at open time. + Regular(File, u64), + /// A symbolic link, FIFO, socket, device, or directory. + NotRegular, +} + +pub(crate) enum Bounded { + Absent, + Within(Vec), + /// A regular file longer than the limit; its contents were not read. + Oversized, + NotRegular, +} + +/// Opens `path` for reading without following a final symbolic link and +/// without blocking on a FIFO, then requires a regular file. +pub(crate) fn open_regular(path: &Path) -> io::Result { + let file = match open_no_follow(path)? { + Raw::Absent => return Ok(Opened::Absent), + Raw::Refused => return Ok(Opened::NotRegular), + Raw::Open(file) => file, + }; + let metadata = file.metadata()?; + if !metadata.is_file() { + return Ok(Opened::NotRegular); + } + clear_nonblocking(&file)?; + Ok(Opened::Regular(file, metadata.len())) +} + +/// Reads a regular file of at most `limit` bytes, rejecting a longer one by +/// its length before reading and again if it grew after opening. +pub(crate) fn read_regular_bounded(path: &Path, limit: u64) -> io::Result { + let (file, length) = match open_regular(path)? { + Opened::Absent => return Ok(Bounded::Absent), + Opened::NotRegular => return Ok(Bounded::NotRegular), + Opened::Regular(file, length) => (file, length), + }; + if length > limit { + return Ok(Bounded::Oversized); + } + let mut bytes = Vec::with_capacity(length as usize); + file.take(limit + 1).read_to_end(&mut bytes)?; + if bytes.len() as u64 > limit { + return Ok(Bounded::Oversized); + } + Ok(Bounded::Within(bytes)) +} + +enum Raw { + Absent, + /// The final component cannot be a regular file. + #[cfg_attr(not(unix), allow(dead_code))] + Refused, + Open(File), +} + +#[cfg(unix)] +fn open_no_follow(path: &Path) -> io::Result { + use rustix::fs::{Mode, OFlags}; + use rustix::io::Errno; + + let flags = OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC; + match rustix::fs::open(path, flags, Mode::empty()) { + Ok(fd) => Ok(Raw::Open(File::from(fd))), + Err(Errno::NOENT) => Ok(Raw::Absent), + // A final symlink (ELOOP) or a socket (ENXIO, EOPNOTSUPP). + Err(Errno::LOOP | Errno::NXIO | Errno::OPNOTSUPP) => Ok(Raw::Refused), + Err(error) => Err(error.into()), + } +} + +/// Windows has no FIFOs in the filesystem namespace, but this follows +/// symbolic links: a link to a regular file reads as that file. +#[cfg(not(unix))] +fn open_no_follow(path: &Path) -> io::Result { + match File::open(path) { + Ok(file) => Ok(Raw::Open(file)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(Raw::Absent), + Err(error) => Err(error), + } +} + +#[cfg(unix)] +fn clear_nonblocking(file: &File) -> io::Result<()> { + let flags = rustix::fs::fcntl_getfl(file)?; + rustix::fs::fcntl_setfl(file, flags - rustix::fs::OFlags::NONBLOCK)?; + Ok(()) +} + +#[cfg(not(unix))] +fn clear_nonblocking(_file: &File) -> io::Result<()> { + Ok(()) +} diff --git a/src/lib.rs b/src/lib.rs index dd312a7..4e013c7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ +mod bounded; pub mod continuity; pub mod crypto; pub mod error; diff --git a/src/repository.rs b/src/repository.rs index 4d6b221..7bc1cf1 100644 --- a/src/repository.rs +++ b/src/repository.rs @@ -23,7 +23,7 @@ use crate::manifest::{ verify_manifest, wrap_predecessor_key, }; use crate::policy::{DeviceRecord, DeviceRoles, PolicyState}; -use crate::storage::{HeadObservation, ObjectKind, Storage, StorageOccupied}; +use crate::storage::{HeadObservation, ObjectKind, Storage, StorageOccupied, StoredHead}; use crate::trace; const MAX_RECOVERY_OUTER_COMMITS: usize = 2048; @@ -854,10 +854,14 @@ impl EncryptedRepository { &mut |id| self.read_manifest_material(id), Some(&mut budget), ), - None if observation.head_bytes.is_some() => Err(proven_invalid(anyhow::anyhow!( - "storage HEAD is not a valid manifest object ID" - ))), - None => Err(MissingStorageHead.into()), + None => match &observation.head { + Some(head) => { + Err(proven_invalid(head.parse().err().unwrap_or_else(|| { + anyhow::anyhow!("storage HEAD is not a valid head") + }))) + } + None => Err(MissingStorageHead.into()), + }, }; match current_chain { Ok(chain) => { @@ -920,7 +924,11 @@ impl EncryptedRepository { if report.classification == RecoveryClass::Invalid { let evidence_id = observation.head_id.clone().unwrap_or_else(|| { hex::encode(Sha256::digest( - observation.head_bytes.as_deref().unwrap_or_default(), + observation + .head + .as_ref() + .map(StoredHead::evidence) + .unwrap_or_default(), )) }); record_rejected_head( @@ -1360,7 +1368,11 @@ impl EncryptedRepository { if self.storage_object_matches(kind, id)? { return Ok(()); } - if self.storage.restore_historical_object(kind, id)? + // Room for one corrupt maximal copy before the valid one. + let restore_budget = kind.max_bytes().saturating_mul(2); + if self + .storage + .restore_historical_object(kind, id, restore_budget)? && self.storage_object_matches(kind, id)? { return Ok(()); diff --git a/src/storage.rs b/src/storage.rs index ac3c213..25a2990 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -12,6 +12,7 @@ use rand::rngs::OsRng; use sha2::{Digest, Sha256}; use thiserror::Error; +use crate::bounded::{self, Bounded, Opened}; use crate::error::{BudgetExhausted, ObjectUnavailable}; use crate::trace; @@ -28,8 +29,13 @@ pub struct CasConflict { #[error("{0}")] pub struct StorageOccupied(pub String); -const MAX_BUFFERED_OBJECT_SIZE: u64 = 16 * 1024 * 1024; -const MAX_STORAGE_HEAD_BYTES: u64 = 4096; +// The size contract every writer enforces before publication and every +// reader enforces before allocating, so nothing published is unreadable. +pub const MAX_HEAD_BYTES: u64 = 4096; +pub const MAX_MANIFEST_BYTES: u64 = 16 * 1024 * 1024; +pub const CARRIER_CHUNK_BYTES: u64 = 32 * 1024 * 1024; +pub const MAX_CHUNKS_PER_OBJECT: u64 = 4096; +pub const MAX_OBJECT_BYTES: u64 = CARRIER_CHUNK_BYTES * MAX_CHUNKS_PER_OBJECT; pub trait ObjectStage: Write + Send { fn finish(self: Box, id: &str) -> Result<()>; @@ -42,21 +48,33 @@ trait ObjectStageSink: Write + Send + Sized { struct ValidatingObjectStage { sink: S, hasher: Sha256, + kind: ObjectKind, + written: u64, } impl ValidatingObjectStage { - fn wrap(sink: S) -> Box { + fn wrap(sink: S, kind: ObjectKind) -> Box { Box::new(Self { sink, hasher: Sha256::new(), + kind, + written: 0, }) } } impl Write for ValidatingObjectStage { fn write(&mut self, data: &[u8]) -> std::io::Result { + let limit = self.kind.max_bytes(); + if self.written.saturating_add(data.len() as u64) > limit { + return Err(std::io::Error::other(format!( + "{} object exceeds the {limit}-byte maximum that readers accept; splitting larger objects is not supported", + self.kind.name() + ))); + } let written = self.sink.write(data)?; self.hasher.update(&data[..written]); + self.written += written as u64; Ok(written) } @@ -68,7 +86,7 @@ impl Write for ValidatingObjectStage { impl ObjectStage for ValidatingObjectStage { fn finish(self: Box, id: &str) -> Result<()> { validate_id(id)?; - let Self { sink, hasher } = *self; + let Self { sink, hasher, .. } = *self; if hex::encode(hasher.finalize()) != id { bail!("staged object hash does not match id") } @@ -86,13 +104,13 @@ pub trait Storage { stage.finish(id) } + /// Buffers a whole object; the bound is the manifest limit, since + /// manifests are the objects read this way. fn get_object(&self, kind: ObjectKind, id: &str) -> Result> { - let mut reader = self - .open_object(kind, id)? - .take(MAX_BUFFERED_OBJECT_SIZE + 1); + let mut reader = self.open_object(kind, id)?.take(MAX_MANIFEST_BYTES + 1); let mut bytes = Vec::new(); reader.read_to_end(&mut bytes)?; - if bytes.len() as u64 > MAX_BUFFERED_OBJECT_SIZE { + if bytes.len() as u64 > MAX_MANIFEST_BYTES { bail!("buffered object exceeds size limit") } Ok(bytes) @@ -114,7 +132,7 @@ pub trait Storage { Ok(HeadObservation { head_id, token: bytes.clone(), - head_bytes: Some(bytes), + head: Some(StoredHead::Bytes(bytes)), }) } @@ -130,7 +148,14 @@ pub trait Storage { Ok(None) } - fn restore_historical_object(&self, _kind: ObjectKind, _id: &str) -> Result { + /// Restores a stored copy of `id` from storage history, reading at most + /// `max_bytes` across every copy it tries. + fn restore_historical_object( + &self, + _kind: ObjectKind, + _id: &str, + _max_bytes: u64, + ) -> Result { Ok(false) } @@ -143,7 +168,60 @@ pub trait Storage { pub struct HeadObservation { pub head_id: Option, pub token: Vec, - pub head_bytes: Option>, + /// What storage holds where HEAD belongs; `None` when nothing is there. + pub head: Option, +} + +/// The contents of storage HEAD. Only `Bytes` can name a manifest. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum StoredHead { + Bytes(Vec), + /// A symbolic link, FIFO, directory, or other non-regular entry. + NotRegular, + /// Longer than `MAX_HEAD_BYTES`; never read. + Oversized, +} + +impl StoredHead { + fn from_bounded(bounded: Bounded) -> Option { + match bounded { + Bounded::Absent => None, + Bounded::Within(bytes) => Some(Self::Bytes(bytes)), + Bounded::Oversized => Some(Self::Oversized), + Bounded::NotRegular => Some(Self::NotRegular), + } + } + + pub fn head_id(&self) -> Option { + let Self::Bytes(bytes) = self else { + return None; + }; + let value = std::str::from_utf8(bytes).ok()?.trim(); + validate_id(value).ok()?; + Some(value.to_owned()) + } + + /// The manifest ID, or why this HEAD names none. + pub fn parse(&self) -> Result { + match self { + Self::Bytes(_) => self + .head_id() + .context("storage HEAD is not a valid head: it does not name a manifest object ID"), + Self::NotRegular => bail!("storage HEAD is not a valid head: it is not a regular file"), + Self::Oversized => { + bail!("storage HEAD is not a valid head: it exceeds {MAX_HEAD_BYTES} bytes") + } + } + } + + /// Bytes that identify this observation in rejected-head evidence. + pub fn evidence(&self) -> &[u8] { + match self { + Self::Bytes(bytes) => bytes, + Self::NotRegular => b"storage HEAD is not a regular file", + Self::Oversized => b"storage HEAD is oversized", + } + } } #[derive(Clone, Debug)] @@ -207,13 +285,16 @@ fn check_storage_format(location: &str, format: Option<&[u8]>, head_present: boo bail!("storage at {location} uses format {found}; this git-remote-e2ee reads format 6") } -fn read_bounded_format(mut reader: impl Read) -> Result> { - let mut bytes = Vec::new(); - reader - .by_ref() - .take(MAX_STORAGE_FORMAT_BYTES) - .read_to_end(&mut bytes)?; - Ok(bytes) +/// Reads a small root file through the bounded opener. +fn read_root_file(path: &Path, max_bytes: u64) -> Result>> { + match bounded::read_regular_bounded(path, max_bytes) + .with_context(|| format!("read {}", path.display()))? + { + Bounded::Absent => Ok(None), + Bounded::Within(bytes) => Ok(Some(bytes)), + Bounded::Oversized => bail!("{} exceeds {max_bytes} bytes", path.display()), + Bounded::NotRegular => bail!("{} is not a regular file", path.display()), + } } impl ObjectKind { @@ -223,6 +304,20 @@ impl ObjectKind { Self::Manifest => "manifests", } } + + fn name(self) -> &'static str { + match self { + Self::Pack => "pack", + Self::Manifest => "manifest", + } + } + + pub fn max_bytes(self) -> u64 { + match self { + Self::Pack => MAX_OBJECT_BYTES, + Self::Manifest => MAX_MANIFEST_BYTES, + } + } } #[derive(Clone)] @@ -253,12 +348,8 @@ impl FilesystemStorage { } fn check_format(&self) -> Result<()> { - let format = match File::open(self.root.join("FORMAT")) { - Ok(file) => Some(read_bounded_format(file)?), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, - Err(error) => return Err(error.into()), - }; - let head_present = self.root.join("HEAD").exists(); + let format = read_root_file(&self.root.join("FORMAT"), MAX_STORAGE_FORMAT_BYTES)?; + let head_present = fs::symlink_metadata(self.root.join("HEAD")).is_ok(); check_storage_format( &self.root.display().to_string(), format.as_deref(), @@ -367,37 +458,48 @@ impl Storage for FilesystemStorage { .write(true) .create_new(true) .open(&temporary)?; - Ok(ValidatingObjectStage::wrap(FilesystemObjectStage { - file, - temporary, - root: self.root.clone(), + Ok(ValidatingObjectStage::wrap( + FilesystemObjectStage { + file, + temporary, + root: self.root.clone(), + kind, + }, kind, - })) + )) } fn open_object(&self, kind: ObjectKind, id: &str) -> Result> { let path = self.object_path(kind, id)?; - let file = File::open(&path) - .map_err(|source| ObjectUnavailable { - storage: "directory", - id: id.to_owned(), - source: Some(source), - }) - .with_context(|| format!("read {}", path.display()))?; - Ok(Box::new(file)) + let refused = |kind, reason: &str| Some(std::io::Error::new(kind, reason.to_owned())); + let source = match bounded::open_regular(&path) { + Ok(Opened::Regular(file, length)) if length <= kind.max_bytes() => { + return Ok(Box::new(file.take(kind.max_bytes()))); + } + Ok(Opened::Regular(..)) => refused( + std::io::ErrorKind::InvalidData, + "object exceeds the maximum object size", + ), + Ok(Opened::NotRegular) => refused( + std::io::ErrorKind::InvalidData, + "object is not a regular file", + ), + Ok(Opened::Absent) => Some(std::io::ErrorKind::NotFound.into()), + Err(error) => Some(error), + }; + Err(ObjectUnavailable { + storage: "directory", + id: id.to_owned(), + source, + }) + .with_context(|| format!("read {}", path.display())) } fn read_head(&self) -> Result> { self.check_format()?; - let Some(bytes) = read_bounded_head_file(&self.root.join("HEAD"))? else { - return Ok(None); - }; - let value = std::str::from_utf8(&bytes) - .context("invalid opaque object id")? - .trim() - .to_owned(); - validate_id(&value)?; - Ok(Some(value)) + read_stored_head(&self.root.join("HEAD"))? + .map(|head| head.parse()) + .transpose() } fn compare_and_swap_head(&self, expected: Option<&str>, next: &str) -> Result<()> { @@ -427,18 +529,14 @@ impl Storage for FilesystemStorage { fn observe_head(&self) -> Result { self.check_format()?; - let head_bytes = read_bounded_head_file(&self.root.join("HEAD"))?; - let head_id = head_bytes.as_deref().and_then(|bytes| { - std::str::from_utf8(bytes) - .ok() - .map(str::trim) - .filter(|value| validate_id(value).is_ok()) - .map(ToOwned::to_owned) - }); + let head = read_stored_head(&self.root.join("HEAD"))?; Ok(HeadObservation { - head_id, - token: head_bytes.clone().unwrap_or_default(), - head_bytes, + head_id: head.as_ref().and_then(StoredHead::head_id), + token: head + .as_ref() + .map(|head| head.evidence().to_vec()) + .unwrap_or_default(), + head, }) } @@ -446,21 +544,15 @@ impl Storage for FilesystemStorage { validate_id(next)?; let lock = self.lock_file()?; lock.lock_exclusive()?; - let actual = match read_bounded_head_file(&self.root.join("HEAD")) { + let actual = match read_stored_head(&self.root.join("HEAD")) { Ok(actual) => actual, Err(error) => { fs2::FileExt::unlock(&lock)?; return Err(error); } }; - if actual != observed.head_bytes { - let actual_id = actual.as_deref().and_then(|bytes| { - std::str::from_utf8(bytes) - .ok() - .map(str::trim) - .filter(|value| validate_id(value).is_ok()) - .map(ToOwned::to_owned) - }); + if actual != observed.head { + let actual_id = actual.as_ref().and_then(StoredHead::head_id); fs2::FileExt::unlock(&lock)?; return Err(CasConflict { expected: observed.head_id.clone(), @@ -481,32 +573,10 @@ impl Storage for FilesystemStorage { } } -/// Opening a FIFO for reading blocks until a writer appears, so the type is -/// checked on the path before opening and again on the opened handle. -fn read_bounded_head_file(path: &Path) -> Result>> { - match fs::metadata(path) { - Ok(metadata) if !metadata.is_file() => bail!("storage HEAD is not a regular file"), - Ok(_) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), - Err(error) => return Err(error.into()), - } - let file = match File::open(path) { - Ok(file) => file, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), - Err(error) => return Err(error.into()), - }; - if !file.metadata()?.is_file() { - bail!("storage HEAD is not a regular file") - } - let mut bytes = Vec::new(); - file.take(MAX_STORAGE_HEAD_BYTES + 1) - .read_to_end(&mut bytes)?; - if bytes.len() as u64 > MAX_STORAGE_HEAD_BYTES { - bail!(BudgetExhausted( - "recovery outer-head byte budget exhausted".to_owned() - )) - } - Ok(Some(bytes)) +fn read_stored_head(path: &Path) -> Result> { + let bounded = + bounded::read_regular_bounded(path, MAX_HEAD_BYTES).context("read storage HEAD")?; + Ok(StoredHead::from_bounded(bounded)) } fn validate_id(id: &str) -> Result<()> { @@ -534,14 +604,19 @@ fn reader_id(mut reader: impl Read) -> Result { } fn verify_file_id(path: &Path, id: &str) -> Result<()> { - if reader_id(File::open(path)?)? != id { + let Opened::Regular(file, _) = bounded::open_regular(path)? else { + bail!("stored object {id} is not a regular file") + }; + if reader_id(file)? != id { bail!("object id collision for {id}") } Ok(()) } const CARRIER_DEFAULT_BRANCH: &str = "main"; -const CARRIER_CHUNK_SIZE: usize = 32 * 1024 * 1024; +const CARRIER_CHUNK_SIZE: usize = CARRIER_CHUNK_BYTES as usize; +/// Bounds one object's `ls-tree` listing: a few hundred bytes per chunk entry. +const MAX_CARRIER_LISTING_BYTES: u64 = 8 * 1024 * 1024; const CARRIER_ATTRIBUTES: &[u8] = b"* -delta\n"; const CARRIER_CACHE_ENV: &str = "GIT_REMOTE_E2EE_CACHE_DIR"; /// Each cache serves one (remote, branch) pair, so one fixed ref suffices. @@ -596,25 +671,34 @@ impl Read for ChunkReader { .record(0, started.map_or(0, |started| started.elapsed().as_nanos())); return Ok(0); } - self.current = Some(File::open(&self.paths[self.next])?); + let Opened::Regular(file, _) = bounded::open_regular(&self.paths[self.next])? else { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "carrier object chunk is not a regular file", + )); + }; + self.current = Some(file); self.next += 1; } } } -struct GitChunkReader { +/// One `git cat-file --batch` process. Each header is checked before its +/// body is read, so a blob's type and size are known before any allocation. +struct CatFile { child: Child, input: Option, output: BufReader, - chunks: Vec<(String, String)>, - next: usize, - remaining: u64, finished: bool, - trace: trace::Io, } -impl GitChunkReader { - fn new(repo: &Path, chunks: Vec<(String, String)>) -> Result { +enum BatchHeader { + Missing, + Object { id: String, kind: String, size: u64 }, +} + +impl CatFile { + fn spawn(repo: &Path) -> Result { let mut child = carrier_git_command() .arg("-C") .arg(repo) @@ -629,54 +713,52 @@ impl GitChunkReader { child, input: Some(input), output: BufReader::new(output), - chunks, - next: 0, - remaining: 0, finished: false, - trace: trace::Io::new("carrier_object_read"), }) } - fn start_next_chunk(&mut self) -> std::io::Result { - if self.next == self.chunks.len() { - return Ok(false); - } - let expected = &self.chunks[self.next].1; + fn request(&mut self, object: &str) -> std::io::Result { let input = self .input .as_mut() - .expect("git cat-file input is open until all chunks are read"); - writeln!(input, "{expected}")?; + .expect("git cat-file input is open until finished"); + writeln!(input, "{object}")?; input.flush()?; let mut header = String::new(); if self.output.read_line(&mut header)? == 0 { return Err(std::io::Error::new( std::io::ErrorKind::UnexpectedEof, - "git cat-file ended before returning a carrier chunk", + "git cat-file ended before answering", )); } let fields: Vec<_> = header.split_ascii_whitespace().collect(); - if fields.len() != 3 || fields[0] != expected || fields[1] != "blob" { - return Err(std::io::Error::new( + match fields[..] { + [_, "missing"] | [_, "ambiguous"] => Ok(BatchHeader::Missing), + [id, kind, size] => Ok(BatchHeader::Object { + id: id.to_owned(), + kind: kind.to_owned(), + size: size + .parse() + .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?, + }), + _ => Err(std::io::Error::new( std::io::ErrorKind::InvalidData, - "git cat-file returned an invalid carrier chunk header", - )); + "git cat-file returned an invalid header", + )), } - let size = fields[2] - .parse::() - .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?; - let last = self.next + 1 == self.chunks.len(); - if (!last && size != CARRIER_CHUNK_SIZE as u64) - || (last && (size == 0 || size > CARRIER_CHUNK_SIZE as u64)) - { + } + + /// Consumes the newline git writes after each object body. + fn end_body(&mut self) -> std::io::Result<()> { + let mut trailer = [0_u8; 1]; + self.output.read_exact(&mut trailer)?; + if trailer[0] != b'\n' { return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, - "carrier object chunk has invalid size", + "git cat-file returned a malformed object trailer", )); } - self.next += 1; - self.remaining = size; - Ok(true) + Ok(()) } fn finish(&mut self) -> std::io::Result<()> { @@ -690,6 +772,81 @@ impl GitChunkReader { } } +impl Drop for CatFile { + fn drop(&mut self) { + if !self.finished { + self.input.take(); + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } +} + +/// Reads one blob of at most `max_bytes` from a carrier commit. +fn carrier_blob(repo: &Path, object: &str, max_bytes: u64) -> Result { + let mut cat = CatFile::spawn(repo)?; + let size = match cat.request(object)? { + BatchHeader::Missing => return Ok(Bounded::Absent), + BatchHeader::Object { kind, .. } if kind != "blob" => return Ok(Bounded::NotRegular), + BatchHeader::Object { size, .. } if size > max_bytes => return Ok(Bounded::Oversized), + BatchHeader::Object { size, .. } => size, + }; + let mut bytes = vec![0_u8; size as usize]; + cat.output.read_exact(&mut bytes)?; + cat.end_body()?; + cat.finish()?; + Ok(Bounded::Within(bytes)) +} + +/// Streams a carrier object's chunk blobs. A chunk whose header shows the +/// wrong type or size fails with `InvalidData` before its body is read. +struct GitChunkReader { + cat: CatFile, + chunks: Vec<(String, String)>, + next: usize, + remaining: u64, + trace: trace::Io, +} + +impl GitChunkReader { + fn new(repo: &Path, chunks: Vec<(String, String)>) -> Result { + Ok(Self { + cat: CatFile::spawn(repo)?, + chunks, + next: 0, + remaining: 0, + trace: trace::Io::new("carrier_object_read"), + }) + } + + /// The index of the chunk the last read came from. + fn chunk_index(&self) -> usize { + self.next.saturating_sub(1) + } + + fn start_next_chunk(&mut self) -> std::io::Result<()> { + let expected = &self.chunks[self.next].1; + let invalid = |message| std::io::Error::new(std::io::ErrorKind::InvalidData, message); + let size = match self.cat.request(expected)? { + BatchHeader::Object { id, kind, size } if id == *expected && kind == "blob" => size, + _ => { + return Err(invalid( + "git cat-file returned an invalid carrier chunk header", + )); + } + }; + let last = self.next + 1 == self.chunks.len(); + if (!last && size != CARRIER_CHUNK_BYTES) + || (last && (size == 0 || size > CARRIER_CHUNK_BYTES)) + { + return Err(invalid("carrier object chunk has invalid size")); + } + self.next += 1; + self.remaining = size; + Ok(()) + } +} + impl Read for GitChunkReader { fn read(&mut self, output: &mut [u8]) -> std::io::Result { if output.is_empty() { @@ -698,17 +855,17 @@ impl Read for GitChunkReader { let started = self.trace.is_active().then(std::time::Instant::now); if self.remaining == 0 { if self.next == self.chunks.len() { - self.finish()?; + if !self.cat.finished { + self.cat.finish()?; + } self.trace .record(0, started.map_or(0, |started| started.elapsed().as_nanos())); return Ok(0); } - if !self.start_next_chunk()? { - return Ok(0); - } + self.start_next_chunk()?; } let amount = output.len().min(self.remaining as usize); - let read = self.output.read(&mut output[..amount])?; + let read = self.cat.output.read(&mut output[..amount])?; if read == 0 { return Err(std::io::Error::new( std::io::ErrorKind::UnexpectedEof, @@ -717,14 +874,7 @@ impl Read for GitChunkReader { } self.remaining -= read as u64; if self.remaining == 0 { - let mut trailer = [0_u8; 1]; - self.output.read_exact(&mut trailer)?; - if trailer[0] != b'\n' { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "git cat-file returned a malformed chunk trailer", - )); - } + self.cat.end_body()?; } self.trace.record( read, @@ -734,16 +884,6 @@ impl Read for GitChunkReader { } } -impl Drop for GitChunkReader { - fn drop(&mut self) { - if !self.finished { - self.input.take(); - let _ = self.child.kill(); - let _ = self.child.wait(); - } - } -} - struct CarrierObjectStage { staging: PathBuf, root: PathBuf, @@ -934,32 +1074,33 @@ impl GitStorage { /// Reads a root file from the checkout, falling back to the opened /// commit because the checkout only populates the index. - fn local_root_file(&self, name: &str, max_bytes: u64) -> Result>> { + fn local_root_file(&self, name: &str, max_bytes: u64) -> Result { let path = self.root().join(name); - match File::open(&path) { - Ok(file) => { - if file.metadata()?.len() > max_bytes { - bail!("carrier {name} exceeds its size budget") - } - let mut bytes = Vec::new(); - file.take(max_bytes).read_to_end(&mut bytes)?; - Ok(Some(bytes)) - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + match bounded::read_regular_bounded(&path, max_bytes) + .with_context(|| format!("read carrier {name}"))? + { + Bounded::Absent => { let Some(commit) = git_rev_parse(self.root(), "HEAD")? else { - return Ok(None); + return Ok(Bounded::Absent); }; - self.file_at_commit(&commit, name, max_bytes) + carrier_blob(self.root(), &format!("{commit}:{name}"), max_bytes) } - Err(error) => Err(error.into()), + found => Ok(found), } } fn check_format(&self) -> Result<()> { - let format = self.local_root_file("FORMAT", MAX_STORAGE_FORMAT_BYTES)?; - let head_present = self - .local_root_file("HEAD", MAX_STORAGE_HEAD_BYTES)? - .is_some(); + let format = match self.local_root_file("FORMAT", MAX_STORAGE_FORMAT_BYTES)? { + Bounded::Absent => None, + Bounded::Within(bytes) => Some(bytes), + Bounded::Oversized | Bounded::NotRegular => { + bail!("carrier FORMAT is not a small regular file") + } + }; + let head_present = !matches!( + self.local_root_file("HEAD", MAX_HEAD_BYTES)?, + Bounded::Absent + ); check_storage_format(&self.location, format.as_deref(), head_present) } @@ -989,86 +1130,30 @@ impl GitStorage { } fn local_head(&self) -> Result> { - self.check_format()?; - let path = self.root().join("HEAD"); - match fs::read_to_string(path) { - Ok(value) => { - let value = value.trim().to_owned(); - validate_id(&value)?; - Ok(Some(value)) - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - let Some(commit) = git_rev_parse(self.root(), "HEAD")? else { - return Ok(None); - }; - let Some(bytes) = self.head_bytes_at_commit(&commit)? else { - return Ok(None); - }; - let value = String::from_utf8(bytes)?.trim().to_owned(); - validate_id(&value)?; - Ok(Some(value)) - } - Err(error) => Err(error.into()), - } + self.local_stored_head()? + .map(|head| head.parse()) + .transpose() } - fn local_head_bytes(&self) -> Result>> { + fn local_stored_head(&self) -> Result> { self.check_format()?; - let path = self.root().join("HEAD"); - match read_bounded_head_file(&path) { - Ok(Some(bytes)) => Ok(Some(bytes)), - Ok(None) => { - let Some(commit) = git_rev_parse(self.root(), "HEAD")? else { - return Ok(None); - }; - self.head_bytes_at_commit(&commit) - } - Err(error) => Err(error), - } + Ok(StoredHead::from_bounded( + self.local_root_file("HEAD", MAX_HEAD_BYTES)?, + )) } /// The manifest ID a conflicting remote tip names, for diagnostics only: /// a tip whose HEAD is missing or malformed still wins the conflict. fn conflicting_head(&self, commit: Option<&str>) -> Option { - let bytes = self.head_bytes_at_commit(commit?).ok()??; - let value = String::from_utf8(bytes).ok()?.trim().to_owned(); - validate_id(&value).ok()?; - Some(value) + self.stored_head_at_commit(commit?).ok()??.head_id() } - fn head_bytes_at_commit(&self, commit: &str) -> Result>> { - self.file_at_commit(commit, "HEAD", MAX_STORAGE_HEAD_BYTES) - } - - fn file_at_commit(&self, commit: &str, name: &str, max_bytes: u64) -> Result>> { - let object = format!("{commit}:{name}"); - let size = carrier_git_command() - .arg("-C") - .arg(self.root()) - .args(["cat-file", "-s", &object]) - .output()?; - if !size.status.success() { - return Ok(None); - } - let size = String::from_utf8(size.stdout)?.trim().parse::()?; - if size > max_bytes { - if name == "HEAD" { - bail!(BudgetExhausted( - "recovery outer-head byte budget exhausted".to_owned() - )) - } - bail!("carrier {name} exceeds its size budget") - } - let output = carrier_git_command() - .arg("-C") - .arg(self.root()) - .args(["show", &object]) - .output()?; - if output.status.success() { - Ok(Some(output.stdout)) - } else { - Ok(None) - } + fn stored_head_at_commit(&self, commit: &str) -> Result> { + Ok(StoredHead::from_bounded(carrier_blob( + self.root(), + &format!("{commit}:HEAD"), + MAX_HEAD_BYTES, + )?)) } fn commit_history(&self, limit: usize) -> Result> { @@ -1139,9 +1224,11 @@ impl GitStorage { .collect()) } - /// Streams the copy of `id` stored at `commit` into `sink` chunk by chunk. - /// Returns false when the commit holds no copy or the copy is malformed or - /// does not hash to `id`; `sink` errors and process failures propagate. + /// Streams the copy of `id` stored at `commit` into `sink` as + /// `(chunk index, bytes)` pieces. Returns false when the commit holds no + /// copy or the copy is malformed or does not hash to `id`; `sink` sees + /// every byte read, including those of a copy later rejected. `sink` + /// errors and process failures propagate. fn read_copy_at( &self, commit: &str, @@ -1155,22 +1242,18 @@ impl GitStorage { Err(error) if error.is::() => return Err(error), Err(_) => return Ok(false), }; + let mut reader = GitChunkReader::new(self.root(), chunks)?; let mut hasher = Sha256::new(); - for (index, (path, _)) in chunks.iter().enumerate() { - let output = carrier_git_command() - .arg("-C") - .arg(self.root()) - .args(["show", &format!("{commit}:{path}")]) - .output()?; - let length = output.stdout.len(); - if !output.status.success() - || (index + 1 < chunks.len() && length != CARRIER_CHUNK_SIZE) - || (index + 1 == chunks.len() && (length == 0 || length > CARRIER_CHUNK_SIZE)) - { - return Ok(false); - } - hasher.update(&output.stdout); - sink(index, &output.stdout)?; + let mut buffer = vec![0_u8; 64 * 1024]; + loop { + let read = match reader.read(&mut buffer) { + Ok(0) => break, + Ok(read) => read, + Err(error) if error.kind() == std::io::ErrorKind::InvalidData => return Ok(false), + Err(error) => return Err(error.into()), + }; + sink(reader.chunk_index(), &buffer[..read])?; + hasher.update(&buffer[..read]); } Ok(hex::encode(hasher.finalize()) == id) } @@ -1305,15 +1388,18 @@ impl Storage for GitStorage { OsRng.fill_bytes(&mut random); let staging = self.root().join(format!(".stage-{}", hex::encode(random))); fs::create_dir(&staging)?; - Ok(ValidatingObjectStage::wrap(CarrierObjectStage { - staging, - root: self.root().to_path_buf(), + Ok(ValidatingObjectStage::wrap( + CarrierObjectStage { + staging, + root: self.root().to_path_buf(), + kind, + current: None, + chunk_index: 0, + chunk_len: 0, + trace: trace::Io::new("carrier_stage_write"), + }, kind, - current: None, - chunk_index: 0, - chunk_len: 0, - trace: trace::Io::new("carrier_stage_write"), - })) + )) } fn open_object(&self, kind: ObjectKind, id: &str) -> Result> { @@ -1405,23 +1491,16 @@ impl Storage for GitStorage { .state .lock() .map_err(|_| anyhow::anyhow!("carrier state lock poisoned"))?; - let head_bytes = self.local_head_bytes()?; - let head_id = head_bytes.as_deref().and_then(|bytes| { - std::str::from_utf8(bytes) - .ok() - .map(str::trim) - .filter(|value| validate_id(value).is_ok()) - .map(ToOwned::to_owned) - }); + let head = self.local_stored_head()?; Ok(HeadObservation { - head_id, + head_id: head.as_ref().and_then(StoredHead::head_id), token: state .base_commit .as_deref() .unwrap_or_default() .as_bytes() .to_vec(), - head_bytes, + head, }) } @@ -1444,21 +1523,15 @@ impl Storage for GitStorage { } .into()); } - let actual_bytes = self.local_head_bytes()?; - if actual_bytes != observed.head_bytes { + let actual = self.local_stored_head()?; + if actual != observed.head { return Err(CasConflict { expected: observed.head_id.clone(), - actual: actual_bytes.as_deref().and_then(|bytes| { - std::str::from_utf8(bytes) - .ok() - .map(str::trim) - .filter(|value| validate_id(value).is_ok()) - .map(ToOwned::to_owned) - }), + actual: actual.as_ref().and_then(StoredHead::head_id), } .into()); } - if actual_bytes.is_none() { + if actual.is_none() { self.refuse_foreign_branch(&expected_tip)?; } let new_commit = @@ -1483,15 +1556,8 @@ impl Storage for GitStorage { let mut total_bytes = 0_u64; for commit_id in &commits { let head_id = self - .head_bytes_at_commit(commit_id)? - .as_deref() - .and_then(|bytes| { - std::str::from_utf8(bytes) - .ok() - .map(str::trim) - .filter(|value| validate_id(value).is_ok()) - .map(ToOwned::to_owned) - }); + .stored_head_at_commit(commit_id)? + .and_then(|head| head.head_id()); if let Some(id) = &head_id && !history.manifests.contains_key(id) { @@ -1524,25 +1590,50 @@ impl Storage for GitStorage { Ok(Some(history)) } - fn restore_historical_object(&self, kind: ObjectKind, id: &str) -> Result { + fn restore_historical_object( + &self, + kind: ObjectKind, + id: &str, + max_bytes: u64, + ) -> Result { let commits = self.commit_history(MAX_RECOVERY_OUTER_COMMITS)?; if commits.len() > MAX_RECOVERY_OUTER_COMMITS { bail!(BudgetExhausted( "carrier recovery outer-commit budget exhausted".to_owned() )) } + // Every byte read counts, including corrupt copies that are skipped. + let mut total_bytes = 0_u64; for copy in self.copy_commits(&commits, kind, id)? { let staging = tempfile::Builder::new() .prefix("git-remote-e2ee-restore-") .tempdir_in(self.root())?; let object_dir = staging.path().join("object"); fs::create_dir(&object_dir)?; - let valid = self.read_copy_at(©, kind, id, |index, chunk| { - let mut file = File::create(object_dir.join(format!("{index:08}")))?; - file.write_all(chunk)?; - file.sync_all()?; + let mut current: Option<(usize, File)> = None; + let valid = self.read_copy_at(©, kind, id, |index, piece| { + total_bytes = total_bytes + .checked_add(piece.len() as u64) + .context("carrier restore byte count overflow")?; + if total_bytes > max_bytes { + bail!(BudgetExhausted( + "carrier recovery restore byte budget exhausted".to_owned() + )) + } + if current.as_ref().map(|(open, _)| *open) != Some(index) { + if let Some((_, file)) = current.take() { + file.sync_all()?; + } + let file = File::create(object_dir.join(format!("{index:08}")))?; + current = Some((index, file)); + } + let (_, file) = current.as_mut().expect("chunk file opened above"); + file.write_all(piece)?; Ok(()) })?; + if let Some((_, file)) = current.take() { + file.sync_all()?; + } if !valid { continue; } @@ -1658,21 +1749,34 @@ fn carrier_chunk_entries( ) -> Result>> { validate_id(id)?; let directory = format!("{}/{}/{id}", kind.directory(), &id[..2]); - let output = carrier_git_command() + let mut child = carrier_git_command() .arg("-C") .arg(repo) .args(["ls-tree", "-r", "-z", commit, "--", &directory]) - .output()?; - ensure_git_success(&output, "git ls-tree carrier object")?; - if output.stdout.is_empty() { - return Ok(None); - } - if output.stdout.len() > 8 * 1024 * 1024 { + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn()?; + let mut listing = Vec::new(); + child + .stdout + .take() + .context("open git ls-tree stdout")? + .take(MAX_CARRIER_LISTING_BYTES + 1) + .read_to_end(&mut listing)?; + if listing.len() as u64 > MAX_CARRIER_LISTING_BYTES { + let _ = child.kill(); + let _ = child.wait(); bail!("carrier object listing exceeds its size budget") } + let status = child.wait()?; + if !status.success() { + bail!("git ls-tree carrier object failed: {status}") + } + if listing.is_empty() { + return Ok(None); + } let mut chunks = Vec::new(); - for entry in output - .stdout + for entry in listing .split(|byte| *byte == 0) .filter(|entry| !entry.is_empty()) { @@ -1694,7 +1798,7 @@ fn carrier_chunk_entries( chunks.push((path, blob_id.to_owned())); } chunks.sort_by(|left, right| left.0.cmp(&right.0)); - if chunks.is_empty() || chunks.len() > 4096 { + if chunks.is_empty() || chunks.len() as u64 > MAX_CHUNKS_PER_OBJECT { bail!("carrier object has an invalid chunk count") } for (index, (path, _)) in chunks.iter().enumerate() { @@ -2257,17 +2361,23 @@ mod tests { fn object_stage_validates_all_written_bytes_before_publishing() { let data = b"partial writes must all contribute to the staged object hash"; let published = Arc::new(AtomicBool::new(false)); - let mut stage = ValidatingObjectStage::wrap(PartialWriteSink { - published: Arc::clone(&published), - }); + let mut stage = ValidatingObjectStage::wrap( + PartialWriteSink { + published: Arc::clone(&published), + }, + ObjectKind::Pack, + ); stage.write_all(data).unwrap(); stage.finish(&object_id(data)).unwrap(); assert!(published.load(Ordering::Relaxed)); let published = Arc::new(AtomicBool::new(false)); - let mut stage = ValidatingObjectStage::wrap(PartialWriteSink { - published: Arc::clone(&published), - }); + let mut stage = ValidatingObjectStage::wrap( + PartialWriteSink { + published: Arc::clone(&published), + }, + ObjectKind::Pack, + ); stage.write_all(data).unwrap(); assert!(stage.finish(&object_id(b"different bytes")).is_err()); assert!(!published.load(Ordering::Relaxed)); @@ -2341,7 +2451,7 @@ mod tests { fn buffered_object_reads_have_a_hard_limit() { let directory = tempfile::tempdir().unwrap(); let storage = FilesystemStorage::new(directory.path()); - let data = vec![1_u8; MAX_BUFFERED_OBJECT_SIZE as usize + 1]; + let data = vec![1_u8; MAX_MANIFEST_BYTES as usize + 1]; let id = object_id(&data); storage .put_object_if_absent(ObjectKind::Pack, &id, &data) @@ -2395,52 +2505,168 @@ mod tests { } #[cfg(unix)] - #[test] - fn filesystem_head_reads_refuse_a_fifo() { - let (directory, storage) = initialized_storage(); - let observed = storage.observe_head().unwrap(); - let head = directory.path().join("HEAD"); - fs::remove_file(&head).unwrap(); - let status = Command::new("mkfifo").arg(&head).status().unwrap(); + fn make_fifo(path: &Path) { + let _ = fs::remove_file(path); + let status = Command::new("mkfifo").arg(path).status().unwrap(); assert!(status.success()); + } + + fn assert_invalid_head(storage: &FilesystemStorage, expected: StoredHead) { + let error = storage.read_head().unwrap_err(); + assert!( + format!("{error:#}").contains("storage HEAD is not a valid head"), + "{error:#}" + ); + let observed = storage.observe_head().unwrap(); + assert_eq!(observed.head_id, None); + assert_eq!(observed.head, Some(expected)); + } + + /// Replaces HEAD with an entry no reader may accept, then checks that + /// reads report an invalid head, a stale observation loses the CAS, and a + /// fresh observation of the invalid head can replace it. + fn assert_head_entry_is_an_invalid_observation( + replace: impl FnOnce(&Path), + expected: StoredHead, + ) { + let (directory, storage) = initialized_storage(); + let stale = storage.observe_head().unwrap(); + replace(&directory.path().join("HEAD")); let storage = Arc::new(storage); let reader = Arc::clone(&storage); - let result = finishes_within(move || reader.read_head().is_err()); - assert_eq!(result, Some(true), "read_head must refuse a FIFO HEAD"); + let shown = expected.clone(); + let observed = finishes_within(move || { + assert_invalid_head(&reader, shown); + reader.observe_head().unwrap() + }) + .expect("reading an invalid HEAD must not block"); + + let writer = Arc::clone(&storage); + let stale_error = finishes_within(move || { + writer + .compare_and_swap_observed_head(&stale, &object_id(b"second")) + .unwrap_err() + }) + .expect("CAS over an invalid HEAD must not block"); + assert!(stale_error.downcast_ref::().is_some()); + + let next = object_id(b"recovered"); + storage + .compare_and_swap_observed_head(&observed, &next) + .unwrap(); + assert_eq!(storage.read_head().unwrap(), Some(next)); + drop(directory); + } + + #[cfg(unix)] + #[test] + fn a_fifo_head_is_an_invalid_observation_not_a_hang() { + assert_head_entry_is_an_invalid_observation(make_fifo, StoredHead::NotRegular); + } + + #[cfg(unix)] + #[test] + fn a_head_symlink_to_a_fifo_is_an_invalid_observation_not_a_hang() { + assert_head_entry_is_an_invalid_observation( + |head| { + let fifo = head.with_file_name("elsewhere"); + make_fifo(&fifo); + fs::remove_file(head).unwrap(); + std::os::unix::fs::symlink(&fifo, head).unwrap(); + }, + StoredHead::NotRegular, + ); + } + + #[cfg(unix)] + #[test] + fn a_head_symlink_to_a_regular_file_is_not_followed() { + assert_head_entry_is_an_invalid_observation( + |head| { + let target = head.with_file_name("elsewhere"); + fs::write(&target, format!("{}\n", object_id(b"first"))).unwrap(); + fs::remove_file(head).unwrap(); + std::os::unix::fs::symlink(&target, head).unwrap(); + }, + StoredHead::NotRegular, + ); + } + + #[test] + fn an_oversized_head_is_an_invalid_observation() { + assert_head_entry_is_an_invalid_observation( + |head| fs::write(head, vec![b'a'; MAX_HEAD_BYTES as usize + 1]).unwrap(), + StoredHead::Oversized, + ); + } + + #[cfg(unix)] + #[test] + fn a_fifo_format_fails_instead_of_hanging() { + let (directory, storage) = initialized_storage(); + make_fifo(&directory.path().join("FORMAT")); + let result = finishes_within(move || format!("{:#}", storage.read_head().unwrap_err())); + assert!( + result + .expect("reading a FIFO FORMAT must not block") + .contains("not a regular file") + ); + } + + #[cfg(unix)] + #[test] + fn a_fifo_pack_fails_instead_of_hanging() { + let (directory, storage) = initialized_storage(); + let data = b"pack bytes"; + let id = object_id(data); + let path = storage.object_path(ObjectKind::Pack, &id).unwrap(); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + make_fifo(&path); + let storage = Arc::new(storage); let reader = Arc::clone(&storage); - let result = finishes_within(move || reader.observe_head().is_err()); - assert_eq!(result, Some(true), "observe_head must refuse a FIFO HEAD"); + let opened = finishes_within(move || { + reader + .open_object(ObjectKind::Pack, &id) + .err() + .map(|error| error.downcast_ref::().is_some()) + }); + assert_eq!(opened, Some(Some(true)), "opening a FIFO pack must fail"); let writer = Arc::clone(&storage); - let result = finishes_within(move || { + let written = finishes_within(move || { writer - .compare_and_swap_observed_head(&observed, &object_id(b"second")) + .put_object_if_absent(ObjectKind::Pack, &object_id(data), data) .is_err() }); - assert_eq!(result, Some(true), "observed CAS must refuse a FIFO HEAD"); + assert_eq!(written, Some(true), "verifying a FIFO pack must fail"); drop(directory); } #[test] - fn filesystem_head_reads_reject_oversized_heads() { + fn writers_refuse_a_manifest_readers_would_reject_before_publication() { let (directory, storage) = initialized_storage(); - let observed = storage.observe_head().unwrap(); - let oversized = vec![b'a'; MAX_STORAGE_HEAD_BYTES as usize + 1]; - fs::write(directory.path().join("HEAD"), &oversized).unwrap(); - for error in [ - storage.read_head().unwrap_err(), - storage.observe_head().unwrap_err(), - storage - .compare_and_swap_observed_head(&observed, &object_id(b"second")) - .unwrap_err(), - ] { - assert!( - error.downcast_ref::().is_some(), - "{error:#}" - ); - } + let head = storage.read_head().unwrap(); + let data = vec![7_u8; MAX_MANIFEST_BYTES as usize + 1]; + let id = object_id(&data); + let error = storage + .put_object_if_absent(ObjectKind::Manifest, &id, &data) + .unwrap_err(); + assert!( + format!("{error:#}").contains("manifest object exceeds"), + "{error:#}" + ); + assert!( + !storage + .object_path(ObjectKind::Manifest, &id) + .unwrap() + .exists() + ); + assert_eq!(storage.read_head().unwrap(), head); + let manifests = directory.path().join("manifests"); + assert_eq!(fs::read_dir(manifests).unwrap().count(), 0); + drop(directory); } #[test] diff --git a/tests/carrier_chunk_bounds.rs b/tests/carrier_chunk_bounds.rs new file mode 100644 index 0000000..8ffaebb --- /dev/null +++ b/tests/carrier_chunk_bounds.rs @@ -0,0 +1,121 @@ +//! One test per binary: it measures this process's peak memory, which other +//! tests running concurrently would disturb. +#![cfg(unix)] + +mod support; + +use std::fs::{self, File}; +use std::io::Write; +use std::path::Path; +use std::process::Command; + +use git_remote_e2ee::crypto::KeyFile; +use git_remote_e2ee::repository::EncryptedRepository; +use git_remote_e2ee::storage::{CARRIER_CHUNK_BYTES, GitStorage, ObjectKind, Storage}; + +fn git(repo: &Path, args: &[&str]) { + let output = Command::new("git") + .arg("-C") + .arg(repo) + .args(args) + .output() + .unwrap(); + assert!( + output.status.success(), + "git {args:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); +} + +fn peak_resident_bytes() -> u64 { + let mut usage = std::mem::MaybeUninit::::zeroed(); + let result = unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) }; + assert_eq!(result, 0); + let peak = unsafe { usage.assume_init() }.ru_maxrss as u64; + if cfg!(target_os = "macos") { + peak + } else { + peak * 1024 + } +} + +#[test] +fn oversized_carrier_chunks_are_rejected_without_being_read() { + let temporary = tempfile::tempdir().unwrap(); + // Only test in this binary, so the process-global override is stable. + unsafe { + std::env::set_var( + "GIT_REMOTE_E2EE_CACHE_DIR", + temporary.path().join("carrier-cache"), + ) + }; + let carrier = temporary.path().join("carrier.git"); + git( + temporary.path(), + &["init", "--bare", "-q", carrier.to_str().unwrap()], + ); + let repository = EncryptedRepository::new( + GitStorage::open(carrier.to_str().unwrap()).unwrap(), + KeyFile::generate(), + support::state_dir(), + ); + repository.initialize("owner").unwrap(); + let (head_id, _) = repository.current_manifest().unwrap(); + let manifest = GitStorage::open(carrier.to_str().unwrap()) + .unwrap() + .get_object(ObjectKind::Manifest, &head_id) + .unwrap(); + + let workdir = temporary.path().join("edit"); + git( + temporary.path(), + &[ + "clone", + "-q", + "-b", + "main", + carrier.to_str().unwrap(), + workdir.to_str().unwrap(), + ], + ); + let chunk = workdir + .join("manifests") + .join(&head_id[..2]) + .join(&head_id) + .join("00000000"); + let mut file = File::create(&chunk).unwrap(); + let block = vec![b'x'; 1024 * 1024]; + for _ in 0..CARRIER_CHUNK_BYTES / block.len() as u64 { + file.write_all(&block).unwrap(); + } + file.write_all(b"!").unwrap(); + drop(file); + git( + &workdir, + &[ + "-c", + "user.name=t", + "-c", + "user.email=t@t", + "commit", + "-qam", + "oversize", + ], + ); + git(&workdir, &["push", "-q", "origin", "main"]); + fs::remove_dir_all(&workdir).unwrap(); + + let storage = GitStorage::open(carrier.to_str().unwrap()).unwrap(); + let before = peak_resident_bytes(); + let history = storage + .recovery_history(2048, 64 * 1024 * 1024) + .unwrap() + .unwrap(); + let grown = peak_resident_bytes().saturating_sub(before); + + assert_eq!(history.manifests[&head_id], manifest); + assert!( + grown < CARRIER_CHUNK_BYTES / 2, + "reading history grew peak memory by {grown} bytes" + ); +} diff --git a/tests/carrier_git.rs b/tests/carrier_git.rs index 75e80a3..9850024 100644 --- a/tests/carrier_git.rs +++ b/tests/carrier_git.rs @@ -2396,7 +2396,7 @@ fn carrier_history_reads_skip_corrupt_copies_for_an_older_valid_one() { assert!( storage - .restore_historical_object(ObjectKind::Pack, &pack_id) + .restore_historical_object(ObjectKind::Pack, &pack_id, u64::MAX) .unwrap() ); assert_eq!( @@ -2405,11 +2405,78 @@ fn carrier_history_reads_skip_corrupt_copies_for_an_older_valid_one() { ); assert!( !storage - .restore_historical_object(ObjectKind::Pack, &"0".repeat(64)) + .restore_historical_object(ObjectKind::Pack, &"0".repeat(64), u64::MAX) .unwrap() ); } +#[test] +fn carrier_historical_restore_charges_rejected_copies_against_its_byte_budget() { + let temporary = tempfile::tempdir().unwrap(); + let _cache = isolated_cache(&temporary); + let carrier = temporary.path().join("carrier.git"); + let source = temporary.path().join("source"); + init_source(&source, "content"); + + let owner = KeyFile::generate(); + init_carrier(temporary.path(), &carrier, &owner); + let repository = EncryptedRepository::new( + GitStorage::open(carrier.to_str().unwrap()).unwrap(), + owner.clone(), + support::state_dir(), + ); + repository + .push( + &source, + "refs/heads/main", + RefChange::Update { + source: "refs/heads/main", + force: false, + }, + ) + .unwrap(); + let (_, head) = repository.current_manifest().unwrap(); + let pack_id = head.new_packs[0].id.clone(); + let pack_bytes = GitStorage::open(carrier.to_str().unwrap()) + .unwrap() + .get_object(ObjectKind::Pack, &pack_id) + .unwrap(); + let corrupt = b"corrupt copy"; + edit_carrier( + &carrier, + &temporary.path().join("carrier-edit"), + |workdir| { + let chunk = workdir + .join("packs") + .join(&pack_id[..2]) + .join(&pack_id) + .join("00000000"); + fs::write(chunk, corrupt).unwrap(); + }, + ); + + let needed = (corrupt.len() + pack_bytes.len()) as u64; + let storage = GitStorage::open(carrier.to_str().unwrap()).unwrap(); + let error = storage + .restore_historical_object(ObjectKind::Pack, &pack_id, needed - 1) + .unwrap_err(); + assert!( + error + .downcast_ref::() + .is_some(), + "{error:#}" + ); + assert!( + storage + .restore_historical_object(ObjectKind::Pack, &pack_id, needed) + .unwrap() + ); + assert_eq!( + storage.get_object(ObjectKind::Pack, &pack_id).unwrap(), + pack_bytes + ); +} + /// The writer joins after genesis, so the heads that predate it are sealed /// to readers it never was. Recovery from a malformed HEAD must still work. fn recover_junk_head_after_later_writer_push(recover_as_owner: bool) { diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs index 1c9549f..000664c 100644 --- a/tests/end_to_end.rs +++ b/tests/end_to_end.rs @@ -1626,6 +1626,39 @@ fn recovery_discovery_budget_exhaustion_fails_closed() { assert_eq!(machine_floor(&repo)["head_id"], floor); } +#[test] +fn an_oversized_storage_head_is_invalid_and_recoverable_from_the_floor() { + let temporary = tempfile::tempdir().unwrap(); + let repo = temporary.path().join("repo"); + let storage_path = temporary.path().join("storage"); + initialize_git(&repo); + let key = KeyFile::generate(); + let storage = FilesystemStorage::new(&storage_path); + let repository = EncryptedRepository::new(storage.clone(), key, support::state_dir()); + repository.initialize("owner").unwrap(); + repository.fetch_into(&repo, "encrypted").unwrap(); + let floor = storage.read_head().unwrap().unwrap(); + fs::write( + storage_path.join("HEAD"), + vec![b'a'; git_remote_e2ee::storage::MAX_HEAD_BYTES as usize + 1], + ) + .unwrap(); + + let error = repository.fetch_into(&repo, "encrypted").unwrap_err(); + assert!( + format!("{error:#}").contains("storage HEAD is not a valid head"), + "{error:#}" + ); + + let recovered = publish_recovery(&repository, &repo).unwrap(); + assert_eq!(recovered.classification, RecoveryClass::Invalid); + let published = recovered.published_manifest.unwrap(); + let (current_id, current) = repository.current_manifest().unwrap(); + assert_eq!(current_id, published); + assert_eq!(current.previous.as_deref(), Some(floor.as_str())); + assert_eq!(machine_floor(&repo)["head_id"], current_id); +} + #[test] fn writer_can_add_branch_without_having_other_remote_branch_objects() { let temporary = tempfile::tempdir().unwrap();