crates/gunnar-lfs/src/verify.rs
//! **The one place an uploaded object is checked**, whatever wire carried it.
//!
//! A backend is never asked to verify, so no backend can forget to, and there is
//! one implementation rather than one per backend — and now rather than one per
//! *transport*. [`crate::session`] wraps a `put-object`'s payload in this;
//! `gunnar_http::lfs` wraps a `PUT`'s request body in the same one. An object
//! that reached gunnar over HTTP is admitted on exactly the terms an object that
//! reached it over SSH is (one implementation).
//!
//! Three things have to hold before [`PendingWrite::commit`] is called, and all
//! three are here:
//!
//! 1. **the byte count matches what was declared** — a short object is
//! discarded, not stored;
//! 2. **an overrun is stopped as it happens** rather than at the end, so a
//! client claiming 1 KiB and sending 40 GiB is cut off after the first chunk
//! past the claim;
//! 3. **the content hashes to the oid it was sent under** — SHA-256 of the raw
//! bytes, which is what an LFS oid *is*.
//!
//! Failing any of them drops the [`PendingWrite`] without committing, which is
//! what discards the partial object: the `fs` backend never renames it into
//! place, so there is no window in which wrong content is visible under a valid
//! oid.
use crate::error::{Error, Result};
use crate::oid::{Oid, Sha256Stream};
use crate::store::PendingWrite;
/// A sink that hashes and counts as the bytes pass, and commits only if both
/// match what the client declared.
pub struct VerifyingSink<'p> {
inner: Option<Box<dyn PendingWrite + 'p>>,
hasher: Option<Sha256Stream>,
expected: Oid,
declared: u64,
received: u64,
}
impl<'p> VerifyingSink<'p> {
/// Wrap a pending write for the object `expected`, which the client says is
/// `declared` bytes long.
pub fn new(inner: Box<dyn PendingWrite + 'p>, expected: Oid, declared: u64) -> Self {
VerifyingSink {
inner: Some(inner),
hasher: Some(Sha256Stream::new()),
expected,
declared,
received: 0,
}
}
/// Take one chunk. Refuses **at the first chunk past the claim** rather
/// than at the end.
pub fn accept(&mut self, data: &[u8]) -> Result<()> {
let next = self.received.saturating_add(data.len() as u64);
if next > self.declared {
return Err(Error::Corrupt(format!(
"object {} declared {} bytes and sent at least {next}",
self.expected, self.declared
)));
}
self.hasher
.as_mut()
.expect("a sink is used before it is committed")
.update(data);
// **Not `?`.** `?` on an `io::Error` here becomes `Error::Io`, which is
// fatal and has no `final_report`, so a full disk mid-upload ended the
// session with no status line on the wire at all — `git lfs push` onto a
// full volume saw a dropped SSH channel, not a 507 and not even a 500.
// `store_io` classifies it, and both of its answers are survivable, so
// `put_object` drains the rest and tells the client which it was.
if let Err(err) = self
.inner
.as_mut()
.expect("a sink is used before it is committed")
.write_all(data)
{
return Err(Error::store_io(format!("write {}", self.expected), err));
}
self.received = next;
Ok(())
}
/// How many bytes have been accepted so far.
pub fn received(&self) -> u64 {
self.received
}
/// Publish the object, if and only if the count and the digest are right.
pub fn commit(mut self) -> Result<()> {
let hasher = self.hasher.take().expect("committed once");
let inner = self.inner.take().expect("committed once");
if self.received != self.declared {
// `inner` is dropped here, which is what discards the partial
// object. Nothing is renamed into place.
return Err(Error::Corrupt(format!(
"object {} declared {} bytes and sent {}",
self.expected, self.declared, self.received
)));
}
let actual = hasher.finish();
if actual != self.expected {
return Err(Error::Corrupt(format!(
"object was sent as {} but hashes to {actual}",
self.expected
)));
}
inner.commit()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::store::{LfsStore, MemLfsStore};
#[test]
fn the_verifier_stops_an_overrun_at_the_first_chunk_past_the_claim() {
let store = MemLfsStore::new();
let oid = Oid::parse(&"e".repeat(64)).unwrap();
let mut sink = VerifyingSink::new(store.begin_write(&oid, 4).unwrap(), oid, 4);
assert!(sink.accept(b"abcd").is_ok());
let err = sink.accept(b"e").expect_err("one byte past the claim");
assert_eq!(err.status(), 422);
}
#[test]
fn a_short_object_is_not_committed() {
let store = MemLfsStore::new();
let oid = Oid::parse(&"f".repeat(64)).unwrap();
let mut sink = VerifyingSink::new(store.begin_write(&oid, 10).unwrap(), oid, 10);
sink.accept(b"abc").unwrap();
assert_eq!(sink.received(), 3);
assert_eq!(sink.commit().unwrap_err().status(), 422);
assert!(store.is_empty());
}
#[test]
fn bytes_that_hash_to_something_else_are_not_committed() {
let store = MemLfsStore::new();
let lie = Oid::parse(&"a".repeat(64)).unwrap();
let mut sink = VerifyingSink::new(store.begin_write(&lie, 3).unwrap(), lie, 3);
sink.accept(b"abc").unwrap();
let err = sink.commit().expect_err("abc does not hash to aaaa…");
assert_eq!(err.status(), 422);
assert!(
err.to_string().contains("ba7816bf"),
"the diagnostic must name the digest that was actually computed: {err}"
);
assert!(store.is_empty());
// The identity-value half: the truthful case must commit, or the three
// tests above would all pass against a sink that rejected everything.
let truth =
Oid::parse("ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad").unwrap();
let mut sink = VerifyingSink::new(store.begin_write(&truth, 3).unwrap(), truth, 3);
sink.accept(b"abc").unwrap();
sink.commit().unwrap();
assert_eq!(store.get(&truth).unwrap(), b"abc");
}
}