src/client/http_client.rs

   1 use std::io::Write;
   2 use std::task::{Context, Poll};
   3 use std::sync::{Arc, Mutex};
   4
   5 use chrono::Utc;
   6 use failure::*;
   7 use futures::*;
   8 use http::Uri;
   9 use http::header::HeaderValue;
  10 use http::{Request, Response};
  11 use hyper::Body;
  12 use hyper::client::{Client, HttpConnector};
  13 use openssl::{ssl::{SslConnector, SslMethod}, x509::X509StoreContextRef};
  14 use serde_json::{json, Value};
  15 use percent_encoding::percent_encode;
  16 use xdg::BaseDirectories;
  17
  18 use proxmox::tools::{
  19     fs::{file_get_json, replace_file, CreateOptions},
  20 };
  21
  22 use super::pipe_to_stream::PipeToSendStream;
  23 use crate::tools::async_io::EitherStream;
  24 use crate::tools::{self, tty, BroadcastFuture, DEFAULT_ENCODE_SET};
  25
  26 #[derive(Clone)]
  27 pub struct AuthInfo {
  28     pub username: String,
  29     pub ticket: String,
  30     pub token: String,
  31 }
  32
  33 pub struct HttpClientOptions {
  34     password: Option<String>,
  35     fingerprint: Option<String>,
  36     interactive: bool,
  37     ticket_cache: bool,
  38     fingerprint_cache: bool,
  39     verify_cert: bool,
  40 }
  41
  42 impl HttpClientOptions {
  43
  44     pub fn new() -> Self {
  45         Self {
  46             password: None,
  47             fingerprint: None,
  48             interactive: false,
  49             ticket_cache: false,
  50             fingerprint_cache: false,
  51             verify_cert: true,
  52         }
  53     }
  54
  55     pub fn password(mut self, password: Option<String>) -> Self {
  56         self.password = password;
  57         self
  58     }
  59
  60     pub fn fingerprint(mut self, fingerprint: Option<String>) -> Self {
  61         self.fingerprint = fingerprint;
  62         self
  63     }
  64
  65     pub fn interactive(mut self, interactive: bool) -> Self {
  66         self.interactive = interactive;
  67         self
  68     }
  69
  70     pub fn ticket_cache(mut self, ticket_cache: bool) -> Self {
  71         self.ticket_cache = ticket_cache;
  72         self
  73     }
  74
  75     pub fn fingerprint_cache(mut self, fingerprint_cache: bool) -> Self {
  76         self.fingerprint_cache = fingerprint_cache;
  77         self
  78     }
  79
  80     pub fn verify_cert(mut self, verify_cert: bool) -> Self {
  81         self.verify_cert = verify_cert;
  82         self
  83     }
  84 }
  85
  86 /// HTTP(S) API client
  87 pub struct HttpClient {
  88     client: Client<HttpsConnector>,
  89     server: String,
  90     fingerprint: Arc<Mutex<Option<String>>>,
  91     auth: BroadcastFuture<AuthInfo>,
  92     _options: HttpClientOptions,
  93 }
  94
  95 /// Delete stored ticket data (logout)
  96 pub fn delete_ticket_info(server: &str, username: &str) -> Result<(), Error> {
  97
  98     let base = BaseDirectories::with_prefix("proxmox-backup")?;
  99
 100     // usually /run/user/<uid>/...
 101     let path = base.place_runtime_file("tickets")?;
 102
 103     let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600);
 104
 105     let mut data = file_get_json(&path, Some(json!({})))?;
 106
 107     if let Some(map) = data[server].as_object_mut() {
 108         map.remove(username);
 109     }
 110
 111     replace_file(path, data.to_string().as_bytes(), CreateOptions::new().perm(mode))?;
 112
 113     Ok(())
 114 }
 115
 116 fn store_fingerprint(server: &str, fingerprint: &str) -> Result<(), Error> {
 117
 118     let base = BaseDirectories::with_prefix("proxmox-backup")?;
 119
 120     // usually ~/.config/proxmox-backup/fingerprints
 121     let path = base.place_config_file("fingerprints")?;
 122
 123     let raw = match std::fs::read_to_string(&path) {
 124         Ok(v) => v,
 125         Err(err) => {
 126             if err.kind() == std::io::ErrorKind::NotFound {
 127                 String::new()
 128             } else {
 129                 bail!("unable to read fingerprints from {:?} - {}", path, err);
 130             }
 131         }
 132     };
 133
 134     let mut result = String::new();
 135
 136     raw.split('\n').for_each(|line| {
 137         let items: Vec<String> = line.split_whitespace().map(String::from).collect();
 138         if items.len() == 2 {
 139             if &items[0] == server {
 140                 // found, add later with new fingerprint
 141             } else {
 142                 result.push_str(line);
 143                 result.push('\n');
 144             }
 145         }
 146     });
 147
 148     result.push_str(server);
 149     result.push(' ');
 150     result.push_str(fingerprint);
 151     result.push('\n');
 152
 153     replace_file(path, result.as_bytes(), CreateOptions::new())?;
 154
 155     Ok(())
 156 }
 157
 158 fn load_fingerprint(server: &str) -> Option<String> {
 159
 160     let base = BaseDirectories::with_prefix("proxmox-backup").ok()?;
 161
 162     // usually ~/.config/proxmox-backup/fingerprints
 163     let path = base.place_config_file("fingerprints").ok()?;
 164
 165     let raw = std::fs::read_to_string(&path).ok()?;
 166
 167     for line in raw.split('\n') {
 168         let items: Vec<String> = line.split_whitespace().map(String::from).collect();
 169         if items.len() == 2 {
 170             if &items[0] == server {
 171                 return Some(items[1].clone());
 172             }
 173         }
 174     }
 175
 176     None
 177 }
 178
 179 fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
 180
 181     let base = BaseDirectories::with_prefix("proxmox-backup")?;
 182
 183     // usually /run/user/<uid>/...
 184     let path = base.place_runtime_file("tickets")?;
 185
 186     let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600);
 187
 188     let mut data = file_get_json(&path, Some(json!({})))?;
 189
 190     let now = Utc::now().timestamp();
 191
 192     data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token});
 193
 194     let mut new_data = json!({});
 195
 196     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
 197
 198     let empty = serde_json::map::Map::new();
 199     for (server, info) in data.as_object().unwrap_or(&empty) {
 200         for (_user, uinfo) in info.as_object().unwrap_or(&empty) {
 201             if let Some(timestamp) = uinfo["timestamp"].as_i64() {
 202                 let age = now - timestamp;
 203                 if age < ticket_lifetime {
 204                     new_data[server][username] = uinfo.clone();
 205                 }
 206             }
 207         }
 208     }
 209
 210     replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new().perm(mode))?;
 211
 212     Ok(())
 213 }
 214
 215 fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
 216     let base = BaseDirectories::with_prefix("proxmox-backup").ok()?;
 217
 218     // usually /run/user/<uid>/...
 219     let path = base.place_runtime_file("tickets").ok()?;
 220     let data = file_get_json(&path, None).ok()?;
 221     let now = Utc::now().timestamp();
 222     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
 223     let uinfo = data[server][username].as_object()?;
 224     let timestamp = uinfo["timestamp"].as_i64()?;
 225     let age = now - timestamp;
 226
 227     if age < ticket_lifetime {
 228         let ticket = uinfo["ticket"].as_str()?;
 229         let token = uinfo["token"].as_str()?;
 230         Some((ticket.to_owned(), token.to_owned()))
 231     } else {
 232         None
 233     }
 234 }
 235
 236 impl HttpClient {
 237
 238     pub fn new(server: &str, username: &str, mut options: HttpClientOptions) -> Result<Self, Error> {
 239
 240         let verified_fingerprint = Arc::new(Mutex::new(None));
 241
 242         let mut fingerprint = options.fingerprint.take();
 243         if options.fingerprint_cache && fingerprint.is_none() {
 244             fingerprint = load_fingerprint(server);
 245         }
 246
 247         let client = Self::build_client(
 248             server.to_string(),
 249             fingerprint,
 250             options.interactive,
 251             verified_fingerprint.clone(),
 252             options.verify_cert,
 253             options.fingerprint_cache,
 254         );
 255
 256         let password = options.password.take();
 257
 258         let password = if let Some(password) = password {
 259             password
 260         } else {
 261             let mut ticket_info = None;
 262             if options.ticket_cache {
 263                 ticket_info = load_ticket_info(server, username);
 264             }
 265             if let Some((ticket, _token)) = ticket_info {
 266                 ticket
 267             } else {
 268                 Self::get_password(&username, options.interactive)?
 269             }
 270         };
 271
 272         let login_future = Self::credentials(
 273             client.clone(),
 274             server.to_owned(),
 275             username.to_owned(),
 276             password,
 277             options.ticket_cache,
 278         );
 279
 280         Ok(Self {
 281             client,
 282             server: String::from(server),
 283             fingerprint: verified_fingerprint,
 284             auth: BroadcastFuture::new(Box::new(login_future)),
 285             _options: options,
 286         })
 287     }
 288
 289     /// Login
 290     ///
 291     /// Login is done on demand, so this is onyl required if you need
 292     /// access to authentication data in 'AuthInfo'.
 293     pub async fn login(&self) -> Result<AuthInfo, Error> {
 294         self.auth.listen().await
 295     }
 296
 297     /// Returns the optional fingerprint passed to the new() constructor.
 298     pub fn fingerprint(&self) -> Option<String> {
 299         (*self.fingerprint.lock().unwrap()).clone()
 300     }
 301
 302     fn get_password(_username: &str, interactive: bool) -> Result<String, Error> {
 303         use std::env::VarError::*;
 304         match std::env::var("PBS_PASSWORD") {
 305             Ok(p) => return Ok(p),
 306             Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"),
 307             Err(NotPresent) => {
 308                 // Try another method
 309             }
 310         }
 311
 312         // If we're on a TTY, query the user for a password
 313         if interactive && tty::stdin_isatty() {
 314             return Ok(String::from_utf8(tty::read_password("Password: ")?)?);
 315         }
 316
 317         bail!("no password input mechanism available");
 318     }
 319
 320     fn verify_callback(
 321         valid: bool, ctx:
 322         &mut X509StoreContextRef,
 323         server: String,
 324         expected_fingerprint: Option<String>,
 325         interactive: bool,
 326         verified_fingerprint: Arc<Mutex<Option<String>>>,
 327         fingerprint_cache: bool,
 328     ) -> bool {
 329         if valid { return true; }
 330
 331         let cert = match ctx.current_cert() {
 332             Some(cert) => cert,
 333             None => return false,
 334         };
 335
 336         let depth = ctx.error_depth();
 337         if depth != 0 { return false; }
 338
 339         let fp = match cert.digest(openssl::hash::MessageDigest::sha256()) {
 340             Ok(fp) => fp,
 341             Err(_) => return false, // should not happen
 342         };
 343         let fp_string = proxmox::tools::digest_to_hex(&fp);
 344         let fp_string = fp_string.as_bytes().chunks(2).map(|v| std::str::from_utf8(v).unwrap())
 345             .collect::<Vec<&str>>().join(":");
 346
 347         if let Some(expected_fingerprint) = expected_fingerprint {
 348             if expected_fingerprint == fp_string {
 349                 *verified_fingerprint.lock().unwrap() = Some(fp_string);
 350                 return true;
 351             } else {
 352                 return false;
 353             }
 354         }
 355
 356         // If we're on a TTY, query the user
 357         if interactive && tty::stdin_isatty() {
 358             println!("fingerprint: {}", fp_string);
 359             loop {
 360                 print!("Want to trust? (y/n): ");
 361                 let _ = std::io::stdout().flush();
 362                 let mut buf = [0u8; 1];
 363                 use std::io::Read;
 364                 match std::io::stdin().read_exact(&mut buf) {
 365                     Ok(()) => {
 366                         if buf[0] == b'y' || buf[0] == b'Y' {
 367                             if fingerprint_cache {
 368                                 if let Err(err) = store_fingerprint(&server, &fp_string) {
 369                                     eprintln!("{}", err);
 370                                 }
 371                             }
 372                             *verified_fingerprint.lock().unwrap() = Some(fp_string);
 373                            return true;
 374                         } else if buf[0] == b'n' || buf[0] == b'N' {
 375                             return false;
 376                         }
 377                     }
 378                     Err(_) => {
 379                         return false;
 380                     }
 381                 }
 382             }
 383         }
 384         false
 385     }
 386
 387     fn build_client(
 388         server: String,
 389         fingerprint: Option<String>,
 390         interactive: bool,
 391         verified_fingerprint: Arc<Mutex<Option<String>>>,
 392         verify_cert: bool,
 393         fingerprint_cache: bool,
 394     ) -> Client<HttpsConnector> {
 395
 396         let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap();
 397
 398         if verify_cert {
 399             ssl_connector_builder.set_verify_callback(openssl::ssl::SslVerifyMode::PEER, move |valid, ctx| {
 400                 Self::verify_callback(
 401                     valid, ctx, server.clone(), fingerprint.clone(), interactive,
 402                     verified_fingerprint.clone(), fingerprint_cache)
 403             });
 404         } else {
 405             ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE);
 406         }
 407
 408         let mut httpc = hyper::client::HttpConnector::new();
 409         httpc.set_nodelay(true); // important for h2 download performance!
 410         httpc.set_recv_buffer_size(Some(1024*1024)); //important for h2 download performance!
 411         httpc.enforce_http(false); // we want https...
 412
 413         let https = HttpsConnector::with_connector(httpc, ssl_connector_builder.build());
 414
 415         Client::builder()
 416         //.http2_initial_stream_window_size( (1 << 31) - 2)
 417         //.http2_initial_connection_window_size( (1 << 31) - 2)
 418             .build::<_, Body>(https)
 419     }
 420
 421     pub async fn request(&self, mut req: Request<Body>) -> Result<Value, Error> {
 422
 423         let client = self.client.clone();
 424
 425         let auth =  self.login().await?;
 426
 427         let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 428         req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 429         req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
 430
 431         Self::api_request(client, req).await
 432     }
 433
 434     pub async fn get(
 435         &self,
 436         path: &str,
 437         data: Option<Value>,
 438     ) -> Result<Value, Error> {
 439         let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
 440         self.request(req).await
 441     }
 442
 443     pub async fn delete(
 444         &mut self,
 445         path: &str,
 446         data: Option<Value>,
 447     ) -> Result<Value, Error> {
 448         let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
 449         self.request(req).await
 450     }
 451
 452     pub async fn post(
 453         &mut self,
 454         path: &str,
 455         data: Option<Value>,
 456     ) -> Result<Value, Error> {
 457         let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
 458         self.request(req).await
 459     }
 460
 461     pub async fn download(
 462         &mut self,
 463         path: &str,
 464         output: &mut (dyn Write + Send),
 465     ) ->  Result<(), Error> {
 466         let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
 467
 468         let client = self.client.clone();
 469
 470         let auth = self.login().await?;
 471
 472         let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 473         req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 474
 475         let resp = client.request(req).await?;
 476         let status = resp.status();
 477         if !status.is_success() {
 478             HttpClient::api_response(resp)
 479                 .map(|_| Err(format_err!("unknown error")))
 480                 .await?
 481         } else {
 482             resp.into_body()
 483                 .map_err(Error::from)
 484                 .try_fold(output, move |acc, chunk| async move {
 485                     acc.write_all(&chunk)?;
 486                     Ok::<_, Error>(acc)
 487                 })
 488                 .await?;
 489         }
 490         Ok(())
 491     }
 492
 493     pub async fn upload(
 494         &mut self,
 495         content_type: &str,
 496         body: Body,
 497         path: &str,
 498         data: Option<Value>,
 499     ) -> Result<Value, Error> {
 500
 501         let path = path.trim_matches('/');
 502         let mut url = format!("https://{}:8007/{}", &self.server, path);
 503
 504         if let Some(data) = data {
 505             let query = tools::json_object_to_query(data).unwrap();
 506             url.push('?');
 507             url.push_str(&query);
 508         }
 509
 510         let url: Uri = url.parse().unwrap();
 511
 512         let req = Request::builder()
 513             .method("POST")
 514             .uri(url)
 515             .header("User-Agent", "proxmox-backup-client/1.0")
 516             .header("Content-Type", content_type)
 517             .body(body).unwrap();
 518
 519         self.request(req).await
 520     }
 521
 522     pub async fn start_h2_connection(
 523         &self,
 524         mut req: Request<Body>,
 525         protocol_name: String,
 526     ) -> Result<(H2Client, futures::future::AbortHandle), Error> {
 527
 528         let auth = self.login().await?;
 529         let client = self.client.clone();
 530
 531         let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
 532         req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 533         req.headers_mut().insert("UPGRADE", HeaderValue::from_str(&protocol_name).unwrap());
 534
 535         let resp = client.request(req).await?;
 536         let status = resp.status();
 537
 538         if status != http::StatusCode::SWITCHING_PROTOCOLS {
 539             Self::api_response(resp).await?;
 540             bail!("unknown error");
 541         }
 542
 543         let upgraded = resp
 544             .into_body()
 545             .on_upgrade()
 546             .await?;
 547
 548         let max_window_size = (1 << 31) - 2;
 549
 550         let (h2, connection) = h2::client::Builder::new()
 551             .initial_connection_window_size(max_window_size)
 552             .initial_window_size(max_window_size)
 553             .max_frame_size(4*1024*1024)
 554             .handshake(upgraded)
 555             .await?;
 556
 557         let connection = connection
 558             .map_err(|_| panic!("HTTP/2.0 connection failed"));
 559
 560         let (connection, abort) = futures::future::abortable(connection);
 561         // A cancellable future returns an Option which is None when cancelled and
 562         // Some when it finished instead, since we don't care about the return type we
 563         // need to map it away:
 564         let connection = connection.map(|_| ());
 565
 566         // Spawn a new task to drive the connection state
 567         tokio::spawn(connection);
 568
 569         // Wait until the `SendRequest` handle has available capacity.
 570         let c = h2.ready().await?;
 571         Ok((H2Client::new(c), abort))
 572     }
 573
 574     async fn credentials(
 575         client: Client<HttpsConnector>,
 576         server: String,
 577         username: String,
 578         password: String,
 579         use_ticket_cache: bool,
 580     ) -> Result<AuthInfo, Error> {
 581         let data = json!({ "username": username, "password": password });
 582         let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
 583         let cred = Self::api_request(client, req).await?;
 584         let auth = AuthInfo {
 585             username: cred["data"]["username"].as_str().unwrap().to_owned(),
 586             ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(),
 587             token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
 588         };
 589
 590         if use_ticket_cache {
 591             let _ = store_ticket_info(&server, &auth.username, &auth.ticket, &auth.token);
 592         }
 593
 594         Ok(auth)
 595     }
 596
 597     async fn api_response(response: Response<Body>) -> Result<Value, Error> {
 598         let status = response.status();
 599         let data = hyper::body::to_bytes(response.into_body()).await?;
 600
 601         let text = String::from_utf8(data.to_vec()).unwrap();
 602         if status.is_success() {
 603             if text.is_empty() {
 604                 Ok(Value::Null)
 605             } else {
 606                 let value: Value = serde_json::from_str(&text)?;
 607                 Ok(value)
 608             }
 609         } else {
 610             bail!("HTTP Error {}: {}", status, text);
 611         }
 612     }
 613
 614     async fn api_request(
 615         client: Client<HttpsConnector>,
 616         req: Request<Body>
 617     ) -> Result<Value, Error> {
 618
 619         client.request(req)
 620             .map_err(Error::from)
 621             .and_then(Self::api_response)
 622             .await
 623     }
 624
 625     // Read-only access to server property
 626     pub fn server(&self) -> &str {
 627         &self.server
 628     }
 629
 630     pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
 631         let path = path.trim_matches('/');
 632         let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 633
 634         if let Some(data) = data {
 635             if method == "POST" {
 636                 let request = Request::builder()
 637                     .method(method)
 638                     .uri(url)
 639                     .header("User-Agent", "proxmox-backup-client/1.0")
 640                     .header(hyper::header::CONTENT_TYPE, "application/json")
 641                     .body(Body::from(data.to_string()))?;
 642                 return Ok(request);
 643             } else {
 644                 let query = tools::json_object_to_query(data)?;
 645                 let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 646                 let request = Request::builder()
 647                     .method(method)
 648                     .uri(url)
 649                     .header("User-Agent", "proxmox-backup-client/1.0")
 650                     .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 651                     .body(Body::empty())?;
 652                 return Ok(request);
 653             }
 654         }
 655
 656         let request = Request::builder()
 657             .method(method)
 658             .uri(url)
 659             .header("User-Agent", "proxmox-backup-client/1.0")
 660             .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
 661             .body(Body::empty())?;
 662
 663         Ok(request)
 664     }
 665 }
 666
 667
 668 #[derive(Clone)]
 669 pub struct H2Client {
 670     h2: h2::client::SendRequest<bytes::Bytes>,
 671 }
 672
 673 impl H2Client {
 674
 675     pub fn new(h2: h2::client::SendRequest<bytes::Bytes>) -> Self {
 676         Self { h2 }
 677     }
 678
 679     pub async fn get(
 680         &self,
 681         path: &str,
 682         param: Option<Value>
 683     ) -> Result<Value, Error> {
 684         let req = Self::request_builder("localhost", "GET", path, param, None).unwrap();
 685         self.request(req).await
 686     }
 687
 688     pub async fn put(
 689         &self,
 690         path: &str,
 691         param: Option<Value>
 692     ) -> Result<Value, Error> {
 693         let req = Self::request_builder("localhost", "PUT", path, param, None).unwrap();
 694         self.request(req).await
 695     }
 696
 697     pub async fn post(
 698         &self,
 699         path: &str,
 700         param: Option<Value>
 701     ) -> Result<Value, Error> {
 702         let req = Self::request_builder("localhost", "POST", path, param, None).unwrap();
 703         self.request(req).await
 704     }
 705
 706     pub async fn download<W: Write + Send>(
 707         &self,
 708         path: &str,
 709         param: Option<Value>,
 710         mut output: W,
 711     ) -> Result<W, Error> {
 712         let request = Self::request_builder("localhost", "GET", path, param, None).unwrap();
 713
 714         let response_future = self.send_request(request, None).await?;
 715
 716         let resp = response_future.await?;
 717
 718         let status = resp.status();
 719         if !status.is_success() {
 720             H2Client::h2api_response(resp).await?; // raise error
 721             unreachable!();
 722         }
 723
 724         let mut body = resp.into_body();
 725         while let Some(chunk) = body.data().await {
 726             let chunk = chunk?;
 727             body.flow_control().release_capacity(chunk.len())?;
 728             output.write_all(&chunk)?;
 729         }
 730
 731         Ok(output)
 732     }
 733
 734     pub async fn upload(
 735         &self,
 736         method: &str, // POST or PUT
 737         path: &str,
 738         param: Option<Value>,
 739         content_type: &str,
 740         data: Vec<u8>,
 741     ) -> Result<Value, Error> {
 742         let request = Self::request_builder("localhost", method, path, param, Some(content_type)).unwrap();
 743
 744         let mut send_request = self.h2.clone().ready().await?;
 745
 746         let (response, stream) = send_request.send_request(request, false).unwrap();
 747
 748         PipeToSendStream::new(bytes::Bytes::from(data), stream).await?;
 749
 750         response
 751             .map_err(Error::from)
 752             .and_then(Self::h2api_response)
 753             .await
 754     }
 755
 756     async fn request(
 757         &self,
 758         request: Request<()>,
 759     ) -> Result<Value, Error> {
 760
 761         self.send_request(request, None)
 762             .and_then(move |response| {
 763                 response
 764                     .map_err(Error::from)
 765                     .and_then(Self::h2api_response)
 766             })
 767             .await
 768     }
 769
 770     pub fn send_request(
 771         &self,
 772         request: Request<()>,
 773         data: Option<bytes::Bytes>,
 774     ) -> impl Future<Output = Result<h2::client::ResponseFuture, Error>> {
 775
 776         self.h2.clone()
 777             .ready()
 778             .map_err(Error::from)
 779             .and_then(move |mut send_request| async move {
 780                 if let Some(data) = data {
 781                     let (response, stream) = send_request.send_request(request, false).unwrap();
 782                     PipeToSendStream::new(data, stream).await?;
 783                     Ok(response)
 784                 } else {
 785                     let (response, _stream) = send_request.send_request(request, true).unwrap();
 786                     Ok(response)
 787                 }
 788             })
 789     }
 790
 791     pub async fn h2api_response(
 792         response: Response<h2::RecvStream>,
 793     ) -> Result<Value, Error> {
 794         let status = response.status();
 795
 796         let (_head, mut body) = response.into_parts();
 797
 798         let mut data = Vec::new();
 799         while let Some(chunk) = body.data().await {
 800             let chunk = chunk?;
 801             // Whenever data is received, the caller is responsible for
 802             // releasing capacity back to the server once it has freed
 803             // the data from memory.
 804             // Let the server send more data.
 805             body.flow_control().release_capacity(chunk.len())?;
 806             data.extend(chunk);
 807         }
 808
 809         let text = String::from_utf8(data.to_vec()).unwrap();
 810         if status.is_success() {
 811             if text.is_empty() {
 812                 Ok(Value::Null)
 813             } else {
 814                 let mut value: Value = serde_json::from_str(&text)?;
 815                 if let Some(map) = value.as_object_mut() {
 816                     if let Some(data) = map.remove("data") {
 817                         return Ok(data);
 818                     }
 819                 }
 820                 bail!("got result without data property");
 821             }
 822         } else {
 823             bail!("HTTP Error {}: {}", status, text);
 824         }
 825     }
 826
 827     // Note: We always encode parameters with the url
 828     pub fn request_builder(
 829         server: &str,
 830         method: &str,
 831         path: &str,
 832         param: Option<Value>,
 833         content_type: Option<&str>,
 834     ) -> Result<Request<()>, Error> {
 835         let path = path.trim_matches('/');
 836
 837         let content_type = content_type.unwrap_or("application/x-www-form-urlencoded");
 838
 839         if let Some(param) = param {
 840             let query = tools::json_object_to_query(param)?;
 841             // We detected problem with hyper around 6000 characters - seo we try to keep on the safe side
 842             if query.len() > 4096 { bail!("h2 query data too large ({} bytes) - please encode data inside body", query.len()); }
 843             let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
 844              let request = Request::builder()
 845                 .method(method)
 846                 .uri(url)
 847                 .header("User-Agent", "proxmox-backup-client/1.0")
 848                 .header(hyper::header::CONTENT_TYPE, content_type)
 849                 .body(())?;
 850             Ok(request)
 851         } else {
 852             let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
 853             let request = Request::builder()
 854                 .method(method)
 855                 .uri(url)
 856                 .header("User-Agent", "proxmox-backup-client/1.0")
 857                 .header(hyper::header::CONTENT_TYPE, content_type)
 858                 .body(())?;
 859
 860             Ok(request)
 861         }
 862     }
 863 }
 864
 865 #[derive(Clone)]
 866 pub struct HttpsConnector {
 867     http: HttpConnector,
 868     ssl_connector: std::sync::Arc<SslConnector>,
 869 }
 870
 871 impl HttpsConnector {
 872     pub fn with_connector(mut http: HttpConnector, ssl_connector: SslConnector) -> Self {
 873         http.enforce_http(false);
 874
 875         Self {
 876             http,
 877             ssl_connector: std::sync::Arc::new(ssl_connector),
 878         }
 879     }
 880 }
 881
 882 type MaybeTlsStream = EitherStream<
 883     tokio::net::TcpStream,
 884     tokio_openssl::SslStream<tokio::net::TcpStream>,
 885 >;
 886
 887 impl hyper::service::Service<Uri> for HttpsConnector {
 888     type Response = MaybeTlsStream;
 889     type Error = Error;
 890     type Future = std::pin::Pin<Box<
 891         dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static
 892     >>;
 893
 894     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
 895         // This connector is always ready, but others might not be.
 896         Poll::Ready(Ok(()))
 897     }
 898
 899     fn call(&mut self, dst: Uri) -> Self::Future {
 900         let mut this = self.clone();
 901         async move {
 902             let is_https = dst
 903                 .scheme()
 904                 .ok_or_else(|| format_err!("missing URL scheme"))?
 905                 == "https";
 906             let host = dst
 907                 .host()
 908                 .ok_or_else(|| format_err!("missing hostname in destination url?"))?
 909                 .to_string();
 910
 911             let config = this.ssl_connector.configure();
 912             let conn = this.http.call(dst).await?;
 913             if is_https {
 914                 let conn = tokio_openssl::connect(config?, &host, conn).await?;
 915                 Ok(MaybeTlsStream::Right(conn))
 916             } else {
 917                 Ok(MaybeTlsStream::Left(conn))
 918             }
 919         }.boxed()
 920     }
 921 }