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