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::Utc;
   8 use std::collections::HashSet;
   9 use std::sync::{Arc, Mutex};
  10
  11 use http::{Request, Response};
  12 use http::header::HeaderValue;
  13
  14 use futures::*;
  15 use futures::stream::Stream;
  16 use std::sync::atomic::{AtomicUsize, Ordering};
  17 use tokio::sync::mpsc;
  18
  19 use serde_json::{json, Value};
  20 use url::percent_encoding::{percent_encode,  DEFAULT_ENCODE_SET};
  21
  22 use crate::tools::{self, BroadcastFuture, tty};
  23 use super::pipe_to_stream::*;
  24
  25 #[derive(Clone)]
  26 struct AuthInfo {
  27     username: String,
  28     ticket: String,
  29     token: String,
  30 }
  31
  32 /// HTTP(S) API client
  33 pub struct HttpClient {
  34     client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
  35     server: String,
  36     auth: BroadcastFuture<AuthInfo>,
  37 }
  38
  39 fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
  40
  41     let base = BaseDirectories::with_prefix("proxmox-backup")?;
  42
  43     // usually /run/user/<uid>/...
  44     let path = base.place_runtime_file("tickets")?;
  45
  46     let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600);
  47
  48     let mut data = tools::file_get_json(&path, Some(json!({})))?;
  49
  50     let now = Utc::now().timestamp();
  51
  52     data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token});
  53
  54     let mut new_data = json!({});
  55
  56     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
  57
  58     let empty = serde_json::map::Map::new();
  59     for (server, info) in data.as_object().unwrap_or(&empty) {
  60         for (_user, uinfo) in info.as_object().unwrap_or(&empty) {
  61             if let Some(timestamp) = uinfo["timestamp"].as_i64() {
  62                 let age = now - timestamp;
  63                 if age < ticket_lifetime {
  64                     new_data[server][username] = uinfo.clone();
  65                 }
  66             }
  67         }
  68     }
  69
  70     tools::file_set_contents(path, new_data.to_string().as_bytes(), Some(mode))?;
  71
  72     Ok(())
  73 }
  74
  75 fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
  76     let base = match BaseDirectories::with_prefix("proxmox-backup") {
  77         Ok(b) => b,
  78         _ => return None,
  79     };
  80
  81     // usually /run/user/<uid>/...
  82     let path = match base.place_runtime_file("tickets") {
  83         Ok(p) => p,
  84         _ => return None,
  85     };
  86
  87     let data = match tools::file_get_json(&path, None) {
  88         Ok(v) => v,
  89         _ => return None,
  90     };
  91
  92     let now = Utc::now().timestamp();
  93
  94     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
  95
  96     if let Some(uinfo) = data[server][username].as_object() {
  97         if let Some(timestamp) = uinfo["timestamp"].as_i64() {
  98             let age = now - timestamp;
  99             if age < ticket_lifetime {
 100                 let ticket = match uinfo["ticket"].as_str() {
 101                     Some(t) => t,
 102                     None => return None,
 103                 };
 104                 let token = match uinfo["token"].as_str() {
 105                     Some(t) => t,
 106                     None => return None,
 107                 };
 108                 return Some((ticket.to_owned(), token.to_owned()));
 109             }
 110         }
 111     }
 112
 113     None
 114 }
 115
 116 impl HttpClient {
 117
 118     pub fn new(server: &str, username: &str) -> Result<Self, Error> {
 119         let client = Self::build_client();
 120
 121         let password = if let Some((ticket, _token)) = load_ticket_info(server, username) {
 122             ticket
 123         } else {
 124             Self::get_password(&username)?
 125         };
 126
 127         let login = Self::credentials(client.clone(), server.to_owned(), username.to_owned(), password);
 128
 129         Ok(Self {
 130             client,
 131             server: String::from(server),
 132             auth: BroadcastFuture::new(login),
 133         })
 134     }
 135
 136     fn get_password(_username: &str) -> Result<String, Error> {
 137         use std::env::VarError::*;
 138         match std::env::var("PBS_PASSWORD") {
 139             Ok(p) => return Ok(p),
 140             Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"),
 141             Err(NotPresent) => {
 142                 // Try another method
 143             }
 144         }
 145
 146         // If we're on a TTY, query the user for a password
 147         if tty::stdin_isatty() {
 148             return Ok(String::from_utf8(tty::read_password("Password: ")?)?);
 149         }
 150
 151         bail!("no password input mechanism available");
 152     }
 153
 154     fn build_client() -> Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>> {
 155         let mut builder = native_tls::TlsConnector::builder();
 156         // FIXME: We need a CLI option for this!
 157         builder.danger_accept_invalid_certs(true);
 158         let tlsconnector = builder.build().unwrap();
 159         let mut httpc = hyper::client::HttpConnector::new(1);
 160         //httpc.set_nodelay(true); // not sure if this help?
 161         httpc.enforce_http(false); // we want https...
 162         let mut https = hyper_tls::HttpsConnector::from((httpc, tlsconnector));
 163         https.https_only(true); // force it!
 164         Client::builder()
 165         //.http2_initial_stream_window_size( (1 << 31) - 2)
 166         //.http2_initial_connection_window_size( (1 << 31) - 2)
 167             .build::<_, Body>(https)
 168     }
 169
 170     pub fn request(&self, mut req: Request<Body>) -> impl Future<Item=Value, Error=Error>  {
 171
 172         let login = self.auth.listen();
 173
 174         let client = self.client.clone();
 175
 176         login.and_then(move |auth| {
 177
 178             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 179             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 180             req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
 181
 182             let request = Self::api_request(client, req);
 183
 184             request
 185         })
 186     }
 187
 188     pub fn get(&self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 189
 190         let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
 191         self.request(req)
 192     }
 193
 194     pub fn delete(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 195
 196         let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
 197         self.request(req)
 198     }
 199
 200     pub fn post(&mut self, path: &str, data: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 201
 202         let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
 203         self.request(req)
 204     }
 205
 206     pub fn download(&mut self, path: &str, mut output: Box<dyn std::io::Write + Send>) ->  impl Future<Item=(), Error=Error> {
 207
 208         let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
 209
 210         let login = self.auth.listen();
 211
 212         let client = self.client.clone();
 213
 214         login.and_then(move |auth| {
 215
 216             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 217             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 218
 219             client.request(req)
 220                 .map_err(Error::from)
 221                 .and_then(|resp| {
 222
 223                     let _status = resp.status(); // fixme: ??
 224
 225                     resp.into_body()
 226                         .map_err(Error::from)
 227                         .for_each(move |chunk| {
 228                             output.write_all(&chunk)?;
 229                             Ok(())
 230                         })
 231
 232                 })
 233         })
 234     }
 235
 236     pub fn upload(&mut self, content_type: &str, body: Body, path: &str) -> impl Future<Item=Value, Error=Error> {
 237
 238         let path = path.trim_matches('/');
 239         let url: Uri = format!("https://{}:8007/{}", &self.server, path).parse().unwrap();
 240
 241         let req = Request::builder()
 242             .method("POST")
 243             .uri(url)
 244             .header("User-Agent", "proxmox-backup-client/1.0")
 245             .header("Content-Type", content_type)
 246             .body(body).unwrap();
 247
 248         self.request(req)
 249     }
 250
 251     pub fn start_backup(
 252         &self,
 253         datastore: &str,
 254         backup_type: &str,
 255         backup_id: &str,
 256     ) -> impl Future<Item=BackupClient, Error=Error> {
 257
 258         let path = format!("/api2/json/admin/datastore/{}/backup", datastore);
 259         let param = json!({"backup-type": backup_type, "backup-id": backup_id});
 260         let mut req = Self::request_builder(&self.server, "GET", &path, Some(param)).unwrap();
 261
 262         let login = self.auth.listen();
 263
 264         let client = self.client.clone();
 265
 266         login.and_then(move |auth| {
 267
 268             let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 269             req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 270             req.headers_mut().insert("UPGRADE", HeaderValue::from_str("proxmox-backup-protocol-h2").unwrap());
 271
 272             client.request(req)
 273                 .map_err(Error::from)
 274                 .and_then(|resp| {
 275
 276                     let status = resp.status();
 277                     if status != http::StatusCode::SWITCHING_PROTOCOLS {
 278                         future::Either::A(Self::api_response(resp).and_then(|_| { bail!("unknown error"); }))
 279                     } else {
 280                         future::Either::B(resp.into_body().on_upgrade().map_err(Error::from))
 281                     }
 282                 })
 283                 .and_then(|upgraded| {
 284                     h2::client::handshake(upgraded).map_err(Error::from)
 285                 })
 286                 .and_then(|(h2, connection)| {
 287                     let connection = connection
 288                         .map_err(|_| panic!("HTTP/2.0 connection failed"));
 289
 290                     // Spawn a new task to drive the connection state
 291                     hyper::rt::spawn(connection);
 292
 293                     // Wait until the `SendRequest` handle has available capacity.
 294                     h2.ready()
 295                         .map(BackupClient::new)
 296                         .map_err(Error::from)
 297                 })
 298         })
 299     }
 300
 301     fn credentials(
 302         client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
 303         server: String,
 304         username: String,
 305         password: String,
 306     ) -> Box<Future<Item=AuthInfo, Error=Error> + Send> {
 307
 308         let server2 = server.clone();
 309
 310         let create_request = futures::future::lazy(move || {
 311             let data = json!({ "username": username, "password": password });
 312             let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
 313             Self::api_request(client, req)
 314         });
 315
 316         let login_future = create_request
 317             .and_then(move |cred| {
 318                 let auth = AuthInfo {
 319                     username: cred["data"]["username"].as_str().unwrap().to_owned(),
 320                     ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(),
 321                     token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
 322                 };
 323
 324                 let _ = store_ticket_info(&server2, &auth.username, &auth.ticket, &auth.token);
 325
 326                 Ok(auth)
 327             });
 328
 329         Box::new(login_future)
 330     }
 331
 332     fn api_response(response: Response<Body>) -> impl Future<Item=Value, Error=Error> {
 333
 334         let status = response.status();
 335
 336         response
 337             .into_body()
 338             .concat2()
 339             .map_err(Error::from)
 340             .and_then(move |data| {
 341
 342                 let text = String::from_utf8(data.to_vec()).unwrap();
 343                 if status.is_success() {
 344                     if text.len() > 0 {
 345                         let value: Value = serde_json::from_str(&text)?;
 346                         Ok(value)
 347                     } else {
 348                         Ok(Value::Null)
 349                     }
 350                 } else {
 351                     bail!("HTTP Error {}: {}", status, text);
 352                 }
 353             })
 354     }
 355
 356     fn api_request(
 357         client: Client<hyper_tls::HttpsConnector<hyper::client::HttpConnector>>,
 358         req: Request<Body>
 359     ) -> impl Future<Item=Value, Error=Error> {
 360
 361         client.request(req)
 362             .map_err(Error::from)
 363             .and_then(Self::api_response)
 364     }
 365
 366     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
 367         let path = path.trim_matches('/');
 368         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 369
 370         if let Some(data) = data {
 371             if method == "POST" {
 372                 let request = Request::builder()
 373                     .method(method)
 374                     .uri(url)
 375                     .header("User-Agent", "proxmox-backup-client/1.0")
 376                     .header(hyper::header::CONTENT_TYPE, "application/json")
 377                     .body(Body::from(data.to_string()))?;
 378                 return Ok(request);
 379             } else {
 380                 let query = tools::json_object_to_query(data)?;
 381                 let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 382                 let request = Request::builder()
 383                     .method(method)
 384                     .uri(url)
 385                     .header("User-Agent", "proxmox-backup-client/1.0")
 386                     .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 387                     .body(Body::empty())?;
 388                 return Ok(request);
 389             }
 390         }
 391
 392         let request = Request::builder()
 393             .method(method)
 394             .uri(url)
 395             .header("User-Agent", "proxmox-backup-client/1.0")
 396             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 397             .body(Body::empty())?;
 398
 399         Ok(request)
 400     }
 401 }
 402
 403 //#[derive(Clone)]
 404 pub struct BackupClient {
 405     h2: H2Client,
 406 }
 407
 408 impl BackupClient {
 409
 410     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
 411         Self { h2: H2Client::new(h2) }
 412     }
 413
 414     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 415         self.h2.get(path, param)
 416     }
 417
 418     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 419         self.h2.put(path, param)
 420     }
 421
 422     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 423         self.h2.post(path, param)
 424     }
 425
 426     fn response_queue() -> (
 427         mpsc::Sender<h2::client::ResponseFuture>,
 428         sync::oneshot::Receiver<Result<(), Error>>
 429     ) {
 430         let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100);
 431         let (verify_result_tx, verify_result_rx) = sync::oneshot::channel();
 432
 433         hyper::rt::spawn(
 434             verify_queue_rx
 435                 .map_err(Error::from)
 436                 .for_each(|response: h2::client::ResponseFuture| {
 437                     response
 438                         .map_err(Error::from)
 439                         .and_then(H2Client::h2api_response)
 440                         .and_then(|result| {
 441                             println!("RESPONSE: {:?}", result);
 442                             Ok(())
 443                         })
 444                         .map_err(|err| format_err!("pipelined request failed: {}", err))
 445                 })
 446                 .then(|result|
 447                       verify_result_tx.send(result)
 448                 )
 449                 .map_err(|_| { /* ignore closed channel */ })
 450         );
 451
 452         (verify_queue_tx, verify_result_rx)
 453     }
 454
 455     fn download_chunk_list(
 456         h2: H2Client,
 457         path: &str,
 458         archive_name: &str,
 459         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 460     ) -> impl Future<Item=(), Error=Error> {
 461
 462         let param = json!({ "archive-name": archive_name });
 463         let request = H2Client::request_builder("localhost", "GET", path, Some(param)).unwrap();
 464
 465         h2.send_request(request, None)
 466             .and_then(move |response| {
 467                 response
 468                     .map_err(Error::from)
 469                     .and_then(move |resp| {
 470                         let status = resp.status();
 471                         if !status.is_success() {
 472                             bail!("download chunk list failed with status {}", status);
 473                         }
 474
 475                         let (_head, body) = resp.into_parts();
 476
 477                         Ok(body)
 478                     })
 479                     .and_then(move |mut body| {
 480
 481                         let mut release_capacity = body.release_capacity().clone();
 482
 483                         crate::backup::DigestListDecoder::new(body.map_err(Error::from))
 484                             .for_each(move |chunk| {
 485                                 let _ = release_capacity.release_capacity(chunk.len());
 486                                 println!("GOT DOWNLOAD {}", tools::digest_to_hex(&chunk));
 487                                 known_chunks.lock().unwrap().insert(chunk);
 488                                 Ok(())
 489                             })
 490                        })
 491             })
 492     }
 493
 494     pub fn finish(&self) -> impl Future<Item=(), Error=Error> {
 495         self.h2.clone().post("finish", None).map(|_| ())
 496     }
 497
 498     pub fn upload_dynamic_stream(
 499         &self,
 500         archive_name: &str,
 501         stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
 502     ) -> impl Future<Item=(), Error=Error> {
 503
 504         let known_chunks = Arc::new(Mutex::new(HashSet::new()));
 505
 506         let h2 = self.h2.clone();
 507         let h2_2 = self.h2.clone();
 508         let h2_3 = self.h2.clone();
 509         let h2_4 = self.h2.clone();
 510
 511         let param = json!({ "archive-name": archive_name });
 512
 513         Self::download_chunk_list(h2, "dynamic_index", archive_name, known_chunks.clone())
 514             .and_then(move |_| {
 515                 h2_2.post("dynamic_index", Some(param))
 516             })
 517             .and_then(move |res| {
 518                 let wid = res.as_u64().unwrap();
 519                 Self::upload_stream(h2_3, wid, stream, known_chunks.clone())
 520                     .and_then(move |_size| {
 521                         h2_4.post("dynamic_close", Some(json!({ "wid": wid })))
 522                     })
 523                     .map(|_| ())
 524             })
 525     }
 526
 527     fn upload_stream(
 528         h2: H2Client,
 529         wid: u64,
 530         stream: impl Stream<Item=bytes::BytesMut, Error=Error>,
 531         known_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
 532     ) -> impl Future<Item=usize, Error=Error> {
 533
 534         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 535         let repeat2 = repeat.clone();
 536
 537         let stream_len = std::sync::Arc::new(AtomicUsize::new(0));
 538         let stream_len2 = stream_len.clone();
 539
 540         let (upload_queue, upload_result) = Self::response_queue();
 541
 542         let start_time = std::time::Instant::now();
 543
 544         stream
 545             .for_each(move |data| {
 546                 let h2 = h2.clone();
 547
 548                 repeat.fetch_add(1, Ordering::SeqCst);
 549                 stream_len.fetch_add(data.len(), Ordering::SeqCst);
 550
 551                 let upload_queue = upload_queue.clone();
 552
 553                 let digest = openssl::sha::sha256(&data);
 554
 555                 let mut known_chunks = known_chunks.lock().unwrap();
 556                 let chunk_is_known = known_chunks.contains(&digest);
 557
 558                 let upload_data;
 559                 let request;
 560
 561                 if chunk_is_known {
 562                     println!("append existing chunk ({} bytes)", data.len());
 563                     let param = json!({ "wid": wid, "digest": tools::digest_to_hex(&digest) });
 564                     request = H2Client::request_builder("localhost", "PUT", "dynamic_index", Some(param)).unwrap();
 565                     upload_data = None;
 566                 } else {
 567                     println!("upload new chunk {} ({} bytes)", tools::digest_to_hex(&digest), data.len());
 568                     known_chunks.insert(digest);
 569                     let param = json!({ "wid": wid, "size" : data.len() });
 570                     request = H2Client::request_builder("localhost", "POST", "dynamic_chunk", Some(param)).unwrap();
 571                     upload_data = Some(bytes::Bytes::from(data));
 572                 }
 573
 574                 h2.send_request(request, upload_data)
 575                     .and_then(move |response| {
 576                         upload_queue.send(response)
 577                             .map(|_| ()).map_err(Error::from)
 578                     })
 579             })
 580             .then(move |result| {
 581                 println!("RESULT {:?}", result);
 582                 upload_result.map_err(Error::from).and_then(|upload1_result| {
 583                     Ok(upload1_result.and(result))
 584                 })
 585             })
 586             .flatten()
 587             .and_then(move |_| {
 588                 let repeat = repeat2.load(Ordering::SeqCst);
 589                 let stream_len = stream_len2.load(Ordering::SeqCst);
 590                 let speed = ((stream_len*1000000)/(1024*1024))/(start_time.elapsed().as_micros() as usize);
 591                 println!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat, start_time.elapsed().as_secs(), speed);
 592                 if repeat > 0 {
 593                     println!("Average chunk size was {} bytes.", stream_len/repeat);
 594                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
 595                 }
 596                 Ok(speed)
 597             })
 598     }
 599
 600     pub fn upload_speedtest(&self) -> impl Future<Item=usize, Error=Error> {
 601
 602         let mut data = vec![];
 603         // generate pseudo random byte sequence
 604         for i in 0..1024*1024 {
 605             for j in 0..4 {
 606                 let byte = ((i >> (j<<3))&0xff) as u8;
 607                 data.push(byte);
 608             }
 609         }
 610
 611         let item_len = data.len();
 612
 613         let repeat = std::sync::Arc::new(AtomicUsize::new(0));
 614         let repeat2 = repeat.clone();
 615
 616         let (upload_queue, upload_result) = Self::response_queue();
 617
 618         let start_time = std::time::Instant::now();
 619
 620         let h2 = self.h2.clone();
 621
 622         futures::stream::repeat(data)
 623             .take_while(move |_| {
 624                 repeat.fetch_add(1, Ordering::SeqCst);
 625                 Ok(start_time.elapsed().as_secs() < 5)
 626             })
 627             .for_each(move |data| {
 628                 let h2 = h2.clone();
 629
 630                 let upload_queue = upload_queue.clone();
 631
 632                 println!("send test data ({} bytes)", data.len());
 633                 let request = H2Client::request_builder("localhost", "POST", "speedtest", None).unwrap();
 634                 h2.send_request(request, Some(bytes::Bytes::from(data)))
 635                     .and_then(move |response| {
 636                         upload_queue.send(response)
 637                             .map(|_| ()).map_err(Error::from)
 638                     })
 639             })
 640             .then(move |result| {
 641                 println!("RESULT {:?}", result);
 642                 upload_result.map_err(Error::from).and_then(|upload1_result| {
 643                     Ok(upload1_result.and(result))
 644                 })
 645             })
 646             .flatten()
 647             .and_then(move |_| {
 648                 let repeat = repeat2.load(Ordering::SeqCst);
 649                 println!("Uploaded {} chunks in {} seconds.", repeat, start_time.elapsed().as_secs());
 650                 let speed = ((item_len*1000000*(repeat as usize))/(1024*1024))/(start_time.elapsed().as_micros() as usize);
 651                 if repeat > 0 {
 652                     println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128));
 653                 }
 654                 Ok(speed)
 655             })
 656     }
 657 }
 658
 659 #[derive(Clone)]
 660 pub struct H2Client {
 661     h2: h2::client::SendRequest<bytes::Bytes>,
 662 }
 663
 664 impl H2Client {
 665
 666     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
 667         Self { h2 }
 668     }
 669
 670     pub fn get(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 671         let req = Self::request_builder("localhost", "GET", path, param).unwrap();
 672         self.request(req)
 673     }
 674
 675     pub fn put(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 676         let req = Self::request_builder("localhost", "PUT", path, param).unwrap();
 677         self.request(req)
 678     }
 679
 680     pub fn post(&self, path: &str, param: Option<Value>) -> impl Future<Item=Value, Error=Error> {
 681         let req = Self::request_builder("localhost", "POST", path, param).unwrap();
 682         self.request(req)
 683     }
 684
 685     pub fn upload(&self, path: &str, param: Option<Value>, data: Vec<u8>) -> impl Future<Item=Value, Error=Error> {
 686         let request = Self::request_builder("localhost", "POST", path, param).unwrap();
 687
 688
 689         self.h2.clone()
 690             .ready()
 691             .map_err(Error::from)
 692             .and_then(move |mut send_request| {
 693                 let (response, stream) = send_request.send_request(request, false).unwrap();
 694                 PipeToSendStream::new(bytes::Bytes::from(data), stream)
 695                     .and_then(|_| {
 696                         response
 697                             .map_err(Error::from)
 698                             .and_then(Self::h2api_response)
 699                     })
 700             })
 701     }
 702
 703     fn request(
 704         &self,
 705         request: Request<()>,
 706     ) -> impl Future<Item=Value, Error=Error> {
 707
 708         self.send_request(request, None)
 709             .and_then(move |response| {
 710                 response
 711                     .map_err(Error::from)
 712                     .and_then(Self::h2api_response)
 713             })
 714     }
 715
 716     fn send_request(
 717         &self,
 718         request: Request<()>,
 719         data: Option<bytes::Bytes>,
 720     ) -> impl Future<Item=h2::client::ResponseFuture, Error=Error> {
 721
 722         self.h2.clone()
 723             .ready()
 724             .map_err(Error::from)
 725             .and_then(move |mut send_request| {
 726                 if let Some(data) = data {
 727                     let (response, stream) = send_request.send_request(request, false).unwrap();
 728                     future::Either::A(PipeToSendStream::new(bytes::Bytes::from(data), stream)
 729                         .and_then(move |_| {
 730                             future::ok(response)
 731                         }))
 732                 } else {
 733                     let (response, _stream) = send_request.send_request(request, true).unwrap();
 734                     future::Either::B(future::ok(response))
 735                 }
 736             })
 737     }
 738
 739     fn h2api_response(response: Response<h2::RecvStream>) -> impl Future<Item=Value, Error=Error> {
 740
 741         let status = response.status();
 742
 743         let (_head, mut body) = response.into_parts();
 744
 745         // The `release_capacity` handle allows the caller to manage
 746         // flow control.
 747         //
 748         // Whenever data is received, the caller is responsible for
 749         // releasing capacity back to the server once it has freed
 750         // the data from memory.
 751         let mut release_capacity = body.release_capacity().clone();
 752
 753         body
 754             .map(move |chunk| {
 755                 // Let the server send more data.
 756                 let _ = release_capacity.release_capacity(chunk.len());
 757                 chunk
 758             })
 759             .concat2()
 760             .map_err(Error::from)
 761             .and_then(move |data| {
 762                 let text = String::from_utf8(data.to_vec()).unwrap();
 763                 if status.is_success() {
 764                     if text.len() > 0 {
 765                         let mut value: Value = serde_json::from_str(&text)?;
 766                         if let Some(map) = value.as_object_mut() {
 767                             if let Some(data) = map.remove("data") {
 768                                 return Ok(data);
 769                             }
 770                         }
 771                         bail!("got result without data property");
 772                     } else {
 773                         Ok(Value::Null)
 774                     }
 775                 } else {
 776                     bail!("HTTP Error {}: {}", status, text);
 777                 }
 778             })
 779     }
 780
 781     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<()>, Error> {
 782         let path = path.trim_matches('/');
 783         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 784
 785         if let Some(data) = data {
 786             let query = tools::json_object_to_query(data)?;
 787             let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 788             let request = Request::builder()
 789                 .method(method)
 790                 .uri(url)
 791                 .header("User-Agent", "proxmox-backup-client/1.0")
 792                 .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 793                 .body(())?;
 794             return Ok(request);
 795         }
 796
 797         let request = Request::builder()
 798             .method(method)
 799             .uri(url)
 800             .header("User-Agent", "proxmox-backup-client/1.0")
 801             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 802             .body(())?;
 803
 804         Ok(request)
 805     }
 806 }