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