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