Bound HEAD reads, refuse generation overflow, and tighten hex and temp-file handling

src/crypto.rs

@@ -686,7 +686,11 @@ fn device_id_for_public_keys(signing: &[u8], wrapping: &[u8]) -> String {
 }
 
 fn validate_root(root: &str) -> Result<()> {
-    if root.len() != 64 || !root.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+    if root.len() != 64
+        || !root
+            .bytes()
+            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
+    {
         bail!("repository root must be a 32-byte lowercase hexadecimal value")
     }
     Ok(())
@@ -957,4 +961,10 @@ mod tests {
         let mut ciphertext = Vec::new();
         assert!(seal_pack_stream(&random_key(), &[][..], &mut ciphertext, b"test-pack").is_err());
     }
+
+    #[test]
+    fn repository_root_must_be_lowercase_hex() {
+        validate_root(&"ab".repeat(32)).unwrap();
+        assert!(validate_root(&"AB".repeat(32)).is_err());
+    }
 }

src/manifest.rs

@@ -105,7 +105,7 @@ impl Manifest {
         if self.repository_root != previous.repository_root {
             bail!("manifest repository root changed")
         }
-        if self.generation != previous.generation + 1 {
+        if previous.generation.checked_add(1) != Some(self.generation) {
             bail!("manifest generation is not consecutive")
         }
         if self.previous.as_deref() != Some(previous_id) {
@@ -134,7 +134,7 @@ impl Manifest {
             }
             ManifestAuthorization::PolicyTransition => {
                 validate_introduced_policy(self, selected_policy, true)?;
-                if self.policy_generation != previous.policy_generation + 1
+                if previous.policy_generation.checked_add(1) != Some(self.policy_generation)
                     || selected_policy.body.previous.as_deref() != Some(previous.policy_id.as_str())
                 {
                     bail!("policy transition did not select the direct policy child")
@@ -568,7 +568,7 @@ pub fn unwrap_predecessor_key(
         .predecessor_key_wrap
         .as_deref()
         .context("successor manifest is missing predecessor key link")?;
-    if parent_header.generation + 1 != manifest.generation
+    if parent_header.generation.checked_add(1) != Some(manifest.generation)
         || parent_header.repository_root != manifest.repository_root
     {
         bail!("predecessor header does not match manifest generation")
@@ -1341,4 +1341,30 @@ mod tests {
             .is_err()
         );
     }
+
+    #[test]
+    fn generation_overflow_is_a_validation_error() {
+        let owner = KeyFile::generate();
+        let (policy, policy_bytes) = PolicyState::genesis(&owner, "owner").unwrap();
+        let previous = manifest_for(&policy, u64::MAX, ManifestAuthorization::Writer);
+        let next = manifest_for(&policy, 0, ManifestAuthorization::Writer);
+        assert!(
+            next.validate_successor(&"11".repeat(32), &previous, &policy)
+                .is_err()
+        );
+
+        let parent = Manifest::genesis(owner.repository_root.clone(), &policy, policy_bytes);
+        let parent_bytes = seal_manifest(
+            &owner,
+            &policy,
+            &random_key(),
+            ManifestAuthorization::PolicyTransition,
+            parent,
+        )
+        .unwrap();
+        let mut parent_header = peek_manifest_header(&parent_bytes).unwrap();
+        parent_header.generation = u64::MAX;
+        let child = manifest_for(&policy, 1, ManifestAuthorization::Writer);
+        assert!(unwrap_predecessor_key(&random_key(), &child, &parent_header).is_err());
+    }
 }

src/policy.rs

@@ -125,7 +125,11 @@ impl PolicyState {
         let body = PolicyBody {
             format_version: POLICY_FORMAT_VERSION,
             repository_root: parent.body.repository_root.clone(),
-            generation: parent.body.generation + 1,
+            generation: parent
+                .body
+                .generation
+                .checked_add(1)
+                .context("policy generation overflow")?,
             previous: Some(parent.id.clone()),
             admin_threshold: 1,
             devices,
@@ -201,7 +205,7 @@ impl PolicyState {
         if self.body.repository_root != parent.body.repository_root {
             bail!("policy repository root changed")
         }
-        if self.body.generation != parent.body.generation + 1 {
+        if parent.body.generation.checked_add(1) != Some(self.body.generation) {
             bail!("policy generation is not consecutive")
         }
         if self.body.previous.as_deref() != Some(parent.id.as_str()) {
@@ -614,4 +618,15 @@ mod tests {
             .validate_successor(&genesis)
             .unwrap();
     }
+
+    #[test]
+    fn generation_overflow_is_a_validation_error() {
+        let owner = KeyFile::generate();
+        let (mut parent, _) = PolicyState::genesis(&owner, "owner").unwrap();
+        let (child, _) =
+            PolicyState::successor(&parent, parent.body.devices.clone(), &owner).unwrap();
+        parent.body.generation = u64::MAX;
+        assert!(PolicyState::successor(&parent, parent.body.devices.clone(), &owner).is_err());
+        assert!(child.validate_successor(&parent).is_err());
+    }
 }

src/repository.rs

@@ -281,7 +281,14 @@ impl<S: Storage> EncryptedRepository<S> {
             .iter_mut()
             .find(|device| device.public.device_id == device_id && device.active())
             .context("device is not active in the current policy")?;
-        target.revoked_at = Some(current.policy.body.generation + 1);
+        target.revoked_at = Some(
+            current
+                .policy
+                .body
+                .generation
+                .checked_add(1)
+                .context("policy generation overflow")?,
+        );
         if !devices
             .iter()
             .any(|device| device.active() && device.roles.reader)
@@ -462,7 +469,11 @@ impl<S: Storage> EncryptedRepository<S> {
         }
 
         let next_key = random_key();
-        let generation = current.manifest.generation + 1;
+        let generation = current
+            .manifest
+            .generation
+            .checked_add(1)
+            .context("manifest generation overflow")?;
         let mut new_packs = Vec::new();
         let mut total_pack_count = current.manifest.total_pack_count;
         if let Some(next_object) = next_object {
@@ -1673,7 +1684,11 @@ fn transition_manifest(
     policy_bytes: Vec<u8>,
     generation_key: &[u8; 32],
 ) -> Result<Manifest> {
-    let generation = current.manifest.generation + 1;
+    let generation = current
+        .manifest
+        .generation
+        .checked_add(1)
+        .context("manifest generation overflow")?;
     Ok(Manifest {
         format_version: current.manifest.format_version,
         repository_root: current.manifest.repository_root.clone(),

src/storage.rs

@@ -376,16 +376,15 @@ impl Storage for FilesystemStorage {
 
     fn read_head(&self) -> Result<Option<String>> {
         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 => Ok(None),
-            Err(error) => Err(error.into()),
-        }
+        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))
     }
 
     fn compare_and_swap_head(&self, expected: Option<&str>, next: &str) -> Result<()> {
@@ -434,12 +433,11 @@ impl Storage for FilesystemStorage {
         validate_id(next)?;
         let lock = self.lock_file()?;
         lock.lock_exclusive()?;
-        let actual = match fs::read(self.root.join("HEAD")) {
-            Ok(bytes) => Some(bytes),
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        let actual = match read_bounded_head_file(&self.root.join("HEAD")) {
+            Ok(actual) => actual,
             Err(error) => {
                 fs2::FileExt::unlock(&lock)?;
-                return Err(error.into());
+                return Err(error);
             }
         };
         if actual != observed.head_bytes {
@@ -478,10 +476,16 @@ fn write_filesystem_head_locked(root: &Path, next: &str) -> Result<()> {
         .write(true)
         .create_new(true)
         .open(&temporary)?;
-    writeln!(file, "{next}")?;
-    file.sync_all()?;
-    crate::persist::durability_checkpoint(crate::persist::STAGE_AFTER_FILE_FLUSH);
-    fs::rename(&temporary, root.join("HEAD"))?;
+    let published = (|| -> std::io::Result<()> {
+        writeln!(file, "{next}")?;
+        file.sync_all()?;
+        crate::persist::durability_checkpoint(crate::persist::STAGE_AFTER_FILE_FLUSH);
+        fs::rename(&temporary, root.join("HEAD"))
+    })();
+    if let Err(error) = published {
+        let _ = fs::remove_file(&temporary);
+        return Err(error.into());
+    }
     crate::persist::durability_checkpoint(crate::persist::STAGE_AFTER_NAME_PUBLISH);
     crate::persist::sync_directory(root)
         .with_context(|| format!("sync directory {}", root.display()))?;
@@ -496,32 +500,54 @@ fn write_file_durable(root: &Path, name: &str, bytes: &[u8]) -> Result<()> {
         .write(true)
         .create_new(true)
         .open(&temporary)?;
-    file.write_all(bytes)?;
-    file.sync_all()?;
-    fs::rename(&temporary, root.join(name))?;
+    let published = (|| -> std::io::Result<()> {
+        file.write_all(bytes)?;
+        file.sync_all()?;
+        fs::rename(&temporary, root.join(name))
+    })();
+    if let Err(error) = published {
+        let _ = fs::remove_file(&temporary);
+        return Err(error.into());
+    }
     crate::persist::sync_directory(root)
         .with_context(|| format!("sync directory {}", root.display()))?;
     Ok(())
 }
 
+/// 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<Option<Vec<u8>>> {
-    let mut file = match File::open(path) {
+    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()?.len() > MAX_STORAGE_HEAD_BYTES {
+    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()
         ))
     }
-    let mut bytes = Vec::new();
-    file.read_to_end(&mut bytes)?;
     Ok(Some(bytes))
 }
 
 fn validate_id(id: &str) -> Result<()> {
-    if id.len() != 64 || !id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+    if id.len() != 64
+        || !id
+            .bytes()
+            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
+    {
         bail!("invalid opaque object id")
     }
     Ok(())
@@ -2367,4 +2393,92 @@ mod tests {
         fs::create_dir(directory.path().join("00000000")).unwrap();
         assert!(validated_chunk_paths(directory.path()).is_err());
     }
+
+    fn initialized_storage() -> (tempfile::TempDir, FilesystemStorage) {
+        let directory = tempfile::tempdir().unwrap();
+        let storage = FilesystemStorage::new(directory.path());
+        storage
+            .compare_and_swap_head(None, &object_id(b"first"))
+            .unwrap();
+        (directory, storage)
+    }
+
+    fn finishes_within<T: Send + 'static>(work: impl FnOnce() -> T + Send + 'static) -> Option<T> {
+        let (sender, receiver) = std::sync::mpsc::channel();
+        thread::spawn(move || {
+            let _ = sender.send(work());
+        });
+        receiver
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .ok()
+    }
+
+    #[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();
+        assert!(status.success());
+        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 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 writer = Arc::clone(&storage);
+        let result = finishes_within(move || {
+            writer
+                .compare_and_swap_observed_head(&observed, &object_id(b"second"))
+                .is_err()
+        });
+        assert_eq!(result, Some(true), "observed CAS must refuse a FIFO HEAD");
+        drop(directory);
+    }
+
+    #[test]
+    fn filesystem_head_reads_reject_oversized_heads() {
+        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::<BudgetExhausted>().is_some(),
+                "{error:#}"
+            );
+        }
+    }
+
+    #[test]
+    fn object_ids_must_be_lowercase_hex() {
+        validate_id(&"ab".repeat(32)).unwrap();
+        assert!(validate_id(&"AB".repeat(32)).is_err());
+    }
+
+    #[test]
+    fn failed_head_write_removes_its_temporary() {
+        let directory = tempfile::tempdir().unwrap();
+        fs::create_dir(directory.path().join("HEAD")).unwrap();
+        fs::write(directory.path().join("HEAD").join("occupant"), b"x").unwrap();
+        assert!(write_filesystem_head_locked(directory.path(), &object_id(b"next")).is_err());
+        let leftovers: Vec<_> = fs::read_dir(directory.path())
+            .unwrap()
+            .map(|entry| entry.unwrap().file_name())
+            .filter(|name| name.to_string_lossy().starts_with(".HEAD-"))
+            .collect();
+        assert!(leftovers.is_empty(), "leftover temporaries: {leftovers:?}");
+    }
 }