X-Git-Url: https://git.proxmox.com/?a=blobdiff_plain;f=src%2Fclient%2Fhttp_client.rs;h=76ab039158510829626b2508c9f35dae6e5dbc6f;hb=a941bbd0c9dfd2dd804780658f17c4f6e154efe2;hp=db7cea3d162d3ab0487317c48832bdeb6d64f4fe;hpb=4f6aaf542c202312db73914ada5c2faab66930aa;p=proxmox-backup.git diff --git a/src/client/http_client.rs b/src/client/http_client.rs index db7cea3d..76ab0391 100644 --- a/src/client/http_client.rs +++ b/src/client/http_client.rs @@ -1,59 +1,233 @@ -use failure::*; +use std::io::Write; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; +use anyhow::{bail, format_err, Error}; +use futures::*; use http::Uri; +use http::header::HeaderValue; +use http::{Request, Response}; use hyper::Body; -use hyper::client::Client; +use hyper::client::{Client, HttpConnector}; +use openssl::{ssl::{SslConnector, SslMethod}, x509::X509StoreContextRef}; +use serde_json::{json, Value}; +use percent_encoding::percent_encode; use xdg::BaseDirectories; -use chrono::{DateTime, Local, Utc}; -use std::collections::HashSet; -use std::sync::{Arc, Mutex}; -use std::io::Write; -use http::{Request, Response}; -use http::header::HeaderValue; +use proxmox::{ + api::error::HttpError, + sys::linux::tty, + tools::fs::{file_get_json, replace_file, CreateOptions}, +}; + +use super::pipe_to_stream::PipeToSendStream; +use crate::api2::types::{Authid, Userid}; +use crate::tools::{ + self, + BroadcastFuture, + DEFAULT_ENCODE_SET, + http::HttpsConnector, +}; + +/// Timeout used for several HTTP operations that are expected to finish quickly but may block in +/// certain error conditions. Keep it generous, to avoid false-positive under high load. +const HTTP_TIMEOUT: Duration = Duration::from_secs(2 * 60); -use futures::*; -use futures::stream::Stream; -use std::sync::atomic::{AtomicUsize, Ordering}; -use tokio::sync::mpsc; -use openssl::ssl::{SslConnector, SslMethod}; +#[derive(Clone)] +pub struct AuthInfo { + pub auth_id: Authid, + pub ticket: String, + pub token: String, +} -use serde_json::{json, Value}; -use url::percent_encoding::{percent_encode, DEFAULT_ENCODE_SET}; +pub struct HttpClientOptions { + prefix: Option, + password: Option, + fingerprint: Option, + interactive: bool, + ticket_cache: bool, + fingerprint_cache: bool, + verify_cert: bool, +} -use crate::tools::{self, BroadcastFuture, tty}; -use crate::tools::futures::{cancellable, Canceller}; -use super::pipe_to_stream::*; -use super::merge_known_chunks::*; +impl HttpClientOptions { + + pub fn new_interactive(password: Option, fingerprint: Option) -> Self { + Self { + password, + fingerprint, + fingerprint_cache: true, + ticket_cache: true, + interactive: true, + prefix: Some("proxmox-backup".to_string()), + ..Self::default() + } + } -use crate::backup::*; + pub fn new_non_interactive(password: String, fingerprint: Option) -> Self { + Self { + password: Some(password), + fingerprint, + ..Self::default() + } + } -#[derive(Clone)] -struct AuthInfo { - username: String, - ticket: String, - token: String, + pub fn prefix(mut self, prefix: Option) -> Self { + self.prefix = prefix; + self + } + + pub fn password(mut self, password: Option) -> Self { + self.password = password; + self + } + + pub fn fingerprint(mut self, fingerprint: Option) -> Self { + self.fingerprint = fingerprint; + self + } + + pub fn interactive(mut self, interactive: bool) -> Self { + self.interactive = interactive; + self + } + + pub fn ticket_cache(mut self, ticket_cache: bool) -> Self { + self.ticket_cache = ticket_cache; + self + } + + pub fn fingerprint_cache(mut self, fingerprint_cache: bool) -> Self { + self.fingerprint_cache = fingerprint_cache; + self + } + + pub fn verify_cert(mut self, verify_cert: bool) -> Self { + self.verify_cert = verify_cert; + self + } +} + +impl Default for HttpClientOptions { + fn default() -> Self { + Self { + prefix: None, + password: None, + fingerprint: None, + interactive: false, + ticket_cache: false, + fingerprint_cache: false, + verify_cert: true, + } + } } /// HTTP(S) API client pub struct HttpClient { - client: Client>, + client: Client, server: String, - auth: BroadcastFuture, + port: u16, + fingerprint: Arc>>, + first_auth: Option>, + auth: Arc>, + ticket_abort: futures::future::AbortHandle, + _options: HttpClientOptions, } -fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> { +/// Delete stored ticket data (logout) +pub fn delete_ticket_info(prefix: &str, server: &str, username: &Userid) -> Result<(), Error> { - let base = BaseDirectories::with_prefix("proxmox-backup")?; + let base = BaseDirectories::with_prefix(prefix)?; // usually /run/user//... let path = base.place_runtime_file("tickets")?; let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600); - let mut data = tools::file_get_json(&path, Some(json!({})))?; + let mut data = file_get_json(&path, Some(json!({})))?; + + if let Some(map) = data[server].as_object_mut() { + map.remove(username.as_str()); + } - let now = Utc::now().timestamp(); + replace_file(path, data.to_string().as_bytes(), CreateOptions::new().perm(mode))?; + + Ok(()) +} + +fn store_fingerprint(prefix: &str, server: &str, fingerprint: &str) -> Result<(), Error> { + + let base = BaseDirectories::with_prefix(prefix)?; + + // usually ~/.config//fingerprints + let path = base.place_config_file("fingerprints")?; + + let raw = match std::fs::read_to_string(&path) { + Ok(v) => v, + Err(err) => { + if err.kind() == std::io::ErrorKind::NotFound { + String::new() + } else { + bail!("unable to read fingerprints from {:?} - {}", path, err); + } + } + }; + + let mut result = String::new(); + + raw.split('\n').for_each(|line| { + let items: Vec = line.split_whitespace().map(String::from).collect(); + if items.len() == 2 { + if items[0] == server { + // found, add later with new fingerprint + } else { + result.push_str(line); + result.push('\n'); + } + } + }); + + result.push_str(server); + result.push(' '); + result.push_str(fingerprint); + result.push('\n'); + + replace_file(path, result.as_bytes(), CreateOptions::new())?; + + Ok(()) +} + +fn load_fingerprint(prefix: &str, server: &str) -> Option { + + let base = BaseDirectories::with_prefix(prefix).ok()?; + + // usually ~/.config//fingerprints + let path = base.place_config_file("fingerprints").ok()?; + + let raw = std::fs::read_to_string(&path).ok()?; + + for line in raw.split('\n') { + let items: Vec = line.split_whitespace().map(String::from).collect(); + if items.len() == 2 && items[0] == server { + return Some(items[1].clone()); + } + } + + None +} + +fn store_ticket_info(prefix: &str, server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> { + + let base = BaseDirectories::with_prefix(prefix)?; + + // usually /run/user//... + let path = base.place_runtime_file("tickets")?; + + let mode = nix::sys::stat::Mode::from_bits_truncate(0o0600); + + let mut data = file_get_json(&path, Some(json!({})))?; + + let now = proxmox::tools::time::epoch_i64(); data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token}); @@ -63,195 +237,393 @@ fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> let empty = serde_json::map::Map::new(); for (server, info) in data.as_object().unwrap_or(&empty) { - for (_user, uinfo) in info.as_object().unwrap_or(&empty) { + for (user, uinfo) in info.as_object().unwrap_or(&empty) { if let Some(timestamp) = uinfo["timestamp"].as_i64() { let age = now - timestamp; if age < ticket_lifetime { - new_data[server][username] = uinfo.clone(); + new_data[server][user] = uinfo.clone(); } } } } - tools::file_set_contents(path, new_data.to_string().as_bytes(), Some(mode))?; + replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new().perm(mode))?; Ok(()) } -fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> { - let base = match BaseDirectories::with_prefix("proxmox-backup") { - Ok(b) => b, - _ => return None, - }; +fn load_ticket_info(prefix: &str, server: &str, userid: &Userid) -> Option<(String, String)> { + let base = BaseDirectories::with_prefix(prefix).ok()?; // usually /run/user//... - let path = match base.place_runtime_file("tickets") { - Ok(p) => p, - _ => return None, - }; + let path = base.place_runtime_file("tickets").ok()?; + let data = file_get_json(&path, None).ok()?; + let now = proxmox::tools::time::epoch_i64(); + let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60; + let uinfo = data[server][userid.as_str()].as_object()?; + let timestamp = uinfo["timestamp"].as_i64()?; + let age = now - timestamp; - let data = match tools::file_get_json(&path, None) { - Ok(v) => v, - _ => return None, - }; + if age < ticket_lifetime { + let ticket = uinfo["ticket"].as_str()?; + let token = uinfo["token"].as_str()?; + Some((ticket.to_owned(), token.to_owned())) + } else { + None + } +} - let now = Utc::now().timestamp(); +impl HttpClient { + pub fn new( + server: &str, + port: u16, + auth_id: &Authid, + mut options: HttpClientOptions, + ) -> Result { + + let verified_fingerprint = Arc::new(Mutex::new(None)); + + let mut fingerprint = options.fingerprint.take(); + + if fingerprint.is_some() { + // do not store fingerprints passed via options in cache + options.fingerprint_cache = false; + } else if options.fingerprint_cache && options.prefix.is_some() { + fingerprint = load_fingerprint(options.prefix.as_ref().unwrap(), server); + } - let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60; + let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap(); - if let Some(uinfo) = data[server][username].as_object() { - if let Some(timestamp) = uinfo["timestamp"].as_i64() { - let age = now - timestamp; - if age < ticket_lifetime { - let ticket = match uinfo["ticket"].as_str() { - Some(t) => t, - None => return None, - }; - let token = match uinfo["token"].as_str() { - Some(t) => t, - None => return None, - }; - return Some((ticket.to_owned(), token.to_owned())); - } + if options.verify_cert { + let server = server.to_string(); + let verified_fingerprint = verified_fingerprint.clone(); + let interactive = options.interactive; + let fingerprint_cache = options.fingerprint_cache; + let prefix = options.prefix.clone(); + ssl_connector_builder.set_verify_callback(openssl::ssl::SslVerifyMode::PEER, move |valid, ctx| { + let (valid, fingerprint) = Self::verify_callback(valid, ctx, fingerprint.clone(), interactive); + if valid { + if let Some(fingerprint) = fingerprint { + if fingerprint_cache && prefix.is_some() { + if let Err(err) = store_fingerprint( + prefix.as_ref().unwrap(), &server, &fingerprint) { + eprintln!("{}", err); + } + } + *verified_fingerprint.lock().unwrap() = Some(fingerprint); + } + } + valid + }); + } else { + ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE); } - } - None -} + let mut httpc = HttpConnector::new(); + httpc.set_nodelay(true); // important for h2 download performance! + httpc.enforce_http(false); // we want https... -impl HttpClient { + httpc.set_connect_timeout(Some(std::time::Duration::new(10, 0))); + let https = HttpsConnector::with_connector(httpc, ssl_connector_builder.build()); - pub fn new(server: &str, username: &str) -> Result { - let client = Self::build_client(); + let client = Client::builder() + //.http2_initial_stream_window_size( (1 << 31) - 2) + //.http2_initial_connection_window_size( (1 << 31) - 2) + .build::<_, Body>(https); + + let password = options.password.take(); + let use_ticket_cache = options.ticket_cache && options.prefix.is_some(); - let password = if let Some((ticket, _token)) = load_ticket_info(server, username) { - ticket + let password = if let Some(password) = password { + password } else { - Self::get_password(&username)? + let userid = if auth_id.is_token() { + bail!("API token secret must be provided!"); + } else { + auth_id.user() + }; + let mut ticket_info = None; + if use_ticket_cache { + ticket_info = load_ticket_info(options.prefix.as_ref().unwrap(), server, userid); + } + if let Some((ticket, _token)) = ticket_info { + ticket + } else { + Self::get_password(userid, options.interactive)? + } }; - let login = Self::credentials(client.clone(), server.to_owned(), username.to_owned(), password); + let auth = Arc::new(RwLock::new(AuthInfo { + auth_id: auth_id.clone(), + ticket: password.clone(), + token: "".to_string(), + })); + + let server2 = server.to_string(); + let client2 = client.clone(); + let auth2 = auth.clone(); + let prefix2 = options.prefix.clone(); + + let renewal_future = async move { + loop { + tokio::time::sleep(Duration::new(60*15, 0)).await; // 15 minutes + let (auth_id, ticket) = { + let authinfo = auth2.read().unwrap().clone(); + (authinfo.auth_id, authinfo.ticket) + }; + match Self::credentials(client2.clone(), server2.clone(), port, auth_id.user().clone(), ticket).await { + Ok(auth) => { + if use_ticket_cache && prefix2.is_some() { + let _ = store_ticket_info(prefix2.as_ref().unwrap(), &server2, &auth.auth_id.to_string(), &auth.ticket, &auth.token); + } + *auth2.write().unwrap() = auth; + }, + Err(err) => { + eprintln!("re-authentication failed: {}", err); + return; + } + } + } + }; + + let (renewal_future, ticket_abort) = futures::future::abortable(renewal_future); + + let login_future = Self::credentials( + client.clone(), + server.to_owned(), + port, + auth_id.user().clone(), + password, + ).map_ok({ + let server = server.to_string(); + let prefix = options.prefix.clone(); + let authinfo = auth.clone(); + + move |auth| { + if use_ticket_cache && prefix.is_some() { + let _ = store_ticket_info(prefix.as_ref().unwrap(), &server, &auth.auth_id.to_string(), &auth.ticket, &auth.token); + } + *authinfo.write().unwrap() = auth; + tokio::spawn(renewal_future); + } + }); + + let first_auth = if auth_id.is_token() { + // TODO check access here? + None + } else { + Some(BroadcastFuture::new(Box::new(login_future))) + }; Ok(Self { client, server: String::from(server), - auth: BroadcastFuture::new(login), + port, + fingerprint: verified_fingerprint, + auth, + ticket_abort, + first_auth, + _options: options, }) } - fn get_password(_username: &str) -> Result { - use std::env::VarError::*; - match std::env::var("PBS_PASSWORD") { - Ok(p) => return Ok(p), - Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"), - Err(NotPresent) => { - // Try another method - } + /// Login + /// + /// Login is done on demand, so this is only required if you need + /// access to authentication data in 'AuthInfo'. + /// + /// Note: tickets a periodially re-newed, so one can use this + /// to query changed ticket. + pub async fn login(&self) -> Result { + if let Some(future) = &self.first_auth { + future.listen().await?; } + let authinfo = self.auth.read().unwrap(); + Ok(authinfo.clone()) + } + + /// Returns the optional fingerprint passed to the new() constructor. + pub fn fingerprint(&self) -> Option { + (*self.fingerprint.lock().unwrap()).clone() + } + + fn get_password(username: &Userid, interactive: bool) -> Result { // If we're on a TTY, query the user for a password - if tty::stdin_isatty() { - return Ok(String::from_utf8(tty::read_password("Password: ")?)?); + if interactive && tty::stdin_isatty() { + let msg = format!("Password for \"{}\": ", username); + return Ok(String::from_utf8(tty::read_password(&msg)?)?); } bail!("no password input mechanism available"); } - fn build_client() -> Client> { + fn verify_callback( + valid: bool, ctx: + &mut X509StoreContextRef, + expected_fingerprint: Option, + interactive: bool, + ) -> (bool, Option) { + if valid { return (true, None); } - let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap(); + let cert = match ctx.current_cert() { + Some(cert) => cert, + None => return (false, None), + }; - ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE); // fixme! + let depth = ctx.error_depth(); + if depth != 0 { return (false, None); } - let mut httpc = hyper::client::HttpConnector::new(1); - httpc.set_nodelay(true); // important for h2 download performance! - httpc.set_recv_buf_size(Some(1024*1024)); //important for h2 download performance! - httpc.enforce_http(false); // we want https... + let fp = match cert.digest(openssl::hash::MessageDigest::sha256()) { + Ok(fp) => fp, + Err(_) => return (false, None), // should not happen + }; + let fp_string = proxmox::tools::digest_to_hex(&fp); + let fp_string = fp_string.as_bytes().chunks(2).map(|v| std::str::from_utf8(v).unwrap()) + .collect::>().join(":"); - let https = hyper_openssl::HttpsConnector::with_connector(httpc, ssl_connector_builder).unwrap(); + if let Some(expected_fingerprint) = expected_fingerprint { + if expected_fingerprint.to_lowercase() == fp_string { + return (true, Some(fp_string)); + } else { + return (false, None); + } + } - Client::builder() - //.http2_initial_stream_window_size( (1 << 31) - 2) - //.http2_initial_connection_window_size( (1 << 31) - 2) - .build::<_, Body>(https) + // If we're on a TTY, query the user + if interactive && tty::stdin_isatty() { + println!("fingerprint: {}", fp_string); + loop { + print!("Are you sure you want to continue connecting? (y/n): "); + let _ = std::io::stdout().flush(); + use std::io::{BufRead, BufReader}; + let mut line = String::new(); + match BufReader::new(std::io::stdin()).read_line(&mut line) { + Ok(_) => { + let trimmed = line.trim(); + if trimmed == "y" || trimmed == "Y" { + return (true, Some(fp_string)); + } else if trimmed == "n" || trimmed == "N" { + return (false, None); + } else { + continue; + } + } + Err(_) => return (false, None), + } + } + } + (false, None) } - pub fn request(&self, mut req: Request) -> impl Future { - - let login = self.auth.listen(); + pub async fn request(&self, mut req: Request) -> Result { let client = self.client.clone(); - login.and_then(move |auth| { - + let auth = self.login().await?; + if auth.auth_id.is_token() { + let enc_api_token = format!("PBSAPIToken {}:{}", auth.auth_id, percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); + req.headers_mut().insert("Authorization", HeaderValue::from_str(&enc_api_token).unwrap()); + } else { let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap()); req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap()); + } - let request = Self::api_request(client, req); - - request - }) + Self::api_request(client, req).await } - pub fn get(&self, path: &str, data: Option) -> impl Future { - - let req = Self::request_builder(&self.server, "GET", path, data).unwrap(); - self.request(req) + pub async fn get( + &self, + path: &str, + data: Option, + ) -> Result { + let req = Self::request_builder(&self.server, self.port, "GET", path, data)?; + self.request(req).await } - pub fn delete(&mut self, path: &str, data: Option) -> impl Future { - - let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap(); - self.request(req) + pub async fn delete( + &mut self, + path: &str, + data: Option, + ) -> Result { + let req = Self::request_builder(&self.server, self.port, "DELETE", path, data)?; + self.request(req).await } - pub fn post(&mut self, path: &str, data: Option) -> impl Future { - - let req = Self::request_builder(&self.server, "POST", path, data).unwrap(); - self.request(req) + pub async fn post( + &mut self, + path: &str, + data: Option, + ) -> Result { + let req = Self::request_builder(&self.server, self.port, "POST", path, data)?; + self.request(req).await } - pub fn download(&mut self, path: &str, output: W) -> impl Future { - - let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap(); + pub async fn put( + &mut self, + path: &str, + data: Option, + ) -> Result { + let req = Self::request_builder(&self.server, self.port, "PUT", path, data)?; + self.request(req).await + } - let login = self.auth.listen(); + pub async fn download( + &mut self, + path: &str, + output: &mut (dyn Write + Send), + ) -> Result<(), Error> { + let mut req = Self::request_builder(&self.server, self.port, "GET", path, None)?; let client = self.client.clone(); - login.and_then(move |auth| { + let auth = self.login().await?; - let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); - req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap()); + let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); + req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap()); + let resp = tokio::time::timeout( + HTTP_TIMEOUT, client.request(req) + ) + .await + .map_err(|_| format_err!("http download request timed out"))??; + let status = resp.status(); + if !status.is_success() { + HttpClient::api_response(resp) + .map(|_| Err(format_err!("unknown error"))) + .await? + } else { + resp.into_body() .map_err(Error::from) - .and_then(|resp| { - let status = resp.status(); - if !status.is_success() { - future::Either::A( - HttpClient::api_response(resp) - .and_then(|_| { bail!("unknown error"); }) - ) - } else { - future::Either::B( - resp.into_body() - .map_err(Error::from) - .fold(output, move |mut acc, chunk| { - acc.write_all(&chunk)?; - Ok::<_, Error>(acc) - }) - ) - } + .try_fold(output, move |acc, chunk| async move { + acc.write_all(&chunk)?; + Ok::<_, Error>(acc) }) - }) + .await?; + } + Ok(()) } - pub fn upload(&mut self, content_type: &str, body: Body, path: &str) -> impl Future { + pub async fn upload( + &mut self, + content_type: &str, + body: Body, + path: &str, + data: Option, + ) -> Result { let path = path.trim_matches('/'); - let url: Uri = format!("https://{}:8007/{}", &self.server, path).parse().unwrap(); + let mut url = format!("https://{}:{}/{}", &self.server, self.port, path); + + if let Some(data) = data { + let query = tools::json_object_to_query(data).unwrap(); + url.push('?'); + url.push_str(&query); + } + + let url: Uri = url.parse().unwrap(); let req = Request::builder() .method("POST") @@ -260,172 +632,133 @@ impl HttpClient { .header("Content-Type", content_type) .body(body).unwrap(); - self.request(req) - } - - pub fn start_backup( - &self, - datastore: &str, - backup_type: &str, - backup_id: &str, - debug: bool, - ) -> impl Future, Error=Error> { - - let param = json!({"backup-type": backup_type, "backup-id": backup_id, "store": datastore, "debug": debug}); - let req = Self::request_builder(&self.server, "GET", "/api2/json/backup", Some(param)).unwrap(); - - self.start_h2_connection(req, String::from(PROXMOX_BACKUP_PROTOCOL_ID_V1!())) - .map(|(h2, canceller)| BackupClient::new(h2, canceller)) - } - - pub fn start_backup_reader( - &self, - datastore: &str, - backup_type: &str, - backup_id: &str, - backup_time: DateTime, - debug: bool, - ) -> impl Future, Error=Error> { - - let param = json!({ - "backup-type": backup_type, - "backup-id": backup_id, - "backup-time": backup_time.timestamp(), - "store": datastore, - "debug": debug, - }); - let req = Self::request_builder(&self.server, "GET", "/api2/json/reader", Some(param)).unwrap(); - - self.start_h2_connection(req, String::from(PROXMOX_BACKUP_READER_PROTOCOL_ID_V1!())) - .map(|(h2, canceller)| BackupReader::new(h2, canceller)) + self.request(req).await } - pub fn start_h2_connection( + pub async fn start_h2_connection( &self, mut req: Request, protocol_name: String, - ) -> impl Future { + ) -> Result<(H2Client, futures::future::AbortHandle), Error> { - let login = self.auth.listen(); let client = self.client.clone(); + let auth = self.login().await?; - login.and_then(move |auth| { - + if auth.auth_id.is_token() { + let enc_api_token = format!("PBSAPIToken {}:{}", auth.auth_id, percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); + req.headers_mut().insert("Authorization", HeaderValue::from_str(&enc_api_token).unwrap()); + } else { let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET)); req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap()); - req.headers_mut().insert("UPGRADE", HeaderValue::from_str(&protocol_name).unwrap()); + req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap()); + } - client.request(req) - .map_err(Error::from) - .and_then(|resp| { + req.headers_mut().insert("UPGRADE", HeaderValue::from_str(&protocol_name).unwrap()); - let status = resp.status(); - if status != http::StatusCode::SWITCHING_PROTOCOLS { - future::Either::A(Self::api_response(resp).and_then(|_| { bail!("unknown error"); })) - } else { - future::Either::B(resp.into_body().on_upgrade().map_err(Error::from)) - } - }) - .and_then(|upgraded| { - let max_window_size = (1 << 31) - 2; - - h2::client::Builder::new() - .initial_connection_window_size(max_window_size) - .initial_window_size(max_window_size) - .max_frame_size(4*1024*1024) - .handshake(upgraded) - .map_err(Error::from) - }) - .and_then(|(h2, connection)| { - let connection = connection - .map_err(|_| panic!("HTTP/2.0 connection failed")); - - let (connection, canceller) = cancellable(connection)?; - // A cancellable future returns an Option which is None when cancelled and - // Some when it finished instead, since we don't care about the return type we - // need to map it away: - let connection = connection.map(|_| ()); - - // Spawn a new task to drive the connection state - hyper::rt::spawn(connection); - - // Wait until the `SendRequest` handle has available capacity. - Ok(h2.ready() - .map(move |c| (H2Client::new(c), canceller)) - .map_err(Error::from)) - }) - .flatten() - }) - } + let resp = tokio::time::timeout( + HTTP_TIMEOUT, + client.request(req) + ) + .await + .map_err(|_| format_err!("http upgrade request timed out"))??; + let status = resp.status(); + + if status != http::StatusCode::SWITCHING_PROTOCOLS { + Self::api_response(resp).await?; + bail!("unknown error"); + } - fn credentials( - client: Client>, - server: String, - username: String, - password: String, - ) -> Box + Send> { + let upgraded = hyper::upgrade::on(resp).await?; - let server2 = server.clone(); + let max_window_size = (1 << 31) - 2; - let create_request = futures::future::lazy(move || { - let data = json!({ "username": username, "password": password }); - let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap(); - Self::api_request(client, req) - }); + let (h2, connection) = h2::client::Builder::new() + .initial_connection_window_size(max_window_size) + .initial_window_size(max_window_size) + .max_frame_size(4*1024*1024) + .handshake(upgraded) + .await?; - let login_future = create_request - .and_then(move |cred| { - let auth = AuthInfo { - username: cred["data"]["username"].as_str().unwrap().to_owned(), - ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(), - token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(), - }; + let connection = connection + .map_err(|_| eprintln!("HTTP/2.0 connection failed")); - let _ = store_ticket_info(&server2, &auth.username, &auth.ticket, &auth.token); + let (connection, abort) = futures::future::abortable(connection); + // A cancellable future returns an Option which is None when cancelled and + // Some when it finished instead, since we don't care about the return type we + // need to map it away: + let connection = connection.map(|_| ()); - Ok(auth) - }); + // Spawn a new task to drive the connection state + tokio::spawn(connection); - Box::new(login_future) + // Wait until the `SendRequest` handle has available capacity. + let c = h2.ready().await?; + Ok((H2Client::new(c), abort)) } - fn api_response(response: Response) -> impl Future { + async fn credentials( + client: Client, + server: String, + port: u16, + username: Userid, + password: String, + ) -> Result { + let data = json!({ "username": username, "password": password }); + let req = Self::request_builder(&server, port, "POST", "/api2/json/access/ticket", Some(data))?; + let cred = Self::api_request(client, req).await?; + let auth = AuthInfo { + auth_id: cred["data"]["username"].as_str().unwrap().parse()?, + ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(), + token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(), + }; + Ok(auth) + } + + async fn api_response(response: Response) -> Result { let status = response.status(); + let data = hyper::body::to_bytes(response.into_body()).await?; - response - .into_body() - .concat2() - .map_err(Error::from) - .and_then(move |data| { - - let text = String::from_utf8(data.to_vec()).unwrap(); - if status.is_success() { - if text.len() > 0 { - let value: Value = serde_json::from_str(&text)?; - Ok(value) - } else { - Ok(Value::Null) - } - } else { - bail!("HTTP Error {}: {}", status, text); - } - }) + let text = String::from_utf8(data.to_vec()).unwrap(); + if status.is_success() { + if text.is_empty() { + Ok(Value::Null) + } else { + let value: Value = serde_json::from_str(&text)?; + Ok(value) + } + } else { + Err(Error::from(HttpError::new(status, text))) + } } - fn api_request( - client: Client>, + async fn api_request( + client: Client, req: Request - ) -> impl Future { + ) -> Result { - client.request(req) - .map_err(Error::from) - .and_then(Self::api_response) + Self::api_response( + tokio::time::timeout( + HTTP_TIMEOUT, + client.request(req) + ) + .await + .map_err(|_| format_err!("http request timed out"))?? + ).await } - pub fn request_builder(server: &str, method: &str, path: &str, data: Option) -> Result, Error> { + // Read-only access to server property + pub fn server(&self) -> &str { + &self.server + } + + pub fn port(&self) -> u16 { + self.port + } + + pub fn request_builder(server: &str, port: u16, method: &str, path: &str, data: Option) -> Result, Error> { let path = path.trim_matches('/'); - let url: Uri = format!("https://{}:8007/{}", server, path).parse()?; + let url: Uri = format!("https://{}:{}/{}", server, port, path).parse()?; if let Some(data) = data { if method == "POST" { @@ -438,7 +771,7 @@ impl HttpClient { return Ok(request); } else { let query = tools::json_object_to_query(data)?; - let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?; + let url: Uri = format!("https://{}:{}/{}?{}", server, port, path, query).parse()?; let request = Request::builder() .method(method) .uri(url) @@ -460,604 +793,105 @@ impl HttpClient { } } - -pub struct BackupReader { - h2: H2Client, - canceller: Canceller, -} - -impl Drop for BackupReader { - +impl Drop for HttpClient { fn drop(&mut self) { - self.canceller.cancel(); + self.ticket_abort.abort(); } } -impl BackupReader { - pub fn new(h2: H2Client, canceller: Canceller) -> Arc { - Arc::new(Self { h2, canceller: canceller }) - } - - pub fn get(&self, path: &str, param: Option) -> impl Future { - self.h2.get(path, param) - } - - pub fn put(&self, path: &str, param: Option) -> impl Future { - self.h2.put(path, param) - } - - pub fn post(&self, path: &str, param: Option) -> impl Future { - self.h2.post(path, param) - } - - pub fn download( - &self, - file_name: &str, - output: W, - ) -> impl Future { - let path = "download"; - let param = json!({ "file-name": file_name }); - self.h2.download(path, Some(param), output) - } - - pub fn speedtest( - &self, - output: W, - ) -> impl Future { - self.h2.download("speedtest", None, output) - } - - pub fn download_chunk( - &self, - digest: &[u8; 32], - output: W, - ) -> impl Future { - let path = "chunk"; - let param = json!({ "digest": proxmox::tools::digest_to_hex(digest) }); - self.h2.download(path, Some(param), output) - } - - pub fn force_close(self) { - self.canceller.cancel(); - } -} - -pub struct BackupClient { - h2: H2Client, - canceller: Canceller, -} - -impl Drop for BackupClient { - - fn drop(&mut self) { - self.canceller.cancel(); - } +#[derive(Clone)] +pub struct H2Client { + h2: h2::client::SendRequest, } -impl BackupClient { - - pub fn new(h2: H2Client, canceller: Canceller) -> Arc { - Arc::new(Self { h2, canceller }) - } - - pub fn get(&self, path: &str, param: Option) -> impl Future { - self.h2.get(path, param) - } - - pub fn put(&self, path: &str, param: Option) -> impl Future { - self.h2.put(path, param) - } - - pub fn post(&self, path: &str, param: Option) -> impl Future { - self.h2.post(path, param) - } - - pub fn finish(self: Arc) -> impl Future { - let canceller = self.canceller.clone(); - self.h2.clone().post("finish", None).map(move |_| { - canceller.cancel(); - () - }) - } +impl H2Client { - pub fn force_close(self) { - self.canceller.cancel(); + pub fn new(h2: h2::client::SendRequest) -> Self { + Self { h2 } } - pub fn upload_blob_from_data( + pub async fn get( &self, - data: Vec, - file_name: &str, - crypt_config: Option>, - compress: bool, - ) -> impl Future { - - let h2 = self.h2.clone(); - let file_name = file_name.to_owned(); - - futures::future::ok(()) - .and_then(move |_| { - let blob = if let Some(ref crypt_config) = crypt_config { - DataBlob::encode(&data, Some(crypt_config), compress)? - } else { - DataBlob::encode(&data, None, compress)? - }; - - let raw_data = blob.into_inner(); - Ok(raw_data) - }) - .and_then(move |raw_data| { - let param = json!({"encoded-size": raw_data.len(), "file-name": file_name }); - h2.upload("blob", Some(param), raw_data) - .map(|_| {}) - }) + path: &str, + param: Option + ) -> Result { + let req = Self::request_builder("localhost", "GET", path, param, None).unwrap(); + self.request(req).await } - pub fn upload_blob_from_file>( + pub async fn put( &self, - src_path: P, - file_name: &str, - crypt_config: Option>, - compress: bool, - ) -> impl Future { - - let h2 = self.h2.clone(); - let file_name = file_name.to_owned(); - let src_path = src_path.as_ref().to_owned(); - - let task = tokio::fs::File::open(src_path.clone()) - .map_err(move |err| format_err!("unable to open file {:?} - {}", src_path, err)) - .and_then(move |file| { - let contents = vec![]; - tokio::io::read_to_end(file, contents) - .map_err(Error::from) - .and_then(move |(_, contents)| { - let blob = if let Some(ref crypt_config) = crypt_config { - DataBlob::encode(&contents, Some(crypt_config), compress)? - } else { - DataBlob::encode(&contents, None, compress)? - }; - let raw_data = blob.into_inner(); - Ok(raw_data) - }) - .and_then(move |raw_data| { - let param = json!({"encoded-size": raw_data.len(), "file-name": file_name }); - h2.upload("blob", Some(param), raw_data) - .map(|_| {}) - }) - }); - - task + path: &str, + param: Option + ) -> Result { + let req = Self::request_builder("localhost", "PUT", path, param, None).unwrap(); + self.request(req).await } - pub fn upload_stream( + pub async fn post( &self, - archive_name: &str, - stream: impl Stream, - prefix: &str, - fixed_size: Option, - crypt_config: Option>, - ) -> impl Future { - - let known_chunks = Arc::new(Mutex::new(HashSet::new())); - - let h2 = self.h2.clone(); - let h2_2 = self.h2.clone(); - let h2_3 = self.h2.clone(); - let h2_4 = self.h2.clone(); - - let mut param = json!({ "archive-name": archive_name }); - if let Some(size) = fixed_size { - param["size"] = size.into(); - } - - let index_path = format!("{}_index", prefix); - let close_path = format!("{}_close", prefix); - - let prefix = prefix.to_owned(); - - Self::download_chunk_list(h2, &index_path, archive_name, known_chunks.clone()) - .and_then(move |_| { - h2_2.post(&index_path, Some(param)) - }) - .and_then(move |res| { - let wid = res.as_u64().unwrap(); - Self::upload_chunk_info_stream(h2_3, wid, stream, &prefix, known_chunks.clone(), crypt_config) - .and_then(move |(chunk_count, size, _speed)| { - let param = json!({ - "wid": wid , - "chunk-count": chunk_count, - "size": size, - }); - h2_4.post(&close_path, Some(param)) - }) - .map(|_| ()) - }) - } - - fn response_queue() -> ( - mpsc::Sender, - sync::oneshot::Receiver> - ) { - let (verify_queue_tx, verify_queue_rx) = mpsc::channel(100); - let (verify_result_tx, verify_result_rx) = sync::oneshot::channel(); - - hyper::rt::spawn( - verify_queue_rx - .map_err(Error::from) - .for_each(|response: h2::client::ResponseFuture| { - response - .map_err(Error::from) - .and_then(H2Client::h2api_response) - .and_then(|result| { - println!("RESPONSE: {:?}", result); - Ok(()) - }) - .map_err(|err| format_err!("pipelined request failed: {}", err)) - }) - .then(|result| - verify_result_tx.send(result) - ) - .map_err(|_| { /* ignore closed channel */ }) - ); - - (verify_queue_tx, verify_result_rx) - } - - fn append_chunk_queue(h2: H2Client, wid: u64, path: String) -> ( - mpsc::Sender<(MergedChunkInfo, Option)>, - sync::oneshot::Receiver> - ) { - let (verify_queue_tx, verify_queue_rx) = mpsc::channel(64); - let (verify_result_tx, verify_result_rx) = sync::oneshot::channel(); - - let h2_2 = h2.clone(); - - hyper::rt::spawn( - verify_queue_rx - .map_err(Error::from) - .and_then(move |(merged_chunk_info, response): (MergedChunkInfo, Option)| { - match (response, merged_chunk_info) { - (Some(response), MergedChunkInfo::Known(list)) => { - future::Either::A( - response - .map_err(Error::from) - .and_then(H2Client::h2api_response) - .and_then(move |_result| { - Ok(MergedChunkInfo::Known(list)) - }) - ) - } - (None, MergedChunkInfo::Known(list)) => { - future::Either::B(future::ok(MergedChunkInfo::Known(list))) - } - _ => unreachable!(), - } - }) - .merge_known_chunks() - .and_then(move |merged_chunk_info| { - match merged_chunk_info { - MergedChunkInfo::Known(chunk_list) => { - let mut digest_list = vec![]; - let mut offset_list = vec![]; - for (offset, digest) in chunk_list { - //println!("append chunk {} (offset {})", proxmox::tools::digest_to_hex(&digest), offset); - digest_list.push(proxmox::tools::digest_to_hex(&digest)); - offset_list.push(offset); - } - println!("append chunks list len ({})", digest_list.len()); - let param = json!({ "wid": wid, "digest-list": digest_list, "offset-list": offset_list }); - let mut request = H2Client::request_builder("localhost", "PUT", &path, None).unwrap(); - request.headers_mut().insert(hyper::header::CONTENT_TYPE, HeaderValue::from_static("application/json")); - let param_data = bytes::Bytes::from(param.to_string().as_bytes()); - let upload_data = Some(param_data); - h2_2.send_request(request, upload_data) - .and_then(move |response| { - response - .map_err(Error::from) - .and_then(H2Client::h2api_response) - .and_then(|_| Ok(())) - }) - .map_err(|err| format_err!("pipelined request failed: {}", err)) - } - _ => unreachable!(), - } - }) - .for_each(|_| Ok(())) - .then(|result| - verify_result_tx.send(result) - ) - .map_err(|_| { /* ignore closed channel */ }) - ); - - (verify_queue_tx, verify_result_rx) - } - - fn download_chunk_list( - h2: H2Client, path: &str, - archive_name: &str, - known_chunks: Arc>>, - ) -> impl Future { - - let param = json!({ "archive-name": archive_name }); - let request = H2Client::request_builder("localhost", "GET", path, Some(param)).unwrap(); - - h2.send_request(request, None) - .and_then(move |response| { - response - .map_err(Error::from) - .and_then(move |resp| { - let status = resp.status(); - - if !status.is_success() { - future::Either::A(H2Client::h2api_response(resp).and_then(|_| { bail!("unknown error"); })) - } else { - future::Either::B(future::ok(resp.into_body())) - } - }) - .and_then(move |mut body| { - - let mut release_capacity = body.release_capacity().clone(); - - DigestListDecoder::new(body.map_err(Error::from)) - .for_each(move |chunk| { - let _ = release_capacity.release_capacity(chunk.len()); - println!("GOT DOWNLOAD {}", proxmox::tools::digest_to_hex(&chunk)); - known_chunks.lock().unwrap().insert(chunk); - Ok(()) - }) - }) - }) + param: Option + ) -> Result { + let req = Self::request_builder("localhost", "POST", path, param, None).unwrap(); + self.request(req).await } - fn upload_chunk_info_stream( - h2: H2Client, - wid: u64, - stream: impl Stream, - prefix: &str, - known_chunks: Arc>>, - crypt_config: Option>, - ) -> impl Future { - - let repeat = std::sync::Arc::new(AtomicUsize::new(0)); - let repeat2 = repeat.clone(); - - let stream_len = std::sync::Arc::new(AtomicUsize::new(0)); - let stream_len2 = stream_len.clone(); - - let append_chunk_path = format!("{}_index", prefix); - let upload_chunk_path = format!("{}_chunk", prefix); - - let (upload_queue, upload_result) = Self::append_chunk_queue(h2.clone(), wid, append_chunk_path.to_owned()); - - let start_time = std::time::Instant::now(); - - stream - .and_then(move |data| { - - let chunk_len = data.len(); - - repeat.fetch_add(1, Ordering::SeqCst); - let offset = stream_len.fetch_add(chunk_len, Ordering::SeqCst) as u64; - - let mut chunk_builder = DataChunkBuilder::new(data.as_ref()) - .compress(true); - - if let Some(ref crypt_config) = crypt_config { - chunk_builder = chunk_builder.crypt_config(crypt_config); - } - - let mut known_chunks = known_chunks.lock().unwrap(); - let digest = chunk_builder.digest(); - let chunk_is_known = known_chunks.contains(digest); - if chunk_is_known { - Ok(MergedChunkInfo::Known(vec![(offset, *digest)])) - } else { - known_chunks.insert(*digest); - let chunk = chunk_builder.build()?; - Ok(MergedChunkInfo::New(ChunkInfo { chunk, chunk_len: chunk_len as u64, offset })) - } - }) - .merge_known_chunks() - .for_each(move |merged_chunk_info| { - - if let MergedChunkInfo::New(chunk_info) = merged_chunk_info { - let offset = chunk_info.offset; - let digest = *chunk_info.chunk.digest(); - let digest_str = proxmox::tools::digest_to_hex(&digest); - let upload_queue = upload_queue.clone(); - - println!("upload new chunk {} ({} bytes, offset {})", digest_str, - chunk_info.chunk_len, offset); - - let chunk_data = chunk_info.chunk.raw_data(); - let param = json!({ - "wid": wid, - "digest": digest_str, - "size": chunk_info.chunk_len, - "encoded-size": chunk_data.len(), - }); - - let request = H2Client::request_builder("localhost", "POST", &upload_chunk_path, Some(param)).unwrap(); - let upload_data = Some(bytes::Bytes::from(chunk_data)); - - let new_info = MergedChunkInfo::Known(vec![(offset, digest)]); - - future::Either::A( - h2.send_request(request, upload_data) - .and_then(move |response| { - upload_queue.clone().send((new_info, Some(response))) - .map(|_| ()).map_err(Error::from) - }) - ) - } else { + pub async fn download( + &self, + path: &str, + param: Option, + mut output: W, + ) -> Result<(), Error> { + let request = Self::request_builder("localhost", "GET", path, param, None).unwrap(); - future::Either::B( - upload_queue.clone().send((merged_chunk_info, None)) - .map(|_| ()).map_err(Error::from) - ) - } - }) - .then(move |result| { - println!("RESULT {:?}", result); - upload_result.map_err(Error::from).and_then(|upload1_result| { - Ok(upload1_result.and(result)) - }) - }) - .flatten() - .and_then(move |_| { - let repeat = repeat2.load(Ordering::SeqCst); - let stream_len = stream_len2.load(Ordering::SeqCst); - let speed = ((stream_len*1000000)/(1024*1024))/(start_time.elapsed().as_micros() as usize); - println!("Uploaded {} chunks in {} seconds ({} MB/s).", repeat, start_time.elapsed().as_secs(), speed); - if repeat > 0 { - println!("Average chunk size was {} bytes.", stream_len/repeat); - println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128)); - } - Ok((repeat, stream_len, speed)) - }) - } + let response_future = self.send_request(request, None).await?; - pub fn upload_speedtest(&self) -> impl Future { + let resp = response_future.await?; - let mut data = vec![]; - // generate pseudo random byte sequence - for i in 0..1024*1024 { - for j in 0..4 { - let byte = ((i >> (j<<3))&0xff) as u8; - data.push(byte); - } + let status = resp.status(); + if !status.is_success() { + H2Client::h2api_response(resp).await?; // raise error + unreachable!(); } - let item_len = data.len(); - - let repeat = std::sync::Arc::new(AtomicUsize::new(0)); - let repeat2 = repeat.clone(); - - let (upload_queue, upload_result) = Self::response_queue(); - - let start_time = std::time::Instant::now(); - - let h2 = self.h2.clone(); - - futures::stream::repeat(data) - .take_while(move |_| { - repeat.fetch_add(1, Ordering::SeqCst); - Ok(start_time.elapsed().as_secs() < 5) - }) - .for_each(move |data| { - let h2 = h2.clone(); - - let upload_queue = upload_queue.clone(); - - println!("send test data ({} bytes)", data.len()); - let request = H2Client::request_builder("localhost", "POST", "speedtest", None).unwrap(); - h2.send_request(request, Some(bytes::Bytes::from(data))) - .and_then(move |response| { - upload_queue.send(response) - .map(|_| ()).map_err(Error::from) - }) - }) - .then(move |result| { - println!("RESULT {:?}", result); - upload_result.map_err(Error::from).and_then(|upload1_result| { - Ok(upload1_result.and(result)) - }) - }) - .flatten() - .and_then(move |_| { - let repeat = repeat2.load(Ordering::SeqCst); - println!("Uploaded {} chunks in {} seconds.", repeat, start_time.elapsed().as_secs()); - let speed = ((item_len*1000000*(repeat as usize))/(1024*1024))/(start_time.elapsed().as_micros() as usize); - if repeat > 0 { - println!("Time per request: {} microseconds.", (start_time.elapsed().as_micros())/(repeat as u128)); - } - Ok(speed) - }) - } -} - -#[derive(Clone)] -pub struct H2Client { - h2: h2::client::SendRequest, -} - -impl H2Client { - - pub fn new(h2: h2::client::SendRequest) -> Self { - Self { h2 } - } - - pub fn get(&self, path: &str, param: Option) -> impl Future { - let req = Self::request_builder("localhost", "GET", path, param).unwrap(); - self.request(req) - } + let mut body = resp.into_body(); + while let Some(chunk) = body.data().await { + let chunk = chunk?; + body.flow_control().release_capacity(chunk.len())?; + output.write_all(&chunk)?; + } - pub fn put(&self, path: &str, param: Option) -> impl Future { - let req = Self::request_builder("localhost", "PUT", path, param).unwrap(); - self.request(req) + Ok(()) } - pub fn post(&self, path: &str, param: Option) -> impl Future { - let req = Self::request_builder("localhost", "POST", path, param).unwrap(); - self.request(req) - } + pub async fn upload( + &self, + method: &str, // POST or PUT + path: &str, + param: Option, + content_type: &str, + data: Vec, + ) -> Result { + let request = Self::request_builder("localhost", method, path, param, Some(content_type)).unwrap(); - pub fn download(&self, path: &str, param: Option, output: W) -> impl Future { - let request = Self::request_builder("localhost", "GET", path, param).unwrap(); + let mut send_request = self.h2.clone().ready().await?; - self.send_request(request, None) - .and_then(move |response| { - response - .map_err(Error::from) - .and_then(move |resp| { - let status = resp.status(); - if !status.is_success() { - future::Either::A( - H2Client::h2api_response(resp) - .and_then(|_| { bail!("unknown error"); }) - ) - } else { - let mut body = resp.into_body(); - let mut release_capacity = body.release_capacity().clone(); - - future::Either::B( - body - .map_err(Error::from) - .fold(output, move |mut acc, chunk| { - let _ = release_capacity.release_capacity(chunk.len()); - acc.write_all(&chunk)?; - Ok::<_, Error>(acc) - }) - ) - } - }) - }) - } + let (response, stream) = send_request.send_request(request, false).unwrap(); - pub fn upload(&self, path: &str, param: Option, data: Vec) -> impl Future { - let request = Self::request_builder("localhost", "POST", path, param).unwrap(); + PipeToSendStream::new(bytes::Bytes::from(data), stream).await?; - self.h2.clone() - .ready() + response .map_err(Error::from) - .and_then(move |mut send_request| { - let (response, stream) = send_request.send_request(request, false).unwrap(); - PipeToSendStream::new(bytes::Bytes::from(data), stream) - .and_then(|_| { - response - .map_err(Error::from) - .and_then(Self::h2api_response) - }) - }) + .and_then(Self::h2api_response) + .await } - fn request( + async fn request( &self, request: Request<()>, - ) -> impl Future { + ) -> Result { self.send_request(request, None) .and_then(move |response| { @@ -1065,79 +899,80 @@ impl H2Client { .map_err(Error::from) .and_then(Self::h2api_response) }) + .await } - fn send_request( + pub fn send_request( &self, request: Request<()>, data: Option, - ) -> impl Future { + ) -> impl Future> { self.h2.clone() .ready() .map_err(Error::from) - .and_then(move |mut send_request| { + .and_then(move |mut send_request| async move { if let Some(data) = data { let (response, stream) = send_request.send_request(request, false).unwrap(); - future::Either::A(PipeToSendStream::new(data, stream) - .and_then(move |_| { - future::ok(response) - })) + PipeToSendStream::new(data, stream).await?; + Ok(response) } else { let (response, _stream) = send_request.send_request(request, true).unwrap(); - future::Either::B(future::ok(response)) + Ok(response) } }) } - fn h2api_response(response: Response) -> impl Future { - + pub async fn h2api_response( + response: Response, + ) -> Result { let status = response.status(); let (_head, mut body) = response.into_parts(); - // The `release_capacity` handle allows the caller to manage - // flow control. - // - // Whenever data is received, the caller is responsible for - // releasing capacity back to the server once it has freed - // the data from memory. - let mut release_capacity = body.release_capacity().clone(); - - body - .map(move |chunk| { - // Let the server send more data. - let _ = release_capacity.release_capacity(chunk.len()); - chunk - }) - .concat2() - .map_err(Error::from) - .and_then(move |data| { - let text = String::from_utf8(data.to_vec()).unwrap(); - if status.is_success() { - if text.len() > 0 { - let mut value: Value = serde_json::from_str(&text)?; - if let Some(map) = value.as_object_mut() { - if let Some(data) = map.remove("data") { - return Ok(data); - } - } - bail!("got result without data property"); - } else { - Ok(Value::Null) + let mut data = Vec::new(); + while let Some(chunk) = body.data().await { + let chunk = chunk?; + // Whenever data is received, the caller is responsible for + // releasing capacity back to the server once it has freed + // the data from memory. + // Let the server send more data. + body.flow_control().release_capacity(chunk.len())?; + data.extend(chunk); + } + + let text = String::from_utf8(data.to_vec()).unwrap(); + if status.is_success() { + if text.is_empty() { + Ok(Value::Null) + } else { + let mut value: Value = serde_json::from_str(&text)?; + if let Some(map) = value.as_object_mut() { + if let Some(data) = map.remove("data") { + return Ok(data); } - } else { - bail!("HTTP Error {}: {}", status, text); } - }) + bail!("got result without data property"); + } + } else { + Err(Error::from(HttpError::new(status, text))) + } } // Note: We always encode parameters with the url - pub fn request_builder(server: &str, method: &str, path: &str, data: Option) -> Result, Error> { + pub fn request_builder( + server: &str, + method: &str, + path: &str, + param: Option, + content_type: Option<&str>, + ) -> Result, Error> { let path = path.trim_matches('/'); - if let Some(data) = data { - let query = tools::json_object_to_query(data)?; + let content_type = content_type.unwrap_or("application/x-www-form-urlencoded"); + + if let Some(param) = param { + let query = tools::json_object_to_query(param)?; // We detected problem with hyper around 6000 characters - seo we try to keep on the safe side if query.len() > 4096 { bail!("h2 query data too large ({} bytes) - please encode data inside body", query.len()); } let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?; @@ -1145,16 +980,16 @@ impl H2Client { .method(method) .uri(url) .header("User-Agent", "proxmox-backup-client/1.0") - .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .header(hyper::header::CONTENT_TYPE, content_type) .body(())?; - return Ok(request); + Ok(request) } else { let url: Uri = format!("https://{}:8007/{}", server, path).parse()?; let request = Request::builder() .method(method) .uri(url) .header("User-Agent", "proxmox-backup-client/1.0") - .header(hyper::header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .header(hyper::header::CONTENT_TYPE, content_type) .body(())?; Ok(request)