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