[pbs-devel] [PATCH pxar 3/3] encoder: improve consistency check for payload reference offsets

[Christian Ebner]

The pxar encoder for split metadata and payload archives cannot keep
exactly track of the payload stream and payload reference offset to
be in sync since it must allow for injecting payloads steming from
reused chunks in PBS. Therefore, currently it is only checked that the
offset to encode in the payload reference is greater than the current
payload writer position (the chunk data is always injected after
advancing the metadata archive).

This check is however not good enough to protect against encoding an
archive with not strict monotonically increasing payload reference
offset, which cannot be restored by the sequential decoder.

Guard against this by keeping track of the previously encoded payload
offset and check for the new one to be always greater. Move the
pre-existing check together with the stricter check into a common
helper and us it also for checking the offsets when creating files
being encoded with split pxar archives, not just when encoding a
reused file in the metadata archive as pxar payload reference.

This will avoid the encoding of corrupt archives as reported in the
community forum [0].

[0] https://forum.proxmox.com/threads/170211/

Signed-off-by: Christian Ebner <c.ebner at proxmox.com>
---
 src/encoder/mod.rs | 66 +++++++++++++++++++++++++++++++++++-----------
 1 file changed, 50 insertions(+), 16 deletions(-)

diff --git a/src/encoder/mod.rs b/src/encoder/mod.rs
index 4e7daef..131cd69 100644
--- a/src/encoder/mod.rs
+++ b/src/encoder/mod.rs
@@ -243,6 +243,9 @@ struct EncoderState {
     /// Track the bytes written to the payload writer
     payload_write_position: u64,
 
+    /// Previously generated payload offset
+    previous_payload_offset: Option<PayloadOffset>,
+
     /// Mark the encoder state as correctly finished, ready to be dropped
     finished: bool,
 }
@@ -258,6 +261,46 @@ impl EncoderState {
         self.payload_write_position
     }
 
+    #[inline]
+    fn previous_payload_offset(&self) -> Option<PayloadOffset> {
+        self.previous_payload_offset
+    }
+
+    /// Generate a PayloadRef from the given payload offset and file size.
+    ///
+    /// Checks if the provided payload offset is greater than the current payload writeer position
+    /// and larger than the previously checked payload offset. Both are required for the sequential
+    /// decoder to be able to restore contents.
+    fn payload_ref_from(
+        &mut self,
+        payload_offset: PayloadOffset,
+        size: u64,
+    ) -> io::Result<PayloadRef> {
+        let payload_position = self.payload_position();
+        if payload_offset.raw() < payload_position {
+            io_bail!(
+                "offset smaller than current position: {} < {payload_position}",
+                payload_offset.raw()
+            );
+        }
+
+        if let Some(previous_payload_offset) = self.previous_payload_offset() {
+            if payload_offset <= previous_payload_offset {
+                io_bail!(
+                    "unexpected payload offset {} not larger than previous offset {}",
+                    payload_offset.raw(),
+                    previous_payload_offset.raw(),
+                );
+            }
+        }
+        self.previous_payload_offset = Some(payload_offset);
+
+        Ok(PayloadRef {
+            offset: payload_offset.raw(),
+            size,
+        })
+    }
+
     fn merge_error(&mut self, error: Option<EncodeError>) {
         // one error is enough:
         if self.encode_error.is_none() {
@@ -454,17 +497,13 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
         if let Some(payload_output) = output.payload_mut() {
             // payload references must point to the position prior to the payload header,
             // separating payload entries in the payload stream
-            let payload_position = state.payload_position();
+            let payload_position = PayloadOffset(state.payload_position());
+            let payload_ref = state.payload_ref_from(payload_position, file_size)?;
 
             let header = format::Header::with_content_size(format::PXAR_PAYLOAD, file_size);
             header.check_header_size()?;
             seq_write_struct(payload_output, header, &mut state.payload_write_position).await?;
 
-            let payload_ref = PayloadRef {
-                offset: payload_position,
-                size: file_size,
-            };
-
             seq_write_pxar_entry(
                 output.archive_mut(),
                 format::PXAR_PAYLOAD_REF,
@@ -566,16 +605,9 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
             io_bail!("unable to add payload reference");
         }
 
-        let offset = payload_offset.raw();
-        let payload_position = self.state()?.payload_position();
-        if offset < payload_position {
-            io_bail!("offset smaller than current position: {offset} < {payload_position}");
-        }
-
-        let payload_ref = PayloadRef {
-            offset,
-            size: file_size,
-        };
+        let payload_ref = self
+            .state_mut()?
+            .payload_ref_from(payload_offset, file_size)?;
         let this_offset: LinkOffset = self
             .add_file_entry(
                 Some(metadata),
@@ -749,6 +781,7 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
         // the child will write to OUR state now:
         let write_position = state.position();
         let payload_write_position = state.payload_position();
+        let previous_payload_offset = state.previous_payload_offset();
 
         self.state.push(EncoderState {
             items: Vec::new(),
@@ -759,6 +792,7 @@ impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
             file_hash,
             write_position,
             payload_write_position,
+            previous_payload_offset,
             finished: false,
         });
 
-- 
2.47.2