[pbs-devel] [RFC proxmox-backup 0/5] proxy support for HttpsConnector

[pbs-devel] [RFC proxmox-backup 0/5] proxy support for HttpsConnector

[Dietmar Maurer]

This time without using an external crate.

Note: Secured proxy connections are not supported (rarely used feature?)

Dietmar Maurer (5):
  http: rename EitherStream to MaybeTlsStream
  MaybeTlsStream: implement poll_write_vectored()
  new http client implementation SimpleHttp (avoid static HTTP_CLIENT)
  HttpsConnector: code cleanup
  HttpsConnector: add proxy support

 src/api2/node/apt.rs      |   8 +-
 src/tools/async_io.rs     |  90 +++++++---
 src/tools/http.rs         | 336 +++++++++++++++++++++++++++-----------
 src/tools/subscription.rs |  10 +-
 4 files changed, 316 insertions(+), 128 deletions(-)

-- 
2.20.1

[pbs-devel] [RFC proxmox-backup 1/5] http: rename EitherStream to MaybeTlsStream

[Dietmar Maurer]

And rename the enum values. Added an additional enum called Proxied.

The enum in now more specialized, but we only use it for the http client anyways.
---
 src/tools/async_io.rs | 64 +++++++++++++++++++++++++++----------------
 src/tools/http.rs     | 20 ++++++--------
 2 files changed, 48 insertions(+), 36 deletions(-)

diff --git a/src/tools/async_io.rs b/src/tools/async_io.rs
index 844afaa9..963f6fdd 100644
--- a/src/tools/async_io.rs
+++ b/src/tools/async_io.rs
@@ -1,4 +1,4 @@
-//! Generic AsyncRead/AsyncWrite utilities.
+//! AsyncRead/AsyncWrite utilities.
 
 use std::io;
 use std::os::unix::io::{AsRawFd, RawFd};
@@ -9,41 +9,52 @@ use futures::stream::{Stream, TryStream};
 use futures::ready;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::TcpListener;
-use hyper::client::connect::Connection;
-
-pub enum EitherStream<L, R> {
-    Left(L),
-    Right(R),
+use tokio_openssl::SslStream;
+use hyper::client::connect::{Connection, Connected};
+
+/// Asynchronous stream, possibly encrypted and proxied
+///
+/// Usefule for HTTP client implementations using hyper.
+pub enum MaybeTlsStream<S> {
+    Normal(S),
+    Proxied(S),
+    Secured(SslStream<S>),
 }
 
-impl<L: AsyncRead + Unpin, R: AsyncRead + Unpin> AsyncRead for EitherStream<L, R> {
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncRead for MaybeTlsStream<S> {
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &mut ReadBuf,
     ) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
+                Pin::new(s).poll_read(cx, buf)
+            }
+            MaybeTlsStream::Proxied(ref mut s) => {
                 Pin::new(s).poll_read(cx, buf)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_read(cx, buf)
             }
         }
     }
 }
 
-impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L, R> {
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for MaybeTlsStream<S> {
     fn poll_write(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &[u8],
     ) -> Poll<Result<usize, io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
+                Pin::new(s).poll_write(cx, buf)
+            }
+            MaybeTlsStream::Proxied(ref mut s) => {
                 Pin::new(s).poll_write(cx, buf)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_write(cx, buf)
             }
         }
@@ -51,10 +62,13 @@ impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_flush(cx)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_flush(cx)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_flush(cx)
             }
         }
@@ -62,25 +76,27 @@ impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L
 
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_shutdown(cx)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_shutdown(cx)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_shutdown(cx)
             }
         }
     }
 }
 
-// we need this for crate::client::http_client:
-impl Connection for EitherStream<
-    tokio::net::TcpStream,
-    Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>,
-> {
-    fn connected(&self) -> hyper::client::connect::Connected {
+// we need this for the hyper http client
+impl <S: Connection + AsyncRead + AsyncWrite + Unpin> Connection for MaybeTlsStream<S>
+{
+    fn connected(&self) -> Connected {
         match self {
-            EitherStream::Left(s) => s.connected(),
-            EitherStream::Right(s) => s.get_ref().connected(),
+            MaybeTlsStream::Normal(s) => s.connected(),
+            MaybeTlsStream::Proxied(s) => s.connected().proxy(true),
+            MaybeTlsStream::Secured(s) => s.get_ref().connected(),
         }
     }
 }
diff --git a/src/tools/http.rs b/src/tools/http.rs
index d08ce451..3cd3af4e 100644
--- a/src/tools/http.rs
+++ b/src/tools/http.rs
@@ -10,9 +10,11 @@ use hyper::client::{Client, HttpConnector};
 use http::{Request, Response};
 use openssl::ssl::{SslConnector, SslMethod};
 use futures::*;
+use tokio::net::TcpStream;
+use tokio_openssl::SslStream;
 
 use crate::tools::{
-    async_io::EitherStream,
+    async_io::MaybeTlsStream,
     socket::{
         set_tcp_keepalive,
         PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
@@ -100,13 +102,8 @@ impl HttpsConnector {
     }
 }
 
-type MaybeTlsStream = EitherStream<
-    tokio::net::TcpStream,
-    Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>,
->;
-
 impl hyper::service::Service<Uri> for HttpsConnector {
-    type Response = MaybeTlsStream;
+    type Response = MaybeTlsStream<TcpStream>;
     type Error = Error;
     #[allow(clippy::type_complexity)]
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
@@ -140,12 +137,11 @@ impl hyper::service::Service<Uri> for HttpsConnector {
             let _ = set_tcp_keepalive(conn.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
 
             if is_https {
-                let conn: tokio_openssl::SslStream<tokio::net::TcpStream> = tokio_openssl::SslStream::new(config?.into_ssl(&host)?, conn)?;
-                let mut conn = Box::pin(conn);
-                conn.as_mut().connect().await?;
-                Ok(MaybeTlsStream::Right(conn))
+                let mut conn: SslStream<TcpStream> = SslStream::new(config?.into_ssl(&host)?, conn)?;
+                Pin::new(&mut conn).connect().await?;
+                Ok(MaybeTlsStream::Secured(conn))
             } else {
-                Ok(MaybeTlsStream::Left(conn))
+                Ok(MaybeTlsStream::Normal(conn))
             }
         }.boxed()
     }
-- 
2.20.1

[pbs-devel] [RFC proxmox-backup 2/5] MaybeTlsStream: implement poll_write_vectored()

[Dietmar Maurer]

This is just an performance optimization.
---
 src/tools/async_io.rs | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/src/tools/async_io.rs b/src/tools/async_io.rs
index 963f6fdd..83110912 100644
--- a/src/tools/async_io.rs
+++ b/src/tools/async_io.rs
@@ -60,6 +60,32 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for MaybeTlsStream<S> {
         }
     }
 
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[io::IoSlice<'_>],
+    ) -> Poll<Result<usize, io::Error>> {
+        match self.get_mut() {
+            MaybeTlsStream::Normal(ref mut s) => {
+                Pin::new(s).poll_write_vectored(cx, bufs)
+            }
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_write_vectored(cx, bufs)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
+                Pin::new(s).poll_write_vectored(cx, bufs)
+            }
+        }
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        match self {
+            MaybeTlsStream::Normal(s) => s.is_write_vectored(),
+            MaybeTlsStream::Proxied(s) => s.is_write_vectored(),
+            MaybeTlsStream::Secured(s) => s.is_write_vectored(),
+        }
+    }
+
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
             MaybeTlsStream::Normal(ref mut s) => {
-- 
2.20.1

[pbs-devel] [RFC proxmox-backup 3/5] new http client implementation SimpleHttp (avoid static HTTP_CLIENT)

[Dietmar Maurer]

This one will have proxy support.
---
 src/api2/node/apt.rs      |   8 ++-
 src/tools/http.rs         | 122 +++++++++++++++++++++-----------------
 src/tools/subscription.rs |  10 ++--
 3 files changed, 80 insertions(+), 60 deletions(-)

diff --git a/src/api2/node/apt.rs b/src/api2/node/apt.rs
index e77b89fa..dbdb2019 100644
--- a/src/api2/node/apt.rs
+++ b/src/api2/node/apt.rs
@@ -7,7 +7,7 @@ use proxmox::api::{api, RpcEnvironment, RpcEnvironmentType, Permission};
 use proxmox::api::router::{Router, SubdirMap};
 
 use crate::server::WorkerTask;
-use crate::tools::{apt, http, subscription};
+use crate::tools::{apt, http::SimpleHttp, subscription};
 
 use crate::config::acl::{PRIV_SYS_AUDIT, PRIV_SYS_MODIFY};
 use crate::api2::types::{Authid, APTUpdateInfo, NODE_SCHEMA, UPID_SCHEMA};
@@ -194,10 +194,12 @@ fn apt_get_changelog(
         bail!("Package '{}' not found", name);
     }
 
+    let mut client = SimpleHttp::new();
+
     let changelog_url = &pkg_info[0].change_log_url;
     // FIXME: use 'apt-get changelog' for proxmox packages as well, once repo supports it
     if changelog_url.starts_with("http://download.proxmox.com/") {
-        let changelog = crate::tools::runtime::block_on(http::get_string(changelog_url, None))
+        let changelog = crate::tools::runtime::block_on(client.get_string(changelog_url, None))
             .map_err(|err| format_err!("Error downloading changelog from '{}': {}", changelog_url, err))?;
         Ok(json!(changelog))
 
@@ -221,7 +223,7 @@ fn apt_get_changelog(
         auth_header.insert("Authorization".to_owned(),
             format!("Basic {}", base64::encode(format!("{}:{}", key, id))));
 
-        let changelog = crate::tools::runtime::block_on(http::get_string(changelog_url, Some(&auth_header)))
+        let changelog = crate::tools::runtime::block_on(client.get_string(changelog_url, Some(&auth_header)))
             .map_err(|err| format_err!("Error downloading changelog from '{}': {}", changelog_url, err))?;
         Ok(json!(changelog))
 
diff --git a/src/tools/http.rs b/src/tools/http.rs
index 3cd3af4e..f19d6527 100644
--- a/src/tools/http.rs
+++ b/src/tools/http.rs
@@ -1,5 +1,4 @@
 use anyhow::{Error, format_err, bail};
-use lazy_static::lazy_static;
 use std::task::{Context, Poll};
 use std::os::unix::io::AsRawFd;
 use std::collections::HashMap;
@@ -21,68 +20,85 @@ use crate::tools::{
     },
 };
 
-lazy_static! {
-    static ref HTTP_CLIENT: Client<HttpsConnector, Body> = {
-        let connector = SslConnector::builder(SslMethod::tls()).unwrap().build();
-        let httpc = HttpConnector::new();
-        let https = HttpsConnector::with_connector(httpc, connector);
-        Client::builder().build(https)
-    };
+/// Asyncrounous HTTP client implementation
+pub struct SimpleHttp {
+    client: Client<HttpsConnector, Body>,
 }
 
-pub async fn get_string(uri: &str, extra_headers: Option<&HashMap<String, String>>) -> Result<String, Error> {
-    let mut request = Request::builder()
-        .method("GET")
-        .uri(uri)
-        .header("User-Agent", "proxmox-backup-client/1.0");
+impl SimpleHttp {
 
-    if let Some(hs) = extra_headers {
-        for (h, v) in hs.iter() {
-            request = request.header(h, v);
-        }
+    pub fn new() -> Self {
+        let ssl_connector = SslConnector::builder(SslMethod::tls()).unwrap().build();
+        Self::with_ssl_connector(ssl_connector)
     }
 
-    let request = request.body(Body::empty())?;
-
-    let res = HTTP_CLIENT.request(request).await?;
+    pub fn with_ssl_connector(ssl_connector: SslConnector) -> Self {
+        let connector = HttpConnector::new();
+        let https = HttpsConnector::with_connector(connector, ssl_connector);
+        let client = Client::builder().build(https);
+        Self { client }
+    }
 
-    let status = res.status();
-    if !status.is_success() {
-        bail!("Got bad status '{}' from server", status)
+    pub async fn post(
+        &mut self,
+        uri: &str,
+        body: Option<String>,
+        content_type: Option<&str>,
+    ) -> Result<Response<Body>, Error> {
+
+        let body = if let Some(body) = body {
+            Body::from(body)
+        } else {
+            Body::empty()
+        };
+        let content_type = content_type.unwrap_or("application/json");
+
+        let request = Request::builder()
+            .method("POST")
+            .uri(uri)
+            .header("User-Agent", "proxmox-backup-client/1.0")
+            .header(hyper::header::CONTENT_TYPE, content_type)
+            .body(body)?;
+
+        self.client.request(request)
+            .map_err(Error::from)
+            .await
     }
 
-    response_body_string(res).await
-}
+    pub async fn get_string(
+        &mut self,
+        uri: &str,
+        extra_headers: Option<&HashMap<String, String>>,
+    ) -> Result<String, Error> {
 
-pub async fn response_body_string(res: Response<Body>) -> Result<String, Error> {
-    let buf = hyper::body::to_bytes(res).await?;
-    String::from_utf8(buf.to_vec())
-        .map_err(|err| format_err!("Error converting HTTP result data: {}", err))
-}
+        let mut request = Request::builder()
+            .method("GET")
+            .uri(uri)
+            .header("User-Agent", "proxmox-backup-client/1.0");
+
+        if let Some(hs) = extra_headers {
+            for (h, v) in hs.iter() {
+                request = request.header(h, v);
+            }
+        }
 
-pub async fn post(
-    uri: &str,
-    body: Option<String>,
-    content_type: Option<&str>,
-) -> Result<Response<Body>, Error> {
-    let body = if let Some(body) = body {
-        Body::from(body)
-    } else {
-        Body::empty()
-    };
-    let content_type = content_type.unwrap_or("application/json");
-
-    let request = Request::builder()
-        .method("POST")
-        .uri(uri)
-        .header("User-Agent", "proxmox-backup-client/1.0")
-        .header(hyper::header::CONTENT_TYPE, content_type)
-        .body(body)?;
-
-
-    HTTP_CLIENT.request(request)
-        .map_err(Error::from)
-        .await
+        let request = request.body(Body::empty())?;
+
+        let res = self.client.request(request).await?;
+
+        let status = res.status();
+        if !status.is_success() {
+            bail!("Got bad status '{}' from server", status)
+        }
+
+        Self::response_body_string(res).await
+    }
+
+    pub async fn response_body_string(res: Response<Body>) -> Result<String, Error> {
+        let buf = hyper::body::to_bytes(res).await?;
+        String::from_utf8(buf.to_vec())
+            .map_err(|err| format_err!("Error converting HTTP result data: {}", err))
+    }
 }
 
 #[derive(Clone)]
diff --git a/src/tools/subscription.rs b/src/tools/subscription.rs
index 9b9534ac..9a920aee 100644
--- a/src/tools/subscription.rs
+++ b/src/tools/subscription.rs
@@ -6,8 +6,7 @@ use regex::Regex;
 
 use proxmox::api::api;
 
-use crate::tools;
-use crate::tools::http;
+use crate::tools::{self, http::SimpleHttp};
 use proxmox::tools::fs::{replace_file, CreateOptions};
 
 /// How long the local key is valid for in between remote checks
@@ -102,10 +101,13 @@ async fn register_subscription(
         "ip": "localhost",
         "check_token": challenge,
     });
+
+    let mut client = SimpleHttp::new();
+
     let uri = "https://shop.maurer-it.com/modules/servers/licensing/verify.php";
     let query = tools::json_object_to_query(params)?;
-    let response = http::post(uri, Some(query), Some("application/x-www-form-urlencoded")).await?;
-    let body = http::response_body_string(response).await?;
+    let response = client.post(uri, Some(query), Some("application/x-www-form-urlencoded")).await?;
+    let body = SimpleHttp::response_body_string(response).await?;
 
     Ok((body, challenge))
 }
-- 
2.20.1

[pbs-devel] [RFC proxmox-backup 4/5] HttpsConnector: code cleanup

[Dietmar Maurer]

---
 src/tools/http.rs | 42 ++++++++++++++++++++----------------------
 1 file changed, 20 insertions(+), 22 deletions(-)

diff --git a/src/tools/http.rs b/src/tools/http.rs
index f19d6527..a0dbfd01 100644
--- a/src/tools/http.rs
+++ b/src/tools/http.rs
@@ -3,6 +3,7 @@ use std::task::{Context, Poll};
 use std::os::unix::io::AsRawFd;
 use std::collections::HashMap;
 use std::pin::Pin;
+use std::sync::Arc;
 
 use hyper::{Uri, Body};
 use hyper::client::{Client, HttpConnector};
@@ -103,17 +104,16 @@ impl SimpleHttp {
 
 #[derive(Clone)]
 pub struct HttpsConnector {
-    http: HttpConnector,
-    ssl_connector: std::sync::Arc<SslConnector>,
+    connector: HttpConnector,
+    ssl_connector: Arc<SslConnector>,
 }
 
 impl HttpsConnector {
-    pub fn with_connector(mut http: HttpConnector, ssl_connector: SslConnector) -> Self {
-        http.enforce_http(false);
-
+    pub fn with_connector(mut connector: HttpConnector, ssl_connector: SslConnector) -> Self {
+        connector.enforce_http(false);
         Self {
-            http,
-            ssl_connector: std::sync::Arc::new(ssl_connector),
+            connector,
+            ssl_connector: Arc::new(ssl_connector),
         }
     }
 }
@@ -130,22 +130,20 @@ impl hyper::service::Service<Uri> for HttpsConnector {
     }
 
     fn call(&mut self, dst: Uri) -> Self::Future {
-        let mut this = self.clone();
-        async move {
-            let is_https = dst
-                .scheme()
-                .ok_or_else(|| format_err!("missing URL scheme"))?
-                == "https";
-
-            let host = dst
-                .host()
-                .ok_or_else(|| format_err!("missing hostname in destination url?"))?
-                .to_string();
+        let mut connector = self.connector.clone();
+        let ssl_connector = Arc::clone(&self.ssl_connector);
+        let is_https = dst.scheme() == Some(&http::uri::Scheme::HTTPS);
+        let host = match dst.host() {
+            Some(host) => host.to_owned(),
+            None => {
+                return futures::future::err(format_err!("missing URL scheme")).boxed();
+            }
+        };
 
-            let config = this.ssl_connector.configure();
+        async move {
+            let config = ssl_connector.configure()?;
             let dst_str = dst.to_string(); // for error messages
-            let conn = this
-                .http
+            let conn = connector
                 .call(dst)
                 .await
                 .map_err(|err| format_err!("error connecting to {} - {}", dst_str, err))?;
@@ -153,7 +151,7 @@ impl hyper::service::Service<Uri> for HttpsConnector {
             let _ = set_tcp_keepalive(conn.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
 
             if is_https {
-                let mut conn: SslStream<TcpStream> = SslStream::new(config?.into_ssl(&host)?, conn)?;
+                let mut conn: SslStream<TcpStream> = SslStream::new(config.into_ssl(&host)?, conn)?;
                 Pin::new(&mut conn).connect().await?;
                 Ok(MaybeTlsStream::Secured(conn))
             } else {
-- 
2.20.1

[pbs-devel] [RFC proxmox-backup 5/5] HttpsConnector: add proxy support

[Dietmar Maurer]

---
 src/api2/node/apt.rs      |   2 +-
 src/tools/http.rs         | 176 +++++++++++++++++++++++++++++++++-----
 src/tools/subscription.rs |   2 +-
 3 files changed, 156 insertions(+), 24 deletions(-)

diff --git a/src/api2/node/apt.rs b/src/api2/node/apt.rs
index dbdb2019..44b13edd 100644
--- a/src/api2/node/apt.rs
+++ b/src/api2/node/apt.rs
@@ -194,7 +194,7 @@ fn apt_get_changelog(
         bail!("Package '{}' not found", name);
     }
 
-    let mut client = SimpleHttp::new();
+    let mut client = SimpleHttp::new(None); // TODO: pass proxy_config
 
     let changelog_url = &pkg_info[0].change_log_url;
     // FIXME: use 'apt-get changelog' for proxmox packages as well, once repo supports it
diff --git a/src/tools/http.rs b/src/tools/http.rs
index a0dbfd01..6f00d6e0 100644
--- a/src/tools/http.rs
+++ b/src/tools/http.rs
@@ -10,7 +10,14 @@ use hyper::client::{Client, HttpConnector};
 use http::{Request, Response};
 use openssl::ssl::{SslConnector, SslMethod};
 use futures::*;
-use tokio::net::TcpStream;
+use tokio::{
+    io::{
+        AsyncBufReadExt,
+        AsyncWriteExt,
+        BufStream,
+    },
+    net::TcpStream,
+};
 use tokio_openssl::SslStream;
 
 use crate::tools::{
@@ -21,6 +28,14 @@ use crate::tools::{
     },
 };
 
+/// HTTP Proxy Configuration
+#[derive(Clone)]
+pub struct ProxyConfig {
+    pub host: String,
+    pub port: u16,
+    pub force_connect: bool,
+}
+
 /// Asyncrounous HTTP client implementation
 pub struct SimpleHttp {
     client: Client<HttpsConnector, Body>,
@@ -28,18 +43,27 @@ pub struct SimpleHttp {
 
 impl SimpleHttp {
 
-    pub fn new() -> Self {
+    pub fn new(proxy_config: Option<ProxyConfig>) -> Self {
         let ssl_connector = SslConnector::builder(SslMethod::tls()).unwrap().build();
-        Self::with_ssl_connector(ssl_connector)
+        Self::with_ssl_connector(ssl_connector, proxy_config)
     }
 
-    pub fn with_ssl_connector(ssl_connector: SslConnector) -> Self {
+    pub fn with_ssl_connector(ssl_connector: SslConnector, proxy_config: Option<ProxyConfig>) -> Self {
         let connector = HttpConnector::new();
-        let https = HttpsConnector::with_connector(connector, ssl_connector);
+        let mut https = HttpsConnector::with_connector(connector, ssl_connector);
+        if let Some(proxy_config) = proxy_config {
+            https.set_proxy(proxy_config);
+        }
         let client = Client::builder().build(https);
         Self { client }
     }
 
+    pub async fn request(&self, request: Request<Body>) -> Result<Response<Body>, Error> {
+        self.client.request(request)
+            .map_err(Error::from)
+            .await
+    }
+
     pub async fn post(
         &mut self,
         uri: &str,
@@ -106,6 +130,7 @@ impl SimpleHttp {
 pub struct HttpsConnector {
     connector: HttpConnector,
     ssl_connector: Arc<SslConnector>,
+    proxy: Option<ProxyConfig>,
 }
 
 impl HttpsConnector {
@@ -114,8 +139,56 @@ impl HttpsConnector {
         Self {
             connector,
             ssl_connector: Arc::new(ssl_connector),
+            proxy: None,
         }
     }
+
+    pub fn set_proxy(&mut self, proxy: ProxyConfig) {
+        self.proxy = Some(proxy);
+    }
+
+    async fn secure_stream(
+        tcp_stream: TcpStream,
+        ssl_connector: &SslConnector,
+        host: &str,
+    ) -> Result<MaybeTlsStream<TcpStream>, Error> {
+        let config = ssl_connector.configure()?;
+        let mut conn: SslStream<TcpStream> = SslStream::new(config.into_ssl(host)?, tcp_stream)?;
+        Pin::new(&mut conn).connect().await?;
+        Ok(MaybeTlsStream::Secured(conn))
+    }
+
+    async fn parse_connect_status(
+        stream: &mut BufStream<TcpStream>,
+    ) -> Result<(), Error> {
+
+        let mut status_str = String::new();
+
+        // TODO: limit read-length
+
+        if stream.read_line(&mut status_str).await? == 0 {
+            bail!("proxy connect failed - unexpected EOF")
+        }
+
+        if !(status_str.starts_with("HTTP/1.1 200") || status_str.starts_with("HTTP/1.0 200")) {
+            bail!("proxy connect failed - invalid status: {}", status_str)
+        }
+
+        loop {
+            // skip rest until \r\n
+            let mut response = String::new();
+            if stream.read_line(&mut response).await? == 0 {
+                bail!("proxy connect failed - unexpected EOF")
+            }
+            if response.len() > 8192 {
+                bail!("proxy connect failed - long lines in connect rtesponse")
+            }
+            if response == "\r\n" {
+                break;
+            }
+        }
+        Ok(())
+    }
 }
 
 impl hyper::service::Service<Uri> for HttpsConnector {
@@ -124,9 +197,10 @@ impl hyper::service::Service<Uri> for HttpsConnector {
     #[allow(clippy::type_complexity)]
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
-    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // This connector is always ready, but others might not be.
-        Poll::Ready(Ok(()))
+    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.connector
+            .poll_ready(ctx)
+            .map_err(|err| err.into())
     }
 
     fn call(&mut self, dst: Uri) -> Self::Future {
@@ -139,24 +213,82 @@ impl hyper::service::Service<Uri> for HttpsConnector {
                 return futures::future::err(format_err!("missing URL scheme")).boxed();
             }
         };
+        let port = dst.port_u16().unwrap_or(if is_https { 443 } else { 80 });
+
+        if let Some(ref proxy) = self.proxy {
+
+            let use_connect = is_https || proxy.force_connect;
+
+            let proxy_url = format!("{}:{}", proxy.host, proxy.port);
+            let proxy_uri = match Uri::builder()
+                .scheme("http")
+                .authority(proxy_url.as_str())
+                .path_and_query("/")
+                .build()
+            {
+                Ok(uri) => uri,
+                Err(err) => return futures::future::err(err.into()).boxed(),
+            };
+
+            if use_connect {
+                async move {
+
+                    let proxy_stream = connector
+                        .call(proxy_uri)
+                        .await
+                        .map_err(|err| format_err!("error connecting to {} - {}", proxy_url, err))?;
 
-        async move {
-            let config = ssl_connector.configure()?;
-            let dst_str = dst.to_string(); // for error messages
-            let conn = connector
-                .call(dst)
-                .await
-                .map_err(|err| format_err!("error connecting to {} - {}", dst_str, err))?;
+                    let _ = set_tcp_keepalive(proxy_stream.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
 
-            let _ = set_tcp_keepalive(conn.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+                    let mut stream = BufStream::new(proxy_stream);
 
-            if is_https {
-                let mut conn: SslStream<TcpStream> = SslStream::new(config.into_ssl(&host)?, conn)?;
-                Pin::new(&mut conn).connect().await?;
-                Ok(MaybeTlsStream::Secured(conn))
+                    let connect_request = format!(
+                        "CONNECT {0}:{1} HTTP/1.1\r\n\
+                         Host: {0}:{1}\r\n\r\n",
+                        host, port,
+                    );
+
+                    stream.write_all(connect_request.as_bytes()).await?;
+                    stream.flush().await?;
+
+                    Self::parse_connect_status(&mut stream).await?;
+
+                    let tcp_stream = stream.into_inner();
+
+                    if is_https {
+                        Self::secure_stream(tcp_stream, &ssl_connector, &host).await
+                    } else {
+                        Ok(MaybeTlsStream::Normal(tcp_stream))
+                    }
+                }.boxed()
             } else {
-                Ok(MaybeTlsStream::Normal(conn))
+               async move {
+                   let tcp_stream = connector
+                       .call(proxy_uri)
+                       .await
+                       .map_err(|err| format_err!("error connecting to {} - {}", proxy_url, err))?;
+
+                   let _ = set_tcp_keepalive(tcp_stream.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+
+                   Ok(MaybeTlsStream::Proxied(tcp_stream))
+               }.boxed()
             }
-        }.boxed()
+        } else {
+            async move {
+                let dst_str = dst.to_string(); // for error messages
+                let tcp_stream = connector
+                    .call(dst)
+                    .await
+                    .map_err(|err| format_err!("error connecting to {} - {}", dst_str, err))?;
+
+                let _ = set_tcp_keepalive(tcp_stream.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+
+                if is_https {
+                    Self::secure_stream(tcp_stream, &ssl_connector, &host).await
+                } else {
+                    Ok(MaybeTlsStream::Normal(tcp_stream))
+                }
+            }.boxed()
+        }
     }
 }
diff --git a/src/tools/subscription.rs b/src/tools/subscription.rs
index 9a920aee..eaaf0389 100644
--- a/src/tools/subscription.rs
+++ b/src/tools/subscription.rs
@@ -102,7 +102,7 @@ async fn register_subscription(
         "check_token": challenge,
     });
 
-    let mut client = SimpleHttp::new();
+    let mut client = SimpleHttp::new(None); // TODO: pass proxy_config
 
     let uri = "https://shop.maurer-it.com/modules/servers/licensing/verify.php";
     let query = tools::json_object_to_query(params)?;
-- 
2.20.1

Re: [pbs-devel] [RFC proxmox-backup 0/5] proxy support for HttpsConnector

[Wolfgang Bumiller]

looks fine to me, I'll have to rebase the http usage in my acme series
when this is pushed

On Wed, Apr 21, 2021 at 01:16:57PM +0200, Dietmar Maurer wrote:
> This time without using an external crate.
> 
> Note: Secured proxy connections are not supported (rarely used feature?)
> 
> Dietmar Maurer (5):
>   http: rename EitherStream to MaybeTlsStream
>   MaybeTlsStream: implement poll_write_vectored()
>   new http client implementation SimpleHttp (avoid static HTTP_CLIENT)
>   HttpsConnector: code cleanup
>   HttpsConnector: add proxy support
> 
>  src/api2/node/apt.rs      |   8 +-
>  src/tools/async_io.rs     |  90 +++++++---
>  src/tools/http.rs         | 336 +++++++++++++++++++++++++++-----------
>  src/tools/subscription.rs |  10 +-
>  4 files changed, 316 insertions(+), 128 deletions(-)
> 
> -- 
> 2.20.1

[pbs-devel] applied-series: [RFC proxmox-backup 0/5] proxy support for HttpsConnector

[Thomas Lamprecht]

On 21.04.21 13:16, Dietmar Maurer wrote:
> This time without using an external crate.
> 
> Note: Secured proxy connections are not supported (rarely used feature?)
> 
> Dietmar Maurer (5):
>   http: rename EitherStream to MaybeTlsStream
>   MaybeTlsStream: implement poll_write_vectored()
>   new http client implementation SimpleHttp (avoid static HTTP_CLIENT)
>   HttpsConnector: code cleanup
>   HttpsConnector: add proxy support
> 
>  src/api2/node/apt.rs      |   8 +-
>  src/tools/async_io.rs     |  90 +++++++---
>  src/tools/http.rs         | 336 +++++++++++++++++++++++++++-----------
>  src/tools/subscription.rs |  10 +-
>  4 files changed, 316 insertions(+), 128 deletions(-)
> 

LGTM too, so applied, thanks!

@Dylan, please use this and Wolfgang's in-progress node config as new base
for the proxy feature. For the latter you may want for the next version,
or talk/coordinate with Wolfgang as we decided against implementing a direct
serde parser for now.