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