src/client/http_client.rs

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