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