src/client/http_client.rs

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