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