src/client/http_client.rs

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