splitstream: Add content_type to header

This is an extra check to avoid confusion.

Author: alexlarsson

crates/cfsctl/src/main.rs

@@ -180,7 +180,7 @@ async fn main() -> Result<()> {
             }
         }
         Command::Cat { name } => {
-            repo.merge_splitstream(&name, None, &mut std::io::stdout())?;
+            repo.merge_splitstream(&name, None, None, &mut std::io::stdout())?;
         }
         Command::ImportImage { reference } => {
             let image_id = repo.import_image(&reference, &mut std::io::stdin())?;

crates/composefs-http/src/lib.rs

@@ -61,7 +61,7 @@ impl<ObjectID: FsVerityHashValue> Downloader<ObjectID> {
     }
 
     fn open_splitstream(&self, id: &ObjectID) -> Result<SplitStreamReader<File, ObjectID>> {
-        SplitStreamReader::new(File::from(self.repo.open_object(id)?))
+        SplitStreamReader::new(File::from(self.repo.open_object(id)?), None)
     }
 
     fn read_object(&self, id: &ObjectID) -> Result<Vec<u8>> {

crates/composefs-oci/src/image.rs

@@ -9,6 +9,7 @@ use composefs::{
     tree::{Directory, FileSystem, Inode, Leaf},
 };
 
+use crate::skopeo::{OCI_CONFIG_CONTENT_TYPE, TAR_LAYER_CONTENT_TYPE};
 use crate::tar::{TarEntry, TarItem};
 
 pub fn process_entry<ObjectID: FsVerityHashValue>(
@@ -74,14 +75,14 @@ pub fn create_filesystem<ObjectID: FsVerityHashValue>(
 ) -> Result<FileSystem<ObjectID>> {
     let mut filesystem = FileSystem::default();
 
-    let mut config_stream = repo.open_stream(config_name, config_verity)?;
+    let mut config_stream = repo.open_stream(config_name, config_verity, Some(OCI_CONFIG_CONTENT_TYPE))?;
     let config = ImageConfiguration::from_reader(&mut config_stream)?;
 
     for diff_id in config.rootfs().diff_ids() {
         let layer_sha256 = super::sha256_from_digest(diff_id)?;
         let layer_verity = config_stream.lookup(&layer_sha256)?;
 
-        let mut layer_stream = repo.open_stream(&hex::encode(layer_sha256), Some(layer_verity))?;
+        let mut layer_stream = repo.open_stream(&hex::encode(layer_sha256), Some(layer_verity), Some(TAR_LAYER_CONTENT_TYPE))?;
         while let Some(entry) = crate::tar::get_entry(&mut layer_stream)? {
             process_entry(&mut filesystem, entry)?;
         }

crates/composefs-oci/src/lib.rs

@@ -16,6 +16,7 @@ use composefs::{
 };
 
 use crate::tar::get_entry;
+use crate::skopeo::{OCI_CONFIG_CONTENT_TYPE, TAR_LAYER_CONTENT_TYPE};
 
 type ContentAndVerity<ObjectID> = (Sha256Digest, ObjectID);
 
@@ -39,14 +40,14 @@ pub fn import_layer<ObjectID: FsVerityHashValue>(
     name: Option<&str>,
     tar_stream: &mut impl Read,
 ) -> Result<ObjectID> {
-    repo.ensure_stream(sha256, |writer| tar::split(tar_stream, writer), name)
+    repo.ensure_stream(sha256, TAR_LAYER_CONTENT_TYPE, |writer| tar::split(tar_stream, writer), name)
 }
 
 pub fn ls_layer<ObjectID: FsVerityHashValue>(
     repo: &Repository<ObjectID>,
     name: &str,
 ) -> Result<()> {
-    let mut split_stream = repo.open_stream(name, None)?;
+    let mut split_stream = repo.open_stream(name, None, Some(TAR_LAYER_CONTENT_TYPE))?;
 
     while let Some(entry) = get_entry(&mut split_stream)? {
         println!("{entry}");
@@ -81,7 +82,7 @@ pub fn open_config<ObjectID: FsVerityHashValue>(
                 .with_context(|| format!("Object {name} is unknown to us"))?
         }
     };
-    let mut stream = repo.open_stream(name, Some(id))?;
+    let mut stream = repo.open_stream(name, Some(id), Some(OCI_CONFIG_CONTENT_TYPE))?;
     let config = ImageConfiguration::from_reader(&mut stream)?;
     Ok((config, stream.get_mappings()))
 }
@@ -104,7 +105,7 @@ pub fn open_config_shallow<ObjectID: FsVerityHashValue>(
             // we need to manually check the content digest
             let expected_hash = parse_sha256(name)
                 .context("Containers must be referred to by sha256 if verity is missing")?;
-            let mut stream = repo.open_stream(name, None)?;
+            let mut stream = repo.open_stream(name, None, Some(OCI_CONFIG_CONTENT_TYPE))?;
             let mut raw_config = vec![];
             stream.read_to_end(&mut raw_config)?;
             ensure!(hash(&raw_config) == expected_hash, "Data integrity issue");
@@ -121,7 +122,7 @@ pub fn write_config<ObjectID: FsVerityHashValue>(
     let json = config.to_string()?;
     let json_bytes = json.as_bytes();
     let sha256 = hash(json_bytes);
-    let mut stream = repo.create_stream(Some(sha256));
+    let mut stream = repo.create_stream(OCI_CONFIG_CONTENT_TYPE, Some(sha256));
     stream.add_sha256_mappings(refs);
     stream.write_inline(json_bytes);
     let id = repo.write_stream(stream, None)?;

crates/composefs-oci/src/skopeo.rs

@@ -16,6 +16,9 @@ use composefs::{
 
 use crate::{sha256_from_descriptor, sha256_from_digest, tar::split_async, ContentAndVerity};
 
+pub const TAR_LAYER_CONTENT_TYPE : u64 = 0x2a037edfcae1ffea;
+pub const OCI_CONFIG_CONTENT_TYPE : u64 = 0x44218c839727a80b;
+
 struct ImageOp<ObjectID: FsVerityHashValue> {
     repo: Arc<Repository<ObjectID>>,
     proxy: ImageProxy,
@@ -78,7 +81,7 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             self.progress
                 .println(format!("Fetching layer {}", hex::encode(layer_sha256)))?;
 
-            let mut splitstream = self.repo.create_stream(Some(layer_sha256));
+            let mut splitstream = self.repo.create_stream(TAR_LAYER_CONTENT_TYPE, Some(layer_sha256));
             match descriptor.media_type() {
                 MediaType::ImageLayer => {
                     split_async(progress, &mut splitstream).await?;
@@ -157,7 +160,7 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
 
             let mut splitstream = self
                 .repo
-                .create_stream(Some(config_sha256));
+                .create_stream(OCI_CONFIG_CONTENT_TYPE, Some(config_sha256));
 
             // Collect the results.
             for (layer_sha256, future) in entries {

crates/composefs/src/repository.rs

@@ -184,9 +184,10 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     /// store the result.
     pub fn create_stream(
         self: &Arc<Self>,
+        content_type: u64,
         sha256: Option<Sha256Digest>,
     ) -> SplitStreamWriter<ObjectID> {
-        SplitStreamWriter::new(self, sha256)
+        SplitStreamWriter::new(self, content_type, sha256)
     }
 
     fn format_object_path(id: &ObjectID) -> String {
@@ -223,7 +224,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
             Ok(stream) => {
                 let measured_verity: ObjectID = measure_verity(&stream)?;
                 let mut context = Sha256::new();
-                let mut split_stream = SplitStreamReader::new(File::from(stream))?;
+                let mut split_stream = SplitStreamReader::new(File::from(stream), None)?;
 
                 // check the verity of all linked streams
                 for (body, verity) in split_stream.iter_mappings() {
@@ -296,6 +297,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     pub fn ensure_stream(
         self: &Arc<Self>,
         sha256: &Sha256Digest,
+        content_type: u64,
         callback: impl FnOnce(&mut SplitStreamWriter<ObjectID>) -> Result<()>,
         reference: Option<&str>,
     ) -> Result<ObjectID> {
@@ -304,7 +306,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         let object_id = match self.has_stream(sha256)? {
             Some(id) => id,
             None => {
-                let mut writer = self.create_stream(Some(*sha256));
+                let mut writer = self.create_stream(content_type, Some(*sha256));
                 callback(&mut writer)?;
                 let object_id = writer.done()?;
 
@@ -326,6 +328,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         &self,
         name: &str,
         verity: Option<&ObjectID>,
+        expected_content_type: Option<u64>,
     ) -> Result<SplitStreamReader<File, ObjectID>> {
         let filename = format!("streams/{name}");
 
@@ -337,7 +340,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
                 .with_context(|| format!("Opening ref 'streams/{name}'"))?
         });
 
-        SplitStreamReader::new(file)
+        SplitStreamReader::new(file, expected_content_type)
     }
 
     pub fn open_object(&self, id: &ObjectID) -> Result<OwnedFd> {
@@ -348,9 +351,10 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         &self,
         name: &str,
         verity: Option<&ObjectID>,
+        expected_content_type: Option<u64>,
         stream: &mut impl Write,
     ) -> Result<()> {
-        let mut split_stream = self.open_stream(name, verity)?;
+        let mut split_stream = self.open_stream(name, verity, expected_content_type)?;
         split_stream.cat(stream, |id| -> Result<Vec<u8>> {
             let mut data = vec![];
             File::from(self.open_object(id)?).read_to_end(&mut data)?;
@@ -553,7 +557,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
             println!("{object:?} lives as a stream");
             objects.insert(object.clone());
 
-            let mut split_stream = self.open_stream(&object.to_hex(), None)?;
+            let mut split_stream = self.open_stream(&object.to_hex(), None, None)?;
             split_stream.get_object_refs(|id| {
                 println!("   with {id:?}");
                 objects.insert(id.clone());

crates/composefs/src/splitstream.rs

@@ -26,6 +26,7 @@ pub const SPLITSTREAM_MAGIC : [u8; 7] = [b'S', b'p', b'l', b't', b'S', b't', b'r
 pub struct SplitstreamHeader {
     pub magic: [u8; 7], // Contains SPLITSTREAM_MAGIC
     pub algorithm: u8,
+    pub content_type: u64, // User can put whatever magic identifier they want there
     pub total_size: u64, // total size of inline chunks and external chunks
     pub n_refs: u64,
     pub n_mappings: u64,
@@ -92,6 +93,7 @@ pub struct SplitStreamWriter<ObjectID: FsVerityHashValue> {
     inline_content: Vec<u8>,
     total_size: u64,
     writer: Encoder<'static, Vec<u8>>,
+    pub content_type: u64,
     pub sha256: Option<(Sha256, Sha256Digest)>,
 }
 
@@ -109,13 +111,15 @@ impl<ObjectID: FsVerityHashValue> std::fmt::Debug for SplitStreamWriter<ObjectID
 impl<ObjectID: FsVerityHashValue> SplitStreamWriter<ObjectID> {
     pub fn new(
         repo: &Arc<Repository<ObjectID>>,
+        content_type: u64,
         sha256: Option<Sha256Digest>,
     ) -> Self {
         // SAFETY: we surely can't get an error writing the header to a Vec<u8>
         let writer = Encoder::new(vec![], 0).unwrap();
 
         Self {
             repo: Arc::clone(repo),
+            content_type: content_type,
             inline_content: vec![],
             refs: vec![],
             total_size: 0,
@@ -218,6 +222,7 @@ impl<ObjectID: FsVerityHashValue> SplitStreamWriter<ObjectID> {
         let header = SplitstreamHeader {
             magic: SPLITSTREAM_MAGIC,
             algorithm:  ObjectID::ALGORITHM,
+            content_type: self.content_type,
             total_size: u64::to_le(self.total_size),
             n_refs: u64::to_le(self.refs.len() as u64),
             n_mappings: u64::to_le(self.mappings.map.len() as u64),
@@ -252,6 +257,7 @@ pub enum SplitStreamData<ObjectID: FsVerityHashValue> {
 pub struct SplitStreamReader<R: Read, ObjectID: FsVerityHashValue> {
     decoder: Decoder<'static, BufReader<R>>,
     inline_bytes: usize,
+    pub content_type: u64,
     pub total_size: u64,
     pub refs: Vec<ObjectID>,
     mappings: Vec<MappingEntry>,
@@ -293,17 +299,24 @@ enum ChunkType<ObjectID: FsVerityHashValue> {
 }
 
 impl<R: Read, ObjectID: FsVerityHashValue> SplitStreamReader<R, ObjectID> {
-    pub fn new(mut reader: R) -> Result<Self> {
+    pub fn new(mut reader: R, expected_content_type: Option<u64>) -> Result<Self> {
 
         let header = SplitstreamHeader::read_from_io(&mut reader)
             .map_err(|e| Error::msg(format!("Error reading splitstream header: {:?}", e)))?;
 
         if header.magic != SPLITSTREAM_MAGIC {
-            bail!("Invalida splitstream header magic value");
+            bail!("Invalid splitstream header magic value");
         }
 
         if header.algorithm != ObjectID::ALGORITHM {
-            bail!("Invalida splitstream algorithm type");
+            bail!("Invalid splitstream algorithm type");
+        }
+
+        let content_type = u64::from_le(header.content_type);
+        if let Some(expected) = expected_content_type {
+            if content_type != expected {
+                bail!("Invalid splitstream content type");
+            }
         }
 
         let total_size = u64::from_le(header.total_size);
@@ -333,6 +346,7 @@ impl<R: Read, ObjectID: FsVerityHashValue> SplitStreamReader<R, ObjectID> {
         Ok(Self {
             decoder,
             inline_bytes: 0,
+            content_type,
             total_size,
             refs,
             mappings,