pbs-client/src/backup_reader.rs

   1 use anyhow::{format_err, Error};
   2 use std::fs::File;
   3 use std::io::{Seek, SeekFrom, Write};
   4 use std::os::unix::fs::OpenOptionsExt;
   5 use std::sync::Arc;
   6
   7 use futures::future::AbortHandle;
   8 use serde_json::{json, Value};
   9
  10 use pbs_api_types::{BackupDir, BackupNamespace};
  11 use pbs_datastore::data_blob::DataBlob;
  12 use pbs_datastore::data_blob_reader::DataBlobReader;
  13 use pbs_datastore::dynamic_index::DynamicIndexReader;
  14 use pbs_datastore::fixed_index::FixedIndexReader;
  15 use pbs_datastore::index::IndexFile;
  16 use pbs_datastore::manifest::MANIFEST_BLOB_NAME;
  17 use pbs_datastore::{BackupManifest, PROXMOX_BACKUP_READER_PROTOCOL_ID_V1};
  18 use pbs_tools::crypt_config::CryptConfig;
  19 use pbs_tools::sha::sha256;
  20
  21 use super::{H2Client, HttpClient};
  22
  23 /// Backup Reader
  24 pub struct BackupReader {
  25     h2: H2Client,
  26     abort: AbortHandle,
  27     crypt_config: Option<Arc<CryptConfig>>,
  28 }
  29
  30 impl Drop for BackupReader {
  31     fn drop(&mut self) {
  32         self.abort.abort();
  33     }
  34 }
  35
  36 impl BackupReader {
  37     fn new(h2: H2Client, abort: AbortHandle, crypt_config: Option<Arc<CryptConfig>>) -> Arc<Self> {
  38         Arc::new(Self {
  39             h2,
  40             abort,
  41             crypt_config,
  42         })
  43     }
  44
  45     /// Create a new instance by upgrading the connection at '/api2/json/reader'
  46     pub async fn start(
  47         client: HttpClient,
  48         crypt_config: Option<Arc<CryptConfig>>,
  49         datastore: &str,
  50         ns: &BackupNamespace,
  51         backup: &BackupDir,
  52         debug: bool,
  53     ) -> Result<Arc<BackupReader>, Error> {
  54         let mut param = json!({
  55             "backup-type": backup.ty(),
  56             "backup-id": backup.id(),
  57             "backup-time": backup.time,
  58             "store": datastore,
  59             "debug": debug,
  60         });
  61
  62         if !ns.is_root() {
  63             param["ns"] = serde_json::to_value(ns)?;
  64         }
  65
  66         let req = HttpClient::request_builder(
  67             client.server(),
  68             client.port(),
  69             "GET",
  70             "/api2/json/reader",
  71             Some(param),
  72         )
  73         .unwrap();
  74
  75         let (h2, abort) = client
  76             .start_h2_connection(req, String::from(PROXMOX_BACKUP_READER_PROTOCOL_ID_V1!()))
  77             .await?;
  78
  79         Ok(BackupReader::new(h2, abort, crypt_config))
  80     }
  81
  82     /// Execute a GET request
  83     pub async fn get(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
  84         self.h2.get(path, param).await
  85     }
  86
  87     /// Execute a PUT request
  88     pub async fn put(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
  89         self.h2.put(path, param).await
  90     }
  91
  92     /// Execute a POST request
  93     pub async fn post(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
  94         self.h2.post(path, param).await
  95     }
  96
  97     /// Execute a GET request and send output to a writer
  98     pub async fn download<W: Write + Send>(&self, file_name: &str, output: W) -> Result<(), Error> {
  99         let path = "download";
 100         let param = json!({ "file-name": file_name });
 101         self.h2.download(path, Some(param), output).await
 102     }
 103
 104     /// Execute a special GET request and send output to a writer
 105     ///
 106     /// This writes random data, and is only useful to test download speed.
 107     pub async fn speedtest<W: Write + Send>(&self, output: W) -> Result<(), Error> {
 108         self.h2.download("speedtest", None, output).await
 109     }
 110
 111     /// Download a specific chunk
 112     pub async fn download_chunk<W: Write + Send>(
 113         &self,
 114         digest: &[u8; 32],
 115         output: W,
 116     ) -> Result<(), Error> {
 117         let path = "chunk";
 118         let param = json!({ "digest": hex::encode(digest) });
 119         self.h2.download(path, Some(param), output).await
 120     }
 121
 122     pub fn force_close(self) {
 123         self.abort.abort();
 124     }
 125
 126     /// Download backup manifest (index.json)
 127     ///
 128     /// The manifest signature is verified if we have a crypt_config.
 129     pub async fn download_manifest(&self) -> Result<(BackupManifest, Vec<u8>), Error> {
 130         let mut raw_data = Vec::with_capacity(64 * 1024);
 131         self.download(MANIFEST_BLOB_NAME, &mut raw_data).await?;
 132         let blob = DataBlob::load_from_reader(&mut &raw_data[..])?;
 133         // no expected digest available
 134         let data = blob.decode(None, None)?;
 135
 136         let manifest =
 137             BackupManifest::from_data(&data[..], self.crypt_config.as_ref().map(Arc::as_ref))?;
 138
 139         Ok((manifest, data))
 140     }
 141
 142     /// Download a .blob file
 143     ///
 144     /// This creates a temporary file in /tmp (using O_TMPFILE). The data is verified using
 145     /// the provided manifest.
 146     pub async fn download_blob(
 147         &self,
 148         manifest: &BackupManifest,
 149         name: &str,
 150     ) -> Result<DataBlobReader<'_, File>, Error> {
 151         let mut tmpfile = std::fs::OpenOptions::new()
 152             .write(true)
 153             .read(true)
 154             .custom_flags(libc::O_TMPFILE)
 155             .open("/tmp")?;
 156
 157         self.download(name, &mut tmpfile).await?;
 158
 159         tmpfile.seek(SeekFrom::Start(0))?;
 160         let (csum, size) = sha256(&mut tmpfile)?;
 161         manifest.verify_file(name, &csum, size)?;
 162
 163         tmpfile.seek(SeekFrom::Start(0))?;
 164
 165         DataBlobReader::new(tmpfile, self.crypt_config.clone())
 166     }
 167
 168     /// Download dynamic index file
 169     ///
 170     /// This creates a temporary file in /tmp (using O_TMPFILE). The index is verified using
 171     /// the provided manifest.
 172     pub async fn download_dynamic_index(
 173         &self,
 174         manifest: &BackupManifest,
 175         name: &str,
 176     ) -> Result<DynamicIndexReader, Error> {
 177         let mut tmpfile = std::fs::OpenOptions::new()
 178             .write(true)
 179             .read(true)
 180             .custom_flags(libc::O_TMPFILE)
 181             .open("/tmp")?;
 182
 183         self.download(name, &mut tmpfile).await?;
 184
 185         let index = DynamicIndexReader::new(tmpfile)
 186             .map_err(|err| format_err!("unable to read dynamic index '{}' - {}", name, err))?;
 187
 188         // Note: do not use values stored in index (not trusted) - instead, computed them again
 189         let (csum, size) = index.compute_csum();
 190         manifest.verify_file(name, &csum, size)?;
 191
 192         Ok(index)
 193     }
 194
 195     /// Download fixed index file
 196     ///
 197     /// This creates a temporary file in /tmp (using O_TMPFILE). The index is verified using
 198     /// the provided manifest.
 199     pub async fn download_fixed_index(
 200         &self,
 201         manifest: &BackupManifest,
 202         name: &str,
 203     ) -> Result<FixedIndexReader, Error> {
 204         let mut tmpfile = std::fs::OpenOptions::new()
 205             .write(true)
 206             .read(true)
 207             .custom_flags(libc::O_TMPFILE)
 208             .open("/tmp")?;
 209
 210         self.download(name, &mut tmpfile).await?;
 211
 212         let index = FixedIndexReader::new(tmpfile)
 213             .map_err(|err| format_err!("unable to read fixed index '{}' - {}", name, err))?;
 214
 215         // Note: do not use values stored in index (not trusted) - instead, computed them again
 216         let (csum, size) = index.compute_csum();
 217         manifest.verify_file(name, &csum, size)?;
 218
 219         Ok(index)
 220     }
 221 }