pbs-client/src/backup_reader.rs

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