[PATCH proxmox-backup 3/3] sync: decrypt client log on pull with matching decryption key

[Christian Ebner <c.ebner@proxmox.com>]

Client logs are currently fetched as is, not decrypting when pulling
even when there is a matching decryption key.

Fix this by:
- Factoring out the DataBlob decryption helper so it can be reused
- Pass the crypt config as conditional parameter to the fetch_log closure
- Extend the try_fetch_client_log() implementation to decrypt the source
  blob on the fly, if the crypt config is given.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/server/pull.rs | 44 ++++++++++++-------------------
 src/server/sync.rs | 65 ++++++++++++++++++++++++++++++++++++++++++----
 2 files changed, 76 insertions(+), 33 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index d37998f56..42c34732f 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -2,7 +2,7 @@
 
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet};
-use std::io::{Read, Seek};
+use std::io::Seek;
 use std::os::fd::AsRawFd;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
@@ -34,8 +34,7 @@ use pbs_datastore::index::IndexFile;
 use pbs_datastore::manifest::{BackupManifest, FileInfo};
 use pbs_datastore::read_chunk::AsyncReadChunk;
 use pbs_datastore::{
-    check_backup_owner, check_namespace_depth_limit, DataBlobReader, DataStore, DatastoreBackend,
-    StoreProgress,
+    check_backup_owner, check_namespace_depth_limit, DataStore, DatastoreBackend, StoreProgress,
 };
 use pbs_tools::bounded_join_set::BoundedJoinSet;
 use pbs_tools::buffered_logger::{BufferedLogger, LogLineSender};
@@ -43,7 +42,6 @@ use pbs_tools::crypt_config::CryptConfig;
 use pbs_tools::sha::sha256;
 use proxmox_human_byte::HumanByte;
 use proxmox_parallel_handler::ParallelHandler;
-use proxmox_sys::fs::{replace_file, CreateOptions};
 use tracing::{info, Level};
 
 pub(crate) struct PullTarget {
@@ -574,26 +572,15 @@ async fn pull_single_archive<'a>(
             })
             .with_context(|| archive_prefix.clone())?;
 
-            if crypt_config.is_some() {
-                let crypt_config = crypt_config.clone();
-
+            if let Some(crypt_config) = &crypt_config {
                 let tmp_dec_path = tmp_dec_path.clone();
 
-                let (csum, size) = tokio::task::spawn_blocking(move || {
-                    // must rewind again since after verifying cursor is at the end of the file
-                    tmpfile.rewind()?;
-                    let mut reader = DataBlobReader::new(tmpfile, crypt_config)?;
-                    let mut dec_raw_data = Vec::new();
-                    reader.read_to_end(&mut dec_raw_data)?;
-                    reader.finish()?;
-
-                    let blob = DataBlob::encode(&dec_raw_data, None, true)?;
-
-                    let (csum, size) = sha256(&mut blob.raw_data())?;
-                    replace_file(tmp_dec_path, blob.raw_data(), CreateOptions::new(), true)?;
-                    Ok((csum, size))
-                })
-                .await?
+                let (csum, size) = super::sync::decrypt_encrypted_data_blob(
+                    tmpfile,
+                    Arc::clone(crypt_config),
+                    tmp_dec_path,
+                )
+                .await
                 .map_err(|err: Error| format_err!("Failed when decrypting blob {path:?}: {err}"))
                 .with_context(|| archive_prefix.clone())?;
 
@@ -676,10 +663,10 @@ async fn pull_snapshot<'a>(
 
     let backend = &params.target.backend;
 
-    let fetch_log = async || {
+    let fetch_log = async |crypt_config: Option<Arc<CryptConfig>>| {
         if !client_log_name.exists() {
             reader
-                .try_fetch_client_log(&client_log_name, Arc::clone(&log_sender))
+                .try_fetch_client_log(&client_log_name, crypt_config, Arc::clone(&log_sender))
                 .await
                 .with_context(|| prefix.clone())?;
 
@@ -731,7 +718,8 @@ async fn pull_snapshot<'a>(
         })?;
 
         if manifest_blob.raw_data() == tmp_manifest_blob.raw_data() {
-            fetch_log().await?;
+            // requires no decryption since either none or both, source and target are encrypted
+            fetch_log(None).await?;
             cleanup().await?;
             return Ok(sync_stats); // nothing changed
         }
@@ -775,9 +763,9 @@ async fn pull_snapshot<'a>(
             let new_manifest = Arc::new(Mutex::new(BackupManifest::new(snapshot.into())));
             (Some(crypt_config), Some(new_manifest))
         }
-        (_, true) => {
+        (crypt_config, true) => {
             // nothing changed
-            fetch_log().await?;
+            fetch_log(crypt_config).await?;
             cleanup().await?;
             return Ok(sync_stats);
         }
@@ -908,7 +896,7 @@ async fn pull_snapshot<'a>(
             .with_context(|| prefix.clone())?;
     }
 
-    fetch_log().await?;
+    fetch_log(crypt_config).await?;
 
     snapshot
         .cleanup_unreferenced_files(&manifest)
diff --git a/src/server/sync.rs b/src/server/sync.rs
index d8eec844e..c5fab0e04 100644
--- a/src/server/sync.rs
+++ b/src/server/sync.rs
@@ -1,7 +1,8 @@
 //! Sync datastore contents from source to target, either in push or pull direction
 
 use std::collections::HashMap;
-use std::io::{Seek, Write};
+use std::fs::File;
+use std::io::{Read, Seek, Write};
 use std::ops::Deref;
 use std::path::Path;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -18,6 +19,7 @@ use tracing::{info, warn, Level};
 use proxmox_human_byte::HumanByte;
 use proxmox_rest_server::WorkerTask;
 use proxmox_router::HttpError;
+use proxmox_sys::fs::{replace_file, CreateOptions};
 
 use pbs_api_types::{
     Authid, BackupDir, BackupGroup, BackupNamespace, CryptMode, GroupListItem, SnapshotListItem,
@@ -28,9 +30,12 @@ use pbs_client::{BackupReader, BackupRepository, HttpClient, RemoteChunkReader};
 use pbs_config::CachedUserInfo;
 use pbs_datastore::data_blob::DataBlob;
 use pbs_datastore::read_chunk::AsyncReadChunk;
-use pbs_datastore::{BackupManifest, DataStore, ListNamespacesRecursive, LocalChunkReader};
+use pbs_datastore::{
+    BackupManifest, DataBlobReader, DataStore, ListNamespacesRecursive, LocalChunkReader,
+};
 use pbs_tools::buffered_logger::LogLineSender;
 use pbs_tools::crypt_config::CryptConfig;
+use pbs_tools::sha::sha256;
 
 use crate::backup::ListAccessibleBackupGroups;
 use crate::server::jobstate::Job;
@@ -107,6 +112,7 @@ pub(crate) trait SyncSourceReader: Send + Sync {
     async fn try_fetch_client_log(
         &self,
         to_path: &Path,
+        crypt_config: Option<Arc<CryptConfig>>,
         log_sender: Arc<LogLineSender>,
     ) -> Result<(), Error>;
 
@@ -174,6 +180,7 @@ impl SyncSourceReader for RemoteSourceReader {
     async fn try_fetch_client_log(
         &self,
         to_path: &Path,
+        crypt_config: Option<Arc<CryptConfig>>,
         log_sender: Arc<LogLineSender>,
     ) -> Result<(), Error> {
         let mut tmp_path = to_path.to_owned();
@@ -190,10 +197,17 @@ impl SyncSourceReader for RemoteSourceReader {
         let client_log_name = &CLIENT_LOG_BLOB_NAME;
         if let Ok(()) = self
             .backup_reader
-            .download(client_log_name.as_ref(), tmpfile)
+            .download(client_log_name.as_ref(), &tmpfile)
             .await
         {
-            if let Err(err) = std::fs::rename(&tmp_path, to_path) {
+            if let Some(crypt_config) = &crypt_config {
+                let (_csum, _size) = decrypt_encrypted_data_blob(
+                    tmpfile,
+                    Arc::clone(crypt_config),
+                    to_path.to_path_buf(),
+                )
+                .await?;
+            } else if let Err(err) = std::fs::rename(&tmp_path, to_path) {
                 bail!("Atomic rename file {to_path:?} failed - {err}");
             }
             log_sender
@@ -245,10 +259,24 @@ impl SyncSourceReader for LocalSourceReader {
     async fn try_fetch_client_log(
         &self,
         to_path: &Path,
+        crypt_config: Option<Arc<CryptConfig>>,
         log_sender: Arc<LogLineSender>,
     ) -> Result<(), Error> {
-        self.load_file_into(CLIENT_LOG_BLOB_NAME.as_ref(), to_path)
+        if let Some(crypt_config) = &crypt_config {
+            let mut from_path = self.dir.full_path();
+            from_path.push(CLIENT_LOG_BLOB_NAME.as_ref());
+            let blob_file = tokio::fs::File::open(from_path).await?;
+            let blob_file = blob_file.into_std().await;
+            let (_csum, _size) = decrypt_encrypted_data_blob(
+                blob_file,
+                Arc::clone(crypt_config),
+                to_path.to_path_buf(),
+            )
             .await?;
+        } else {
+            self.load_file_into(CLIENT_LOG_BLOB_NAME.as_ref(), to_path)
+                .await?;
+        }
         log_sender
             .log(
                 Level::INFO,
@@ -834,6 +862,33 @@ pub(super) fn exclude_not_verified_or_encrypted(
     false
 }
 
+/// Decrypt data blob stored in given file and store it to target path.
+///
+/// Returns the checksum and size of the resulting decrypted blob file.
+pub(super) async fn decrypt_encrypted_data_blob<P: AsRef<Path> + Send + 'static>(
+    mut blob_file: File,
+    crypt_config: Arc<CryptConfig>,
+    target_path: P,
+) -> Result<([u8; 32], u64), Error> {
+    let (csum, size) = tokio::task::spawn_blocking(move || {
+        // assure to start at the beginning of the file
+        blob_file.rewind()?;
+        let mut reader = DataBlobReader::new(blob_file, Some(crypt_config))?;
+        let mut dec_raw_data = Vec::new();
+        reader.read_to_end(&mut dec_raw_data)?;
+        reader.finish()?;
+
+        let blob = DataBlob::encode(&dec_raw_data, None, true)?;
+
+        let (csum, size) = sha256(&mut blob.raw_data())?;
+        replace_file(target_path, blob.raw_data(), CreateOptions::new(), true)?;
+        Ok::<([u8; 32], u64), Error>((csum, size))
+    })
+    .await??;
+
+    Ok((csum, size))
+}
+
 /// Helper to check if user has access to given encryption key and load it from config.
 pub(crate) fn check_privs_and_load_key_config(
     key_id: &str,
-- 
2.47.3