Re-constituting splitstreams now works

src/repository.rs

@@ -3,13 +3,17 @@ use std::io::{
     Read,
     Write,
 };
+use std::fs::File;
 use std::os::fd::OwnedFd;
 use std::path::{
     Path,
     PathBuf,
 };
 
-use anyhow::Result;
+use anyhow::{
+    Result,
+    bail,
+};
 use rustix::fs::{
     Access,
     AtFlags,
@@ -29,13 +33,15 @@ use rustix::fs::{
 
 use crate::{
     fsverity::{
+        FsVerityHashValue,
         Sha256HashValue,
         digest::FsVerityHasher,
         ioctl::{
             fs_ioc_enable_verity,
             fs_ioc_measure_verity,
         },
     },
+    splitstream::splitstream_merge,
     tar,
     util::proc_self_fd,
 };
@@ -119,7 +125,49 @@ impl Repository {
         Ok(digest)
     }
 
+    pub fn open_with_verity(&self, filename: &str, expected_verity: Sha256HashValue) -> Result<OwnedFd> {
+        let fd = openat(&self.repository, filename, OFlags::RDONLY, Mode::empty())?;
+        let measured_verity: Sha256HashValue = fs_ioc_measure_verity(&fd)?;
+        if measured_verity != expected_verity {
+            bail!("bad verity!")
+        } else {
+            Ok(fd)
+        }
+    }
+
+    /// category is like "streams" or "images"
+    /// name is like "refs/1000/user/xyz" (with '/') or a sha256 hex hash value (without '/')
+    fn open_in_category(&self, category: &str, name: &str) -> Result<OwnedFd> {
+        let filename = format!("{}/{}", category, name);
+
+        if name.contains("/") {
+            // no fsverity checking on this path
+            Ok(openat(&self.repository, filename, OFlags::RDONLY, Mode::empty())?)
+        } else {
+            // this must surely be a hash value, and we want to verify it
+            let mut hash = Sha256HashValue::EMPTY;
+            hex::decode_to_slice(name, &mut hash)?;
+            self.open_with_verity(&filename, hash)
+        }
+    }
+
+    fn open_object(&self, id: Sha256HashValue) -> Result<OwnedFd> {
+        self.open_with_verity(&format!("objects/{:02x}/{}", id[0], hex::encode(&id[1..])), id)
+    }
+
     pub fn merge_splitstream<W: Write>(&self, name: &str, stream: &mut W) -> Result<()> {
+        let file = File::from(self.open_in_category("streams", name)?);
+        let mut split_stream = zstd::stream::read::Decoder::new(file)?;
+        splitstream_merge(
+            &mut split_stream,
+            stream,
+            |id: Sha256HashValue| -> Result<Vec<u8>> {
+                let mut data = vec![];
+                File::from(self.open_object(id)?).read_to_end(&mut data)?;
+                Ok(data)
+            }
+        )?;
+
         Ok(())
     }
 
@@ -131,7 +179,8 @@ impl Repository {
             &mut split_stream,
             |data: &[u8]| -> Result<Sha256HashValue> {
                 self.ensure_object(data)
-        })?;
+            }
+        )?;
 
         let object_id = self.ensure_object(&split_stream.finish()?)?;
         self.link_ref(name, "streams", object_id)
@@ -142,7 +191,7 @@ impl Repository {
     ) -> Result<()> {
         let object_path = format!("objects/{:02x}/{}", object_id[0], hex::encode(&object_id[1..]));
         let category_path = format!("{}/{}", category, hex::encode(&object_id));
-        let ref_path = format!("refs/{}", name);
+        let ref_path = format!("{}/refs/{}", category, name);
 
         self.symlink(&ref_path, &category_path)?;
         self.symlink(&category_path, &object_path)?;

src/splitstream.rs

@@ -27,11 +27,20 @@
  * That's it, really.  There's no header.  The file is over when there's no more blocks.
  */
 
-use std::io::Write;
+use std::io::{
+    Read,
+    Write,
+};
 
 use anyhow::Result;
 
-use crate::fsverity::Sha256HashValue;
+use crate::{
+    fsverity::{
+        FsVerityHashValue,
+        Sha256HashValue,
+    },
+    util::read_exactish,
+};
 
 // utility class to help write splitstreams
 pub struct SplitStreamWriter<'w, W: Write> {
@@ -81,4 +90,31 @@ impl<'w, W: Write> SplitStreamWriter<'w, W> {
     }
 }
 
+fn read_u64_le<R: Read>(reader: &mut R) -> Result<Option<u64>> {
+    let mut buf = [0u8; 8];
+    if read_exactish(reader, &mut buf)? {
+        Ok(Some(u64::from_le_bytes(buf)))
+    } else {
+        Ok(None)
+    }
+}
+
 // TODO: reader side...
+pub fn splitstream_merge<R: Read, W: Write, F: FnMut(Sha256HashValue) -> Result<Vec<u8>>>(
+    split_stream: &mut R, result: &mut W, mut load_data: F,
+) -> Result<()> {
+    while let Some(size) = read_u64_le(split_stream)? {
+        if size == 0 {
+            let mut hash = Sha256HashValue::EMPTY;
+            split_stream.read_exact(&mut hash)?;
+            let data = load_data(hash)?;
+            result.write_all(&data)?;
+        } else {
+            let mut data = vec![0u8; size as usize]; // TODO: bzzt bzzt
+            split_stream.read_exact(&mut data)?;
+            result.write_all(&data)?;
+        }
+    }
+
+    Ok(())
+}

src/tar.rs

@@ -2,8 +2,11 @@ use std::io::{Read, Write};
 
 use anyhow::Result;
 
-use crate::fsverity::Sha256HashValue;
-use crate::splitstream::SplitStreamWriter;
+use crate::{
+    fsverity::Sha256HashValue,
+    splitstream::SplitStreamWriter,
+    util::read_exactish,
+};
 
 struct TarHeader {
     data: [u8; 512],
@@ -13,28 +16,12 @@ impl TarHeader {
     // we can't use Read::read_exact() because we need to be able to detect EOF
     fn read<R: Read>(reader: &mut R) -> Result<Option<TarHeader>> {
         let mut header = TarHeader { data: [0u8; 512] };
-        let mut todo: &mut [u8] = &mut header.data;
-
-        while !todo.is_empty() {
-            match reader.read(todo) {
-                Ok(0) => match todo.len() {
-                    512 => return Ok(None),  // clean EOF
-                    _ => Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof))?
-                },
-                Ok(n) => {
-                    todo = &mut todo[n..];
-                }
-                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
-                    continue;
-                }
-                Err(e) => {
-                    Err(e)?;
-                }
-            }
+        if read_exactish(reader, &mut header.data)? {
+            Ok(Some(header))
+        } else {
+            Ok(None)
         }
-
-        Ok(Some(header))
-    }
+     }
 
     fn get_size(&self) -> usize {
         let size_field = &self.data[124..124 + 12];

src/util.rs

@@ -1,8 +1,39 @@
-use std::os::fd::{
-    AsFd,
-    AsRawFd,
+use std::{
+    io::Read,
+    os::fd::{
+        AsFd,
+        AsRawFd,
+    },
 };
 
+use anyhow::Result;
+
 pub fn proc_self_fd<A: AsFd>(fd: &A) -> String {
     format!("/proc/self/fd/{}", fd.as_fd().as_raw_fd())
 }
+
+// we can't use Read::read_exact() because we need to be able to detect EOF
+pub fn read_exactish<R: Read>(reader: &mut R, buf: &mut [u8]) -> Result<bool> {
+    let buflen = buf.len();
+    let mut todo: &mut [u8] = buf;
+
+    while !todo.is_empty() {
+        match reader.read(todo) {
+            Ok(0) => match todo.len() {
+                s if s == buflen => return Ok(false),  // clean EOF
+                _ => Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof))?
+            },
+            Ok(n) => {
+                todo = &mut todo[n..];
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
+                continue;
+            }
+            Err(e) => {
+                Err(e)?;
+            }
+        }
+    }
+
+    Ok(true)
+}