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