src/client/http_client.rs

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