src/client/backup_reader.rs

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