src/client/http_client.rs

   1 use failure::*;
   2
   3 use http::Uri;
   4 use hyper::Body;
   5 use hyper::client::Client;
   6 use xdg::BaseDirectories;
   7 use chrono::{DateTime, Utc};
   8 use std::collections::HashSet;
   9 use std::sync::{Arc, Mutex};
  10 use std::io::Write;
  11
  12 use http::{Request, Response};
  13 use http::header::HeaderValue;
  14
  15 use futures::*;
  16 use futures::stream::Stream;
  17 use std::sync::atomic::{AtomicUsize, Ordering};
  18 use tokio::sync::mpsc;
  19 use openssl::ssl::{SslConnector, SslMethod};
  20
  21 use serde_json::{json, Value};
  22 use url::percent_encoding::{percent_encode,  DEFAULT_ENCODE_SET};
  23
  24 use proxmox::tools::{
  25     digest_to_hex,
  26     fs::{file_get_json, file_set_contents},
  27 };
  28
  29 use crate::tools::{self, BroadcastFuture, tty};
  30 use crate::tools::futures::{cancellable, Canceller};
  31 use super::pipe_to_stream::*;
  32 use super::merge_known_chunks::*;
  33
  34 use crate::backup::*;
  35
  36 #[derive(Clone)]
  37 struct AuthInfo {
  38     username: String,
  39     ticket: String,
  40     token: String,
  41 }
  42
  43 /// HTTP(S) API client
  44 pub struct HttpClient {
  45     client: Client<hyper_openssl::HttpsConnector<hyper::client::HttpConnector>>,
  46     server: String,
  47     auth: BroadcastFuture<AuthInfo>,
  48 }
  49
  50 fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
  51
  52     let base = BaseDirectories::with_prefix("proxmox-backup")?;
  53
  54     // usually /run/user/<uid>/...
  55     let path = base.place_runtime_file("tickets")?;
  56
  57     let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600);
  58
  59     let mut data = file_get_json(&path, Some(json!({})))?;
  60
  61     let now = Utc::now().timestamp();
  62
  63     data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token});
  64
  65     let mut new_data = json!({});
  66
  67     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
  68
  69     let empty = serde_json::map::Map::new();
  70     for (server, info) in data.as_object().unwrap_or(&empty) {
  71         for (_user, uinfo) in info.as_object().unwrap_or(&empty) {
  72             if let Some(timestamp) = uinfo["timestamp"].as_i64() {
  73                 let age = now - timestamp;
  74                 if age < ticket_lifetime {
  75                     new_data[server][username] = uinfo.clone();
  76                 }
  77             }
  78         }
  79     }
  80
  81     file_set_contents(path, new_data.to_string().as_bytes(), Some(mode))?;
  82
  83     Ok(())
  84 }
  85
  86 fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
  87     let base = match BaseDirectories::with_prefix("proxmox-backup") {
  88         Ok(b) => b,
  89         _ => return None,
  90     };
  91
  92     // usually /run/user/<uid>/...
  93     let path = match base.place_runtime_file("tickets") {
  94         Ok(p) => p,
  95         _ => return None,
  96     };
  97
  98     let data = match file_get_json(&path, None) {
  99         Ok(v) => v,
 100         _ => return None,
 101     };
 102
 103     let now = Utc::now().timestamp();
 104
 105     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
 106
 107     if let Some(uinfo) = data[server][username].as_object() {
 108         if let Some(timestamp) = uinfo["timestamp"].as_i64() {
 109             let age = now - timestamp;
 110             if age < ticket_lifetime {
 111                 let ticket = match uinfo["ticket"].as_str() {
 112                     Some(t) => t,
 113                     None => return None,
 114                 };
 115                 let token = match uinfo["token"].as_str() {
 116                     Some(t) => t,
 117                     None => return None,
 118                 };
 119                 return Some((ticket.to_owned(), token.to_owned()));
 120             }
 121         }
 122     }
 123
 124     None
 125 }
 126
 127 impl HttpClient {
 128
 129     pub fn new(server: &str, username: &str) -> Result<Self, Error> {
 130         let client = Self::build_client();
 131
 132         let password = if let Some((ticket, _token)) = load_ticket_info(server, username) {
 133             ticket
 134         } else {
 135             Self::get_password(&username)?
 136         };
 137
 138         let login = Self::credentials(client.clone(), server.to_owned(), username.to_owned(), password);
 139
 140         Ok(Self {
 141             client,
 142             server: String::from(server),
 143             auth: BroadcastFuture::new(login),
 144         })
 145     }
 146
 147     fn get_password(_username: &str) -> Result<String, Error> {
 148         use std::env::VarError::*;
 149         match std::env::var("PBS_PASSWORD") {
 150             Ok(p) => return Ok(p),
 151             Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"),
 152             Err(NotPresent) => {
 153                 // Try another method
 154             }
 155         }
 156
 157         // If we're on a TTY, query the user for a password
 158         if tty::stdin_isatty() {
 159             return Ok(String::from_utf8(tty::read_password("Password: ")?)?);
 160         }
 161
 162         bail!("no password input mechanism available");
 163     }
 164
 165     fn build_client() -> Client<hyper_openssl::HttpsConnector<hyper::client::HttpConnector>> {
 166
 167         let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap();
 168
 169         ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE); // fixme!
 170
 171         let mut httpc = hyper::client::HttpConnector::new(1);
 172         httpc.set_nodelay(true); // important for h2 download performance!
 173         httpc.set_recv_buffer_size(Some(1024*1024)); //important for h2 download performance!
 174         httpc.enforce_http(false); // we want https...
 175
 176         let https = hyper_openssl::HttpsConnector::with_connector(httpc,  ssl_connector_builder).unwrap();
 177
 178         Client::builder()
 179         //.http2_initial_stream_window_size( (1 << 31) - 2)
 180         //.http2_initial_connection_window_size( (1 << 31) - 2)
 181             .build::<_, Body>(https)
 182     }
 183
 184     pub fn request(&self, mut req: Request<Body>) -> impl Future<Item=Value, Error=Error>  {
 185
 186         let login = self.auth.listen();
 187
 188         let client = self.client.clone();
 189
 190         login.and_then(move |auth| {
 191
 192             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 193             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 194             req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
 195
 196             let request = Self::api_request(client, req);
 197
 198             request
 199         })
 200     }
 201
 202     pub fn get(&self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 203
 204         let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
 205         self.request(req)
 206     }
 207
 208     pub fn delete(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 209
 210         let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
 211         self.request(req)
 212     }
 213
 214     pub fn post(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 215
 216         let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
 217         self.request(req)
 218     }
 219
 220     pub fn download<W: Write>(&mut self, path: &str, output: W) ->  impl Future<Item=W, Error=Error> {
 221
 222         let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
 223
 224         let login = self.auth.listen();
 225
 226         let client = self.client.clone();
 227
 228         login.and_then(move |auth| {
 229
 230             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 231             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 232
 233             client.request(req)
 234                 .map_err(Error::from)
 235                 .and_then(|resp| {
 236                     let status = resp.status();
 237                     if !status.is_success() {
 238                         future::Either::A(
 239                             HttpClient::api_response(resp)
 240                                 .and_then(|_| { bail!("unknown error"); })
 241                         )
 242                     } else {
 243                         future::Either::B(
 244                             resp.into_body()
 245                                 .map_err(Error::from)
 246                                 .fold(output, move |mut acc, chunk| {
 247                                     acc.write_all(&chunk)?;
 248                                     Ok::<_, Error>(acc)
 249                                 })
 250                         )
 251                     }
 252                 })
 253         })
 254     }
 255
 256     pub fn upload(
 257         &mut self,
 258         content_type: &str,
 259         body: Body,
 260         path: &str,
 261         data: Option<Value>,
 262     ) -> impl Future<Item=Value, Error=Error> {
 263
 264         let path = path.trim_matches('/');
 265         let mut url = format!("https://{}:8007/{}", &self.server, path);
 266
 267         if let Some(data) = data {
 268             let query = tools::json_object_to_query(data).unwrap();
 269             url.push('?');
 270             url.push_str(&query);
 271         }
 272
 273         let url: Uri = url.parse().unwrap();
 274
 275         let req = Request::builder()
 276             .method("POST")
 277             .uri(url)
 278             .header("User-Agent", "proxmox-backup-client/1.0")
 279             .header("Content-Type", content_type)
 280             .body(body).unwrap();
 281
 282         self.request(req)
 283     }
 284
 285     pub fn start_backup(
 286         &self,
 287         datastore: &str,
 288         backup_type: &str,
 289         backup_id: &str,
 290         backup_time: DateTime<Utc>,
 291         debug: bool,
 292     ) -> impl Future<Item=Arc<BackupClient>, Error=Error> {
 293
 294         let param = json!({
 295             "backup-type": backup_type,
 296             "backup-id": backup_id,
 297             "backup-time": backup_time.timestamp(),
 298             "store": datastore,
 299             "debug": debug
 300         });
 301
 302         let req = Self::request_builder(&self.server, "GET", "/api2/json/backup", Some(param)).unwrap();
 303
 304         self.start_h2_connection(req, String::from(PROXMOX_BACKUP_PROTOCOL_ID_V1!()))
 305             .map(|(h2, canceller)| BackupClient::new(h2, canceller))
 306     }
 307
 308     pub fn start_backup_reader(
 309         &self,
 310         datastore: &str,
 311         backup_type: &str,
 312         backup_id: &str,
 313         backup_time: DateTime<Utc>,
 314         debug: bool,
 315     ) -> impl Future<Item=Arc<BackupReader>, Error=Error> {
 316
 317         let param = json!({
 318             "backup-type": backup_type,
 319             "backup-id": backup_id,
 320             "backup-time": backup_time.timestamp(),
 321             "store": datastore,
 322             "debug": debug,
 323         });
 324         let req = Self::request_builder(&self.server, "GET", "/api2/json/reader", Some(param)).unwrap();
 325
 326         self.start_h2_connection(req, String::from(PROXMOX_BACKUP_READER_PROTOCOL_ID_V1!()))
 327             .map(|(h2, canceller)| BackupReader::new(h2, canceller))
 328     }
 329
 330     pub fn start_h2_connection(
 331         &self,
 332         mut req: Request<Body>,
 333         protocol_name: String,
 334     ) -> impl Future<Item=(H2Client, Canceller), Error=Error> {
 335
 336         let login = self.auth.listen();
 337         let client = self.client.clone();
 338
 339         login.and_then(move |auth| {
 340
 341             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 342             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 343             req.headers_mut().insert("UPGRADE", HeaderValue::from_str(&protocol_name).unwrap());
 344
 345             client.request(req)
 346                 .map_err(Error::from)
 347                 .and_then(|resp| {
 348
 349                     let status = resp.status();
 350                     if status != http::StatusCode::SWITCHING_PROTOCOLS {
 351                         future::Either::A(Self::api_response(resp).and_then(|_| { bail!("unknown error"); }))
 352                     } else {
 353                         future::Either::B(resp.into_body().on_upgrade().map_err(Error::from))
 354                     }
 355                 })
 356                 .and_then(|upgraded| {
 357                    let max_window_size = (1 << 31) - 2;
 358
 359                     h2::client::Builder::new()
 360                         .initial_connection_window_size(max_window_size)
 361                         .initial_window_size(max_window_size)
 362                         .max_frame_size(4*1024*1024)
 363                         .handshake(upgraded)
 364                         .map_err(Error::from)
 365                 })
 366                 .and_then(|(h2, connection)| {
 367                     let connection = connection
 368                         .map_err(|_| panic!("HTTP/2.0 connection failed"));
 369
 370                     let (connection, canceller) = cancellable(connection)?;
 371                     // A cancellable future returns an Option which is None when cancelled and
 372                     // Some when it finished instead, since we don't care about the return type we
 373                     // need to map it away:
 374                     let connection = connection.map(|_| ());
 375
 376                     // Spawn a new task to drive the connection state
 377                     hyper::rt::spawn(connection);
 378
 379                     // Wait until the `SendRequest` handle has available capacity.
 380                     Ok(h2.ready()
 381                        .map(move |c| (H2Client::new(c), canceller))
 382                        .map_err(Error::from))
 383                 })
 384                 .flatten()
 385         })
 386     }
 387
 388     fn credentials(
 389         client: Client<hyper_openssl::HttpsConnector<hyper::client::HttpConnector>>,
 390         server: String,
 391         username: String,
 392         password: String,
 393     ) -> Box<dyn Future<Item=AuthInfo, Error=Error> + Send> {
 394
 395         let server2 = server.clone();
 396
 397         let create_request = futures::future::lazy(move || {
 398             let data = json!({ "username": username, "password": password });
 399             let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
 400             Self::api_request(client, req)
 401         });
 402
 403         let login_future = create_request
 404             .and_then(move |cred| {
 405                 let auth = AuthInfo {
 406                     username: cred["data"]["username"].as_str().unwrap().to_owned(),
 407                     ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(),
 408                     token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
 409                 };
 410
 411                 let _ = store_ticket_info(&server2, &auth.username, &auth.ticket, &auth.token);
 412
 413                 Ok(auth)
 414             });
 415
 416         Box::new(login_future)
 417     }
 418
 419     fn api_response(response: Response<Body>) -> impl Future<Item=Value, Error=Error> {
 420
 421         let status = response.status();
 422
 423         response
 424             .into_body()
 425             .concat2()
 426             .map_err(Error::from)
 427             .and_then(move |data| {
 428
 429                 let text = String::from_utf8(data.to_vec()).unwrap();
 430                 if status.is_success() {
 431                     if text.len() > 0 {
 432                         let value: Value = serde_json::from_str(&text)?;
 433                         Ok(value)
 434                     } else {
 435                         Ok(Value::Null)
 436                     }
 437                 } else {
 438                     bail!("HTTP Error {}: {}", status, text);
 439                 }
 440             })
 441     }
 442
 443     fn api_request(
 444         client: Client<hyper_openssl::HttpsConnector<hyper::client::HttpConnector>>,
 445         req: Request<Body>
 446     ) -> impl Future<Item=Value, Error=Error> {
 447
 448         client.request(req)
 449             .map_err(Error::from)
 450             .and_then(Self::api_response)
 451     }
 452
 453     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
 454         let path = path.trim_matches('/');
 455         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 456
 457         if let Some(data) = data {
 458             if method == "POST" {
 459                 let request = Request::builder()
 460                     .method(method)
 461                     .uri(url)
 462                     .header("User-Agent", "proxmox-backup-client/1.0")
 463                     .header(hyper::header::CONTENT_TYPE, "application/json")
 464                     .body(Body::from(data.to_string()))?;
 465                 return Ok(request);
 466             } else {
 467                 let query = tools::json_object_to_query(data)?;
 468                 let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 469                 let request = Request::builder()
 470                     .method(method)
 471                     .uri(url)
 472                     .header("User-Agent", "proxmox-backup-client/1.0")
 473                     .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 474                     .body(Body::empty())?;
 475                 return Ok(request);
 476             }
 477         }
 478
 479         let request = Request::builder()
 480             .method(method)
 481             .uri(url)
 482             .header("User-Agent", "proxmox-backup-client/1.0")
 483             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 484             .body(Body::empty())?;
 485
 486         Ok(request)
 487     }
 488 }
 489
 490
 491 pub struct BackupReader {
 492     h2: H2Client,
 493     canceller: Canceller,
 494 }
 495
 496 impl Drop for BackupReader {
 497
 498     fn drop(&mut self) {
 499         self.canceller.cancel();
 500     }
 501 }
 502
 503 impl BackupReader {
 504
 505     pub fn new(h2: H2Client, canceller: Canceller) -> Arc<Self> {
 506         Arc::new(Self { h2, canceller: canceller })
 507     }
 508
 509     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 510         self.h2.get(path, param)
 511     }
 512
 513     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 514         self.h2.put(path, param)
 515     }
 516
 517     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 518         self.h2.post(path, param)
 519     }
 520
 521     pub fn download<W: Write>(
 522         &self,
 523         file_name: &str,
 524         output: W,
 525     ) -> impl Future<Item=W, Error=Error> {
 526         let path = "download";
 527         let param = json!({ "file-name": file_name });
 528         self.h2.download(path, Some(param), output)
 529     }
 530
 531     pub fn speedtest<W: Write>(
 532         &self,
 533         output: W,
 534     ) -> impl Future<Item=W, Error=Error> {
 535         self.h2.download("speedtest", None, output)
 536     }
 537
 538     pub fn download_chunk<W: Write>(
 539         &self,
 540         digest: &[u8; 32],
 541         output: W,
 542     ) -> impl Future<Item=W, Error=Error> {
 543         let path = "chunk";
 544         let param = json!({ "digest": digest_to_hex(digest) });
 545         self.h2.download(path, Some(param), output)
 546     }
 547
 548     pub fn force_close(self) {
 549         self.canceller.cancel();
 550     }
 551 }
 552
 553 pub struct BackupClient {
 554     h2: H2Client,
 555     canceller: Canceller,
 556 }
 557
 558 impl Drop for BackupClient {
 559
 560     fn drop(&mut self) {
 561         self.canceller.cancel();
 562     }
 563 }
 564
 565 pub struct BackupStats {
 566     pub size: u64,
 567     pub csum: [u8; 32],
 568 }
 569
 570 impl BackupClient {
 571
 572     pub fn new(h2: H2Client, canceller: Canceller) -> Arc<Self> {
 573         Arc::new(Self { h2, canceller })
 574     }
 575
 576     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 577         self.h2.get(path, param)
 578     }
 579
 580     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 581         self.h2.put(path, param)
 582     }
 583
 584     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 585         self.h2.post(path, param)
 586     }
 587
 588     pub fn finish(self: Arc<Self>) -> impl Future<Item=(), Error=Error> {
 589         self.h2.clone()
 590             .post("finish", None)
 591             .map(move |_| {
 592                 self.canceller.cancel();
 593             })
 594     }
 595
 596     pub fn force_close(self) {
 597         self.canceller.cancel();
 598     }
 599
 600     pub fn upload_blob_from_data(
 601         &self,
 602         data: Vec<u8>,
 603         file_name: &str,
 604         crypt_config: Option<Arc<CryptConfig>>,
 605         compress: bool,
 606         sign_only: bool,
 607      ) -> impl Future<Item=BackupStats, Error=Error> {
 608
 609         let h2 = self.h2.clone();
 610         let file_name = file_name.to_owned();
 611         let size = data.len() as u64;
 612
 613         futures::future::ok(())
 614             .and_then(move |_| {
 615                 let blob = if let Some(ref crypt_config) = crypt_config {
 616                     if sign_only {
 617                         DataBlob::create_signed(&data, crypt_config, compress)?
 618                     } else {
 619                         DataBlob::encode(&data, Some(crypt_config), compress)?
 620                     }
 621                 } else {
 622                     DataBlob::encode(&data, None, compress)?
 623                 };
 624
 625                 let raw_data = blob.into_inner();
 626                 Ok(raw_data)
 627             })
 628             .and_then(move |raw_data| {
 629                 let csum = openssl::sha::sha256(&raw_data);
 630                 let param = json!({"encoded-size": raw_data.len(), "file-name": file_name });
 631                 h2.upload("blob", Some(param), raw_data)
 632                     .map(move |_| {
 633                         BackupStats { size, csum }
 634                     })
 635             })
 636     }
 637
 638     pub fn upload_blob_from_file<P: AsRef<std::path::Path>>(
 639         &self,
 640         src_path: P,
 641         file_name: &str,
 642         crypt_config: Option<Arc<CryptConfig>>,
 643         compress: bool,
 644      ) -> impl Future<Item=BackupStats, Error=Error> {
 645
 646         let h2 = self.h2.clone();
 647         let file_name = file_name.to_owned();
 648         let src_path = src_path.as_ref().to_owned();
 649
 650         let task = tokio::fs::File::open(src_path.clone())
 651             .map_err(move |err| format_err!("unable to open file {:?} - {}", src_path, err))
 652             .and_then(move |file| {
 653                 let contents = vec![];
 654                 tokio::io::read_to_end(file, contents)
 655                     .map_err(Error::from)
 656                     .and_then(move |(_, contents)| {
 657                         let blob = if let Some(ref crypt_config) = crypt_config {
 658                             DataBlob::encode(&contents, Some(crypt_config), compress)?
 659                         } else {
 660                             DataBlob::encode(&contents, None, compress)?
 661                         };
 662                         let raw_data = blob.into_inner();
 663                         Ok((raw_data, contents.len() as u64))
 664                     })
 665                     .and_then(move |(raw_data, size)| {
 666                         let csum = openssl::sha::sha256(&raw_data);
 667                         let param = json!({"encoded-size": raw_data.len(), "file-name": file_name });
 668                         h2.upload("blob", Some(param), raw_data)
 669                             .map(move |_| {
 670                                 BackupStats { size, csum }
 671                             })
 672                     })
 673             });
 674
 675         task
 676     }
 677
 678     pub fn upload_stream(
 679         &self,
 680         archive_name: &str,
 681         stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
 682         prefix: &str,
 683         fixed_size: Option<u64>,
 684         crypt_config: Option<Arc<CryptConfig>>,
 685     ) -> impl Future<Item=BackupStats, Error=Error> {
 686
 687         let known_chunks = Arc::new(Mutex::new(HashSet::new()));
 688
 689         let h2 = self.h2.clone();
 690         let h2_2 = self.h2.clone();
 691         let h2_3 = self.h2.clone();
 692         let h2_4 = self.h2.clone();
 693
 694         let mut param = json!({ "archive-name": archive_name });
 695         if let Some(size) = fixed_size {
 696             param["size"] = size.into();
 697         }
 698
 699         let index_path = format!("{}_index", prefix);
 700         let close_path = format!("{}_close", prefix);
 701
 702         let prefix = prefix.to_owned();
 703
 704         Self::download_chunk_list(h2, &index_path, archive_name, known_chunks.clone())
 705             .and_then(move |_| {
 706                 h2_2.post(&index_path, Some(param))
 707             })
 708             .and_then(move |res| {
 709                 let wid = res.as_u64().unwrap();
 710                 Self::upload_chunk_info_stream(h2_3, wid, stream, &prefix, known_chunks.clone(), crypt_config)
 711                     .and_then(move |(chunk_count, size, _speed)| {
 712                         let param = json!({
 713                             "wid": wid ,
 714                             "chunk-count": chunk_count,
 715                             "size": size,
 716                         });
 717                         h2_4.post(&close_path, Some(param))
 718                             .map(move |_| {
 719                                 BackupStats { size: size as u64, csum: [0u8; 32] }
 720                             })
 721                     })
 722             })
 723     }
 724
 725     fn response_queue() -> (
 726         mpsc::Sender<h2::client::ResponseFuture>,
 727         sync::oneshot::Receiver<Result<(), Error>>
 728     ) {
 729         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100);
 730         let (verify_result_tx, verify_result_rx) = sync::oneshot::channel();
 731
 732         hyper::rt::spawn(
 733             verify_queue_rx
 734                 .map_err(Error::from)
 735                 .for_each(|response: h2::client::ResponseFuture| {
 736                     response
 737                         .map_err(Error::from)
 738                         .and_then(H2Client::h2api_response)
 739                         .and_then(|result| {
 740                             println!("RESPONSE: {:?}", result);
 741                             Ok(())
 742                         })
 743                         .map_err(|err| format_err!("pipelined request failed: {}", err))
 744                 })
 745                 .then(|result|
 746                       verify_result_tx.send(result)
 747                 )
 748                 .map_err(|_| { /* ignore closed channel */ })
 749         );
 750
 751         (verify_queue_tx, verify_result_rx)
 752     }
 753
 754     fn append_chunk_queue(h2: H2Client, wid: u64, path: String) -> (
 755         mpsc::Sender<(MergedChunkInfo, Option<h2::client::ResponseFuture>)>,
 756         sync::oneshot::Receiver<Result<(), Error>>
 757     ) {
 758         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(64);
 759         let (verify_result_tx, verify_result_rx) = sync::oneshot::channel();
 760
 761         let h2_2 = h2.clone();
 762
 763         hyper::rt::spawn(
 764             verify_queue_rx
 765                 .map_err(Error::from)
 766                 .and_then(move |(merged_chunk_info, response): (MergedChunkInfo, Option<h2::client::ResponseFuture>)| {
 767                     match (response, merged_chunk_info) {
 768                         (Some(response), MergedChunkInfo::Known(list)) => {
 769                             future::Either::A(
 770                                 response
 771                                     .map_err(Error::from)
 772                                     .and_then(H2Client::h2api_response)
 773                                     .and_then(move |_result| {
 774                                         Ok(MergedChunkInfo::Known(list))
 775                                     })
 776                             )
 777                         }
 778                         (None, MergedChunkInfo::Known(list)) => {
 779                             future::Either::B(future::ok(MergedChunkInfo::Known(list)))
 780                         }
 781                         _ => unreachable!(),
 782                     }
 783                 })
 784                 .merge_known_chunks()
 785                 .and_then(move |merged_chunk_info| {
 786                     match merged_chunk_info {
 787                         MergedChunkInfo::Known(chunk_list) => {
 788                             let mut digest_list = vec![];
 789                             let mut offset_list = vec![];
 790                             for (offset, digest) in chunk_list {
 791                                 //println!("append chunk {} (offset {})", proxmox::tools::digest_to_hex(&digest), offset);
 792                                 digest_list.push(digest_to_hex(&digest));
 793                                 offset_list.push(offset);
 794                             }
 795                             println!("append chunks list len ({})", digest_list.len());
 796                             let param = json!({ "wid": wid, "digest-list": digest_list, "offset-list": offset_list });
 797                             let mut request = H2Client::request_builder("localhost", "PUT", &path, None).unwrap();
 798                             request.headers_mut().insert(hyper::header::CONTENT_TYPE,  HeaderValue::from_static("application/json"));
 799                             let param_data = bytes::Bytes::from(param.to_string().as_bytes());
 800                             let upload_data = Some(param_data);
 801                             h2_2.send_request(request, upload_data)
 802                                 .and_then(move |response| {
 803                                     response
 804                                         .map_err(Error::from)
 805                                         .and_then(H2Client::h2api_response)
 806                                         .and_then(|_| Ok(()))
 807                                 })
 808                                 .map_err(|err| format_err!("pipelined request failed: {}", err))
 809                         }
 810                         _ => unreachable!(),
 811                     }
 812                 })
 813                 .for_each(|_| Ok(()))
 814                 .then(|result|
 815                       verify_result_tx.send(result)
 816                 )
 817                 .map_err(|_| { /* ignore closed channel */ })
 818         );
 819
 820         (verify_queue_tx, verify_result_rx)
 821     }
 822
 823     fn download_chunk_list(
 824         h2: H2Client,
 825         path: &str,
 826         archive_name: &str,
 827         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 828     ) -> impl Future<Item=(), Error=Error> {
 829
 830         let param = json!({ "archive-name": archive_name });
 831         let request = H2Client::request_builder("localhost", "GET", path, Some(param)).unwrap();
 832
 833         h2.send_request(request, None)
 834             .and_then(move |response| {
 835                 response
 836                     .map_err(Error::from)
 837                     .and_then(move |resp| {
 838                         let status = resp.status();
 839
 840                         if !status.is_success() {
 841                             future::Either::A(H2Client::h2api_response(resp).and_then(|_| { bail!("unknown error"); }))
 842                         } else {
 843                             future::Either::B(future::ok(resp.into_body()))
 844                         }
 845                     })
 846                     .and_then(move |mut body| {
 847
 848                         let mut release_capacity = body.release_capacity().clone();
 849
 850                         DigestListDecoder::new(body.map_err(Error::from))
 851                             .for_each(move |chunk| {
 852                                 let _ = release_capacity.release_capacity(chunk.len());
 853                                 println!("GOT DOWNLOAD {}", digest_to_hex(&chunk));
 854                                 known_chunks.lock().unwrap().insert(chunk);
 855                                 Ok(())
 856                             })
 857                        })
 858             })
 859     }
 860
 861     fn upload_chunk_info_stream(
 862         h2: H2Client,
 863         wid: u64,
 864         stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
 865         prefix: &str,
 866         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 867         crypt_config: Option<Arc<CryptConfig>>,
 868     ) -> impl Future<Item=(usize, usize, usize), Error=Error> {
 869
 870         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 871         let repeat2 = repeat.clone();
 872
 873         let stream_len = std::sync::Arc::new(AtomicUsize::new(0));
 874         let stream_len2 = stream_len.clone();
 875
 876         let append_chunk_path = format!("{}_index", prefix);
 877         let upload_chunk_path = format!("{}_chunk", prefix);
 878
 879         let (upload_queue, upload_result) = Self::append_chunk_queue(h2.clone(), wid, append_chunk_path.to_owned());
 880
 881         let start_time = std::time::Instant::now();
 882
 883         stream
 884             .and_then(move |data| {
 885
 886                 let chunk_len = data.len();
 887
 888                 repeat.fetch_add(1, Ordering::SeqCst);
 889                 let offset = stream_len.fetch_add(chunk_len, Ordering::SeqCst) as u64;
 890
 891                 let mut chunk_builder = DataChunkBuilder::new(data.as_ref())
 892                     .compress(true);
 893
 894                 if let Some(ref crypt_config) = crypt_config {
 895                     chunk_builder = chunk_builder.crypt_config(crypt_config);
 896                 }
 897
 898                 let mut known_chunks = known_chunks.lock().unwrap();
 899                 let digest = chunk_builder.digest();
 900                 let chunk_is_known = known_chunks.contains(digest);
 901                 if chunk_is_known {
 902                     Ok(MergedChunkInfo::Known(vec![(offset, *digest)]))
 903                 } else {
 904                     known_chunks.insert(*digest);
 905                     let chunk = chunk_builder.build()?;
 906                     Ok(MergedChunkInfo::New(ChunkInfo { chunk, chunk_len: chunk_len as u64, offset }))
 907                 }
 908             })
 909             .merge_known_chunks()
 910             .for_each(move |merged_chunk_info| {
 911
 912                 if let MergedChunkInfo::New(chunk_info) = merged_chunk_info {
 913                     let offset = chunk_info.offset;
 914                     let digest = *chunk_info.chunk.digest();
 915                     let digest_str = digest_to_hex(&digest);
 916                     let upload_queue = upload_queue.clone();
 917
 918                     println!("upload new chunk {} ({} bytes, offset {})", digest_str,
 919                              chunk_info.chunk_len, offset);
 920
 921                     let chunk_data = chunk_info.chunk.raw_data();
 922                     let param = json!({
 923                         "wid": wid,
 924                         "digest": digest_str,
 925                         "size": chunk_info.chunk_len,
 926                         "encoded-size": chunk_data.len(),
 927                     });
 928
 929                     let request = H2Client::request_builder("localhost", "POST", &upload_chunk_path, Some(param)).unwrap();
 930                     let upload_data = Some(bytes::Bytes::from(chunk_data));
 931
 932                     let new_info = MergedChunkInfo::Known(vec![(offset, digest)]);
 933
 934                     future::Either::A(
 935                         h2.send_request(request, upload_data)
 936                             .and_then(move |response| {
 937                                 upload_queue.clone().send((new_info, Some(response)))
 938                                     .map(|_| ()).map_err(Error::from)
 939                             })
 940                     )
 941                 } else {
 942
 943                     future::Either::B(
 944                         upload_queue.clone().send((merged_chunk_info, None))
 945                             .map(|_| ()).map_err(Error::from)
 946                     )
 947                 }
 948             })
 949             .then(move |result| {
 950                 //println!("RESULT {:?}", result);
 951                 upload_result.map_err(Error::from).and_then(|upload1_result| {
 952                     Ok(upload1_result.and(result))
 953                 })
 954             })
 955             .flatten()
 956             .and_then(move |_| {
 957                 let repeat = repeat2.load(Ordering::SeqCst);
 958                 let stream_len = stream_len2.load(Ordering::SeqCst);
 959                 let speed = ((stream_len*1000000)/(1024*1024))/(start_time.elapsed().as_micros() as usize);
 960                 println!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat, start_time.elapsed().as_secs(), speed);
 961                 if repeat > 0 {
 962                     println!("Average chunk size was {} bytes.", stream_len/repeat);
 963                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
 964                 }
 965                 Ok((repeat, stream_len, speed))
 966             })
 967     }
 968
 969     pub fn upload_speedtest(&self) -> impl Future<Item=usize, Error=Error> {
 970
 971         let mut data = vec![];
 972         // generate pseudo random byte sequence
 973         for i in 0..1024*1024 {
 974             for j in 0..4 {
 975                 let byte = ((i >> (j<<3))&0xff) as u8;
 976                 data.push(byte);
 977             }
 978         }
 979
 980         let item_len = data.len();
 981
 982         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 983         let repeat2 = repeat.clone();
 984
 985         let (upload_queue, upload_result) = Self::response_queue();
 986
 987         let start_time = std::time::Instant::now();
 988
 989         let h2 = self.h2.clone();
 990
 991         futures::stream::repeat(data)
 992             .take_while(move |_| {
 993                 repeat.fetch_add(1, Ordering::SeqCst);
 994                 Ok(start_time.elapsed().as_secs() < 5)
 995             })
 996             .for_each(move |data| {
 997                 let h2 = h2.clone();
 998
 999                 let upload_queue = upload_queue.clone();
1000
1001                 println!("send test data ({} bytes)", data.len());
1002                 let request = H2Client::request_builder("localhost", "POST", "speedtest", None).unwrap();
1003                 h2.send_request(request, Some(bytes::Bytes::from(data)))
1004                     .and_then(move |response| {
1005                         upload_queue.send(response)
1006                             .map(|_| ()).map_err(Error::from)
1007                     })
1008             })
1009             .then(move |result| {
1010                 println!("RESULT {:?}", result);
1011                 upload_result.map_err(Error::from).and_then(|upload1_result| {
1012                     Ok(upload1_result.and(result))
1013                 })
1014             })
1015             .flatten()
1016             .and_then(move |_| {
1017                 let repeat = repeat2.load(Ordering::SeqCst);
1018                 println!("Uploaded {} chunks in {} seconds.", repeat, start_time.elapsed().as_secs());
1019                 let speed = ((item_len*1000000*(repeat as usize))/(1024*1024))/(start_time.elapsed().as_micros() as usize);
1020                 if repeat > 0 {
1021                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
1022                 }
1023                 Ok(speed)
1024             })
1025     }
1026 }
1027
1028 #[derive(Clone)]
1029 pub struct H2Client {
1030     h2: h2::client::SendRequest<bytes::Bytes>,
1031 }
1032
1033 impl H2Client {
1034
1035     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
1036         Self { h2 }
1037     }
1038
1039     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
1040         let req = Self::request_builder("localhost", "GET", path, param).unwrap();
1041         self.request(req)
1042     }
1043
1044     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
1045         let req = Self::request_builder("localhost", "PUT", path, param).unwrap();
1046         self.request(req)
1047     }
1048
1049     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
1050         let req = Self::request_builder("localhost", "POST", path, param).unwrap();
1051         self.request(req)
1052     }
1053
1054     pub fn download<W: Write>(&self, path: &str, param: Option<Value>, output: W) -> impl Future<Item=W, Error=Error> {
1055         let request = Self::request_builder("localhost", "GET", path, param).unwrap();
1056
1057         self.send_request(request, None)
1058             .and_then(move |response| {
1059                 response
1060                     .map_err(Error::from)
1061                     .and_then(move |resp| {
1062                         let status = resp.status();
1063                         if !status.is_success() {
1064                             future::Either::A(
1065                                 H2Client::h2api_response(resp)
1066                                     .and_then(|_| { bail!("unknown error"); })
1067                             )
1068                         } else {
1069                             let mut body = resp.into_body();
1070                             let mut release_capacity = body.release_capacity().clone();
1071
1072                             future::Either::B(
1073                                 body
1074                                     .map_err(Error::from)
1075                                     .fold(output, move |mut acc, chunk| {
1076                                         let _ = release_capacity.release_capacity(chunk.len());
1077                                         acc.write_all(&chunk)?;
1078                                         Ok::<_, Error>(acc)
1079                                     })
1080                             )
1081                         }
1082                     })
1083             })
1084     }
1085
1086     pub fn upload(&self, path: &str, param: Option<Value>, data: Vec<u8>) -> impl Future<Item=Value, Error=Error> {
1087         let request = Self::request_builder("localhost", "POST", path, param).unwrap();
1088
1089         self.h2.clone()
1090             .ready()
1091             .map_err(Error::from)
1092             .and_then(move |mut send_request| {
1093                 let (response, stream) = send_request.send_request(request, false).unwrap();
1094                 PipeToSendStream::new(bytes::Bytes::from(data), stream)
1095                     .and_then(|_| {
1096                         response
1097                             .map_err(Error::from)
1098                             .and_then(Self::h2api_response)
1099                     })
1100             })
1101     }
1102
1103     fn request(
1104         &self,
1105         request: Request<()>,
1106     ) -> impl Future<Item=Value, Error=Error> {
1107
1108         self.send_request(request, None)
1109             .and_then(move |response| {
1110                 response
1111                     .map_err(Error::from)
1112                     .and_then(Self::h2api_response)
1113             })
1114     }
1115
1116     fn send_request(
1117         &self,
1118         request: Request<()>,
1119         data: Option<bytes::Bytes>,
1120     ) -> impl Future<Item=h2::client::ResponseFuture, Error=Error> {
1121
1122         self.h2.clone()
1123             .ready()
1124             .map_err(Error::from)
1125             .and_then(move |mut send_request| {
1126                 if let Some(data) = data {
1127                     let (response, stream) = send_request.send_request(request, false).unwrap();
1128                     future::Either::A(PipeToSendStream::new(data, stream)
1129                         .and_then(move |_| {
1130                             future::ok(response)
1131                         }))
1132                 } else {
1133                     let (response, _stream) = send_request.send_request(request, true).unwrap();
1134                     future::Either::B(future::ok(response))
1135                 }
1136             })
1137     }
1138
1139     fn h2api_response(response: Response<h2::RecvStream>) -> impl Future<Item=Value, Error=Error> {
1140
1141         let status = response.status();
1142
1143         let (_head, mut body) = response.into_parts();
1144
1145         // The `release_capacity` handle allows the caller to manage
1146         // flow control.
1147         //
1148         // Whenever data is received, the caller is responsible for
1149         // releasing capacity back to the server once it has freed
1150         // the data from memory.
1151         let mut release_capacity = body.release_capacity().clone();
1152
1153         body
1154             .map(move |chunk| {
1155                 // Let the server send more data.
1156                 let _ = release_capacity.release_capacity(chunk.len());
1157                 chunk
1158             })
1159             .concat2()
1160             .map_err(Error::from)
1161             .and_then(move |data| {
1162                 let text = String::from_utf8(data.to_vec()).unwrap();
1163                 if status.is_success() {
1164                     if text.len() > 0 {
1165                         let mut value: Value = serde_json::from_str(&text)?;
1166                         if let Some(map) = value.as_object_mut() {
1167                             if let Some(data) = map.remove("data") {
1168                                 return Ok(data);
1169                             }
1170                         }
1171                         bail!("got result without data property");
1172                     } else {
1173                         Ok(Value::Null)
1174                     }
1175                 } else {
1176                     bail!("HTTP Error {}: {}", status, text);
1177                 }
1178             })
1179     }
1180
1181     // Note: We always encode parameters with the url
1182     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<()>, Error> {
1183         let path = path.trim_matches('/');
1184
1185         if let Some(data) = data {
1186             let query = tools::json_object_to_query(data)?;
1187             // We detected problem with hyper around 6000 characters - seo we try to keep on the safe side
1188             if query.len() > 4096 { bail!("h2 query data too large ({} bytes) - please encode data inside body", query.len()); }
1189             let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
1190              let request = Request::builder()
1191                 .method(method)
1192                 .uri(url)
1193                 .header("User-Agent", "proxmox-backup-client/1.0")
1194                 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
1195                 .body(())?;
1196             return Ok(request);
1197         } else {
1198             let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
1199             let request = Request::builder()
1200                 .method(method)
1201                 .uri(url)
1202                 .header("User-Agent", "proxmox-backup-client/1.0")
1203                 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
1204                 .body(())?;
1205
1206             Ok(request)
1207         }
1208     }
1209 }