[PATCH proxmox-backup v4 4/5] fix #6990: server: drop verify state on non-decrypt pull job

[Jakob Klocker <j.klocker@proxmox.com>]

On a non-decrypt pull the source manifest is written to the target
as-is, so the target inherits the source's verify_state flag instead of
being verified independently on its own storage. Because a snapshot
carrying a verify_state is skipped by verify jobs, the target's copy can
never be checked.

The decrypt path already drops verify_state; do the same on the
non-decrypt path when the snapshot is newly pulled or re-synced due to
corruption. Also factor the manifest blob encoding and writing into a
helper, shared by both paths.

Link: https://bugzilla.proxmox.com/show_bug.cgi?id=6990
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@proxmox.com>
Signed-off-by: Jakob Klocker <j.klocker@proxmox.com>
---
 pbs-datastore/src/manifest.rs | 34 +++++++++++++++++++++--
 src/server/pull.rs            | 51 ++++++++++++++++++++++-------------
 2 files changed, 64 insertions(+), 21 deletions(-)

diff --git a/pbs-datastore/src/manifest.rs b/pbs-datastore/src/manifest.rs
index 8c7a5f8b6..ea8642f6e 100644
--- a/pbs-datastore/src/manifest.rs
+++ b/pbs-datastore/src/manifest.rs
@@ -1,7 +1,9 @@
 use std::collections::HashSet;
+use std::io::Write;
+use std::os::fd::AsRawFd;
+use std::path::Path;
 
-use anyhow::{Error, bail, format_err};
-
+use anyhow::{Context, Error, bail, format_err};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 
@@ -301,6 +303,34 @@ impl BackupManifest {
         Ok(Some(Deserialize::deserialize(value)?))
     }
 
+    /// Encode the manifest and write it atomically to `target_path`.
+    ///
+    /// The encoded blob is first written to `tmp_path` and fsync'd, then
+    /// atomically renamed onto `target_path`. Writing via a temp file and
+    /// rename ensures concurrent readers never observe a partially written,
+    /// corrupt manifest.
+    ///
+    /// Returns the raw blob data.
+    pub fn write_to_path(&self, tmp_path: &Path, target_path: &Path) -> Result<Vec<u8>, Error> {
+        let raw_data = self.to_data_blob(None)?.raw_data().to_vec();
+
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(tmp_path)
+            .with_context(|| format!("failed to open manifest {tmp_path:?}"))?;
+
+        file.write_all(&raw_data)?;
+        file.flush()?;
+        nix::unistd::fsync(file.as_raw_fd())?;
+
+        std::fs::rename(tmp_path, target_path)
+            .with_context(|| format!("atomic rename manifest {target_path:?} failed"))?;
+
+        Ok(raw_data)
+    }
+
     /// Returns the set of all files which might be encountered in
     /// the snapshot represented by given manifest.
     pub fn expected_files(&self) -> HashSet<BackupArchiveName> {
diff --git a/src/server/pull.rs b/src/server/pull.rs
index ef7f714b7..1665bee3c 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -3,7 +3,6 @@
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet};
 use std::io::Seek;
-use std::os::fd::AsRawFd;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
@@ -16,8 +15,6 @@ use super::sync::{
 use crate::backup::{check_ns_modification_privs, check_ns_privs};
 use crate::server::sync::SharedGroupProgress;
 use anyhow::{Context, Error, bail, format_err};
-use tokio::fs::OpenOptions;
-use tokio::io::AsyncWriteExt;
 
 use pbs_api_types::{
     ArchiveType, Authid, BackupDir, BackupGroup, BackupNamespace, CLIENT_LOG_BLOB_NAME, CryptMode,
@@ -744,7 +741,8 @@ async fn pull_snapshot<'a>(
     };
 
     let mut manifest_data = tmp_manifest_blob.raw_data().to_vec();
-    let manifest = BackupManifest::try_from(tmp_manifest_blob).with_context(|| prefix.clone())?;
+    let mut manifest =
+        BackupManifest::try_from(tmp_manifest_blob).with_context(|| prefix.clone())?;
 
     if ignore_not_verified_or_encrypted(
         &manifest,
@@ -852,7 +850,11 @@ async fn pull_snapshot<'a>(
         sync_stats.add(stats);
     }
 
+    let target_verify_state = existing_target_manifest
+        .as_ref()
+        .and_then(|m| m.unprotected.get("verify_state").cloned());
     let files_to_keep = manifest.expected_files();
+
     if let Some(new_manifest) = new_manifest {
         let mut new_manifest = Arc::try_unwrap(new_manifest)
             .map_err(|_arc| {
@@ -879,24 +881,35 @@ async fn pull_snapshot<'a>(
             new_manifest.set_sync_source_signature(expected.bytes())?;
         }
 
-        // keep signature
-        let manifest_blob = new_manifest.to_data_blob(None)?;
-        // update contents to be uploaded to backend
-        manifest_data = manifest_blob.raw_data().to_vec();
+        manifest_data = tokio::task::spawn_blocking(move || {
+            new_manifest.write_to_path(&tmp_manifest_name, &manifest_name)
+        })
+        .await??;
+    } else if manifest.unprotected.get("verify_state") != target_verify_state.as_ref() {
+        // the manifest is copied as-is on the non-decrypted path: never inherit the
+        // source's verify state, but keep one the target obtained on its own
+        let Some(unprotected) = manifest.unprotected.as_object_mut() else {
+            bail!("{prefix}: unexpected manifest without 'unprotected' section");
+        };
+        match &target_verify_state {
+            Some(state) => {
+                unprotected.insert("verify_state".to_string(), state.clone());
+            }
+            None => {
+                unprotected.remove("verify_state");
+            }
+        }
 
-        let mut tmp_manifest_file = OpenOptions::new()
-            .write(true)
-            .truncate(true) // clear pre-existing manifest content
-            .open(&tmp_manifest_name)
-            .await?;
-        tmp_manifest_file.write_all(&manifest_data).await?;
-        tmp_manifest_file.flush().await?;
-        nix::unistd::fsync(tmp_manifest_file.as_raw_fd())?;
+        manifest_data = tokio::task::spawn_blocking(move || {
+            manifest.write_to_path(&tmp_manifest_name, &manifest_name)
+        })
+        .await??;
+    } else {
+        if let Err(err) = tokio::fs::rename(&tmp_manifest_name, &manifest_name).await {
+            bail!("{prefix}: Atomic rename file {manifest_name:?} failed - {err}");
+        }
     }
 
-    if let Err(err) = tokio::fs::rename(&tmp_manifest_name, &manifest_name).await {
-        bail!("{prefix}: Atomic rename file {manifest_name:?} failed - {err}");
-    }
     if let DatastoreBackend::S3(s3_client) = backend {
         let object_key = pbs_datastore::s3::object_key_from_path(
             &snapshot.relative_path(),
-- 
2.47.3