src/client/http_client.rs

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