[PATCH proxmox-backup v3 3/3] 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.

Additionally, refactor `cleanup_unreferenced_files` to accept the set of
expected files (built by new `expected_files` helper) rather than the
manifest itself, avoiding the need to share the manifest across the
write and cleanup steps.

Link: https://bugzilla.proxmox.com/show_bug.cgi?id=6990
Signed-off-by: Jakob Klocker <j.klocker@proxmox.com>
---
 pbs-datastore/src/backup_info.rs | 21 ++++++-------
 pbs-datastore/src/manifest.rs    | 50 +++++++++++++++++++++++++++--
 src/server/pull.rs               | 54 ++++++++++++++++++++------------
 3 files changed, 91 insertions(+), 34 deletions(-)

diff --git a/pbs-datastore/src/backup_info.rs b/pbs-datastore/src/backup_info.rs
index be4ec8b3e..0c146533e 100644
--- a/pbs-datastore/src/backup_info.rs
+++ b/pbs-datastore/src/backup_info.rs
@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::fmt;
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::os::unix::prelude::OsStrExt;
@@ -16,8 +17,7 @@ use proxmox_systemd::escape_unit;
 
 use pbs_api_types::{
     ArchiveType, Authid, BACKUP_DATE_REGEX, BackupArchiveName, BackupGroupDeleteStats,
-    BackupNamespace, BackupType, CLIENT_LOG_BLOB_NAME, GroupFilter, MANIFEST_BLOB_NAME,
-    VerifyState,
+    BackupNamespace, BackupType, GroupFilter, MANIFEST_BLOB_NAME, VerifyState,
 };
 use pbs_config::{BackupLockGuard, open_backup_lockfile};
 
@@ -1029,17 +1029,14 @@ impl BackupDir {
         Ok(())
     }
 
-    /// Cleans up the backup directory by removing any file not mentioned in the manifest.
-    pub fn cleanup_unreferenced_files(&self, manifest: &BackupManifest) -> Result<(), Error> {
+    /// Removes any file in the backup directory not present in `files_to_keep`.
+    ///
+    /// The set is supplied by the caller (see `BackupManifest::expected_files`)
+    /// rather than derived here, so passing an incomplete set will delete files
+    /// that should be retained.
+    pub fn cleanup_unreferenced_files(&self, files_to_keep: &HashSet<String>) -> Result<(), Error> {
         let full_path = self.full_path();
 
-        let mut wanted_files = std::collections::HashSet::new();
-        wanted_files.insert(MANIFEST_BLOB_NAME.to_string());
-        wanted_files.insert(CLIENT_LOG_BLOB_NAME.to_string());
-        manifest.files().iter().for_each(|item| {
-            wanted_files.insert(item.filename.to_string());
-        });
-
         for item in proxmox_sys::fs::read_subdir(libc::AT_FDCWD, &full_path)?.flatten() {
             if let Some(file_type) = item.file_type() {
                 if file_type != nix::dir::Type::File {
@@ -1051,7 +1048,7 @@ impl BackupDir {
                 continue;
             };
             if let Ok(name) = std::str::from_utf8(file_name) {
-                if wanted_files.contains(name) {
+                if files_to_keep.contains(name) {
                     continue;
                 }
             }
diff --git a/pbs-datastore/src/manifest.rs b/pbs-datastore/src/manifest.rs
index 4d482c252..9250ecc97 100644
--- a/pbs-datastore/src/manifest.rs
+++ b/pbs-datastore/src/manifest.rs
@@ -1,9 +1,15 @@
-use anyhow::{Error, bail, format_err};
+use std::io::Write;
+use std::os::fd::AsRawFd;
+use std::path::Path;
 
+use anyhow::{Context, Error, bail, format_err};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 
-use pbs_api_types::{BackupArchiveName, BackupType, CryptMode, Fingerprint, SnapshotVerifyState};
+use pbs_api_types::{
+    BackupArchiveName, BackupType, CLIENT_LOG_BLOB_NAME, CryptMode, Fingerprint,
+    MANIFEST_BLOB_NAME, SnapshotVerifyState,
+};
 use pbs_tools::crypt_config::CryptConfig;
 
 use super::DataBlob;
@@ -295,6 +301,46 @@ 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 expected in a complete snapshot: every
+    /// manifest entry plus the manifest and client-log blobs themselves.
+    pub fn expected_files(&self) -> std::collections::HashSet<String> {
+        let mut expected = std::collections::HashSet::new();
+        expected.insert(MANIFEST_BLOB_NAME.to_string());
+        expected.insert(CLIENT_LOG_BLOB_NAME.to_string());
+        self.files().iter().for_each(|item| {
+            expected.insert(item.filename.to_string());
+        });
+        expected
+    }
 }
 
 impl TryFrom<DataBlob> for BackupManifest {
diff --git a/src/server/pull.rs b/src/server/pull.rs
index d4bd07d94..ac2d4065c 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,
@@ -741,7 +738,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,
@@ -849,6 +847,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| {
@@ -875,24 +878,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(),
@@ -911,7 +925,7 @@ async fn pull_snapshot<'a>(
     fetch_log(crypt_config).await?;
 
     let snapshot = snapshot.clone();
-    tokio::task::spawn_blocking(move || snapshot.cleanup_unreferenced_files(&manifest))
+    tokio::task::spawn_blocking(move || snapshot.cleanup_unreferenced_files(&files_to_keep))
         .await?
         .map_err(|err| format_err!("{prefix}: failed to cleanup unreferenced files - {err}"))?;
 
-- 
2.47.3