gunnar-lfs 1.6.0

//! **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");
    }
}