Add support for http proxies (#1426)

Adds support for HTTP(S) proxies, in addition to SOCKS5 proxies that
we've been supporting for a little while.

Closes svix/monorepo-private#8990.

Author: svix-jplatte

server/Cargo.lock

@@ -74,6 +74,9 @@ form_urlencoded = "1.1.0"
 lapin = "2.1.1"
 sentry = { version = "0.32.2", features = ["tracing"] }
 omniqueue = { git = "https://github.com/svix/omniqueue-rs", rev = "5ae22000e2ea214ba707cac81657f098e5785a76", default-features = false, features = ["in_memory", "rabbitmq-with-message-ids", "redis_cluster"] }
+# Not a well-known author, and no longer gets updates => pinned.
+# Switch to hyper-http-proxy when upgrading hyper to 1.0.
+hyper-proxy = { version = "=0.9.1", default-features = false, features = ["openssl-tls"] }
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemallocator = { version = "0.5", optional = true }

server/svix-server/Cargo.toml

@@ -7,7 +7,6 @@ use figment::{
     providers::{Env, Format, Toml},
     Figment,
 };
-use http::uri::InvalidUri;
 use ipnet::IpNet;
 use serde::{Deserialize, Deserializer};
 use tracing::Level;
@@ -199,28 +198,36 @@ pub struct ConfigurationInner {
 
 #[derive(Clone, Debug, Deserialize)]
 pub struct ProxyConfig {
-    /// SOCKS5 proxy address.
+    /// Proxy address.
     ///
-    /// More proxy types may be supported in the future.
+    /// Currently supported proxy types are:
+    /// - `socks5://`, i.e. a SOCKS5 proxy, with domain name resolution being
+    ///   done before the proxy gets involved
+    /// - `http://` or `https://` proxy, sending HTTP requests to the proxy;
+    ///   both HTTP and HTTPS targets are supported
     #[serde(rename = "proxy_addr")]
     pub addr: ProxyAddr,
 }
 
 #[derive(Clone, Debug)]
-pub struct ProxyAddr {
-    parsed: http::Uri,
+pub enum ProxyAddr {
+    /// A SOCKS5 proxy.
+    Socks5(http::Uri),
+    /// An HTTP / HTTPs proxy.
+    Http(http::Uri),
 }
 
 impl ProxyAddr {
-    pub fn new(raw: impl AsRef<str>) -> Result<Self, InvalidUri> {
-        let parsed = raw.as_ref().parse()?;
-        Ok(Self { parsed })
-    }
-}
-
-impl From<ProxyAddr> for http::Uri {
-    fn from(value: ProxyAddr) -> Self {
-        value.parsed
+    pub fn new(raw: impl Into<String>) -> Result<Self, Box<dyn std::error::Error>> {
+        let raw = raw.into();
+        let parsed: http::Uri = raw.parse()?;
+        match parsed.scheme_str().unwrap_or("") {
+            "socks5" => Ok(Self::Socks5(parsed)),
+            "http" | "https" => Ok(Self::Http(parsed)),
+            _ => Err("Unsupported proxy scheme. \
+                Supported schemes are `socks5://`, `http://` and `https://`."
+                .into()),
+        }
     }
 }
 
@@ -510,7 +517,7 @@ mod tests {
         figment::Jail::expect_with(|jail| {
             jail.set_env("SVIX_QUEUE_TYPE", "memory");
             jail.set_env("SVIX_JWT_SECRET", "x");
-            jail.set_env("SVIX_PROXY_ADDR", "x");
+            jail.set_env("SVIX_PROXY_ADDR", "socks5://127.0.0.1");
 
             let cfg = load().unwrap();
             assert!(cfg.proxy_config.is_some());

server/svix-server/src/cfg.rs

@@ -21,16 +21,17 @@ use hyper::{
     ext::HeaderCaseMap,
     Body, Client, Uri,
 };
-use hyper_openssl::HttpsConnector;
+use hyper_openssl::{HttpsConnector, MaybeHttpsStream};
+use hyper_proxy::{Intercept, Proxy, ProxyConnector as HttpProxyConnector, ProxyStream};
 use hyper_socks2::SocksConnector;
 use ipnet::IpNet;
-use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
+use openssl::ssl::{SslConnector, SslConnectorBuilder, SslMethod, SslVerifyMode};
 use serde::Serialize;
 use thiserror::Error;
 use tokio::{net::TcpStream, sync::Mutex};
 use tower::Service;
 
-use crate::cfg::ProxyConfig;
+use crate::cfg::{ProxyAddr, ProxyConfig};
 
 pub type CaseSensitiveHeaderMap = HashMap<String, HeaderValue>;
 
@@ -55,7 +56,7 @@ pub enum Error {
 
 #[derive(Clone)]
 pub struct WebhookClient {
-    client: Client<HttpsConnector<SvixHttpConnector>, Body>,
+    client: Client<SvixHttpsConnector, Body>,
     whitelist_nets: Arc<Vec<IpNet>>,
 }
 
@@ -64,7 +65,7 @@ impl WebhookClient {
         whitelist_nets: Option<Arc<Vec<IpNet>>>,
         whitelist_names: Option<Arc<Vec<String>>>,
         dangerous_disable_tls_verification: bool,
-        proxy_config: Option<ProxyConfig>,
+        proxy_config: Option<&ProxyConfig>,
     ) -> Self {
         let whitelist_nets = whitelist_nets.unwrap_or_else(|| Arc::new(Vec::new()));
         let whitelist_names = whitelist_names.unwrap_or_else(|| Arc::new(Vec::new()));
@@ -81,8 +82,8 @@ impl WebhookClient {
             ssl.set_verify(SslVerifyMode::NONE);
         }
 
-        let http = SvixHttpConnector::new(http, proxy_config);
-        let https = HttpsConnector::with_connector(http, ssl).expect("HttpsConnector build failed");
+        let https = SvixHttpsConnector::new(http, proxy_config, ssl)
+            .expect("SvixHttpsConnector build failed");
 
         let client: Client<_, hyper::Body> = Client::builder()
             .http1_ignore_invalid_headers_in_responses(true)
@@ -428,54 +429,87 @@ impl RequestBuilder {
     }
 }
 
-/// Plain-HTTP connector that blocks outgoing requests to private IPs with
-/// support for optionally proxying via SOCKS5.
-#[derive(Clone, Debug)]
-enum SvixHttpConnector {
-    Regular(NonLocalHttpConnector),
-    Proxied(SocksConnector<NonLocalHttpConnector>),
+/// HTTP connector that blocks outgoing requests to private IPs with support
+/// for HTTPS and optionally proxying via SOCKS5 or HTTP(S).
+#[derive(Clone)]
+enum SvixHttpsConnector {
+    Regular(HttpsConnector<NonLocalHttpConnector>),
+    Socks5Proxy(HttpsConnector<SocksConnector<NonLocalHttpConnector>>),
+    HttpProxy(HttpProxyConnector<HttpConnector<NonLocalDnsResolver>>),
 }
 
-impl SvixHttpConnector {
-    fn new(inner: NonLocalHttpConnector, proxy_cfg: Option<ProxyConfig>) -> Self {
+impl SvixHttpsConnector {
+    fn new(
+        inner: NonLocalHttpConnector,
+        proxy_cfg: Option<&ProxyConfig>,
+        ssl: SslConnectorBuilder,
+    ) -> Result<Self, Box<dyn std::error::Error>> {
         match proxy_cfg {
-            Some(proxy_cfg) => Self::Proxied(SocksConnector {
-                proxy_addr: proxy_cfg.addr.into(),
-                auth: None,
-                connector: inner,
-            }),
-            None => Self::Regular(inner),
+            Some(proxy_cfg) => match proxy_cfg.addr.clone() {
+                // In the SOCKS5 case, TLS is handled inside of the proxy
+                // TcpStream, by the same code that would do it without a proxy
+                ProxyAddr::Socks5(proxy_addr) => {
+                    let socks = SocksConnector {
+                        proxy_addr,
+                        auth: None,
+                        connector: inner,
+                    };
+                    let socks_https = HttpsConnector::with_connector(socks, ssl)?;
+                    Ok(Self::Socks5Proxy(socks_https))
+                }
+                // In the HTTP proxy case, TLS is handled by the proxy connector
+                ProxyAddr::Http(proxy_addr) => {
+                    let proxy = Proxy::new(Intercept::All, proxy_addr);
+                    Ok(Self::HttpProxy(HttpProxyConnector::from_proxy(
+                        inner, proxy,
+                    )?))
+                }
+            },
+            None => {
+                let https = HttpsConnector::with_connector(inner, ssl)?;
+                Ok(Self::Regular(https))
+            }
         }
     }
 }
 
-impl Service<Uri> for SvixHttpConnector {
-    type Response = TcpStream;
-    type Error = hyper_socks2::Error;
+impl Service<Uri> for SvixHttpsConnector {
+    type Response = ProxyStream<TcpStream>;
+    type Error = Box<dyn std::error::Error + Send + Sync>;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
 
     fn poll_ready(
         &mut self,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Result<(), Self::Error>> {
         match self {
-            Self::Regular(inner) => inner
-                .poll_ready(cx)
-                .map_err(|e| hyper_socks2::Error::Connector(e.into())),
-            Self::Proxied(inner) => inner.poll_ready(cx),
+            Self::Regular(inner) => inner.poll_ready(cx),
+            Self::Socks5Proxy(inner) => inner.poll_ready(cx),
+            Self::HttpProxy(inner) => inner.poll_ready(cx).map_err(Into::into),
         }
     }
 
     fn call(&mut self, req: Uri) -> Self::Future {
+        fn convert_stream(maybe_https: MaybeHttpsStream<TcpStream>) -> ProxyStream<TcpStream> {
+            match maybe_https {
+                MaybeHttpsStream::Http(stream) => ProxyStream::NoProxy(stream),
+                MaybeHttpsStream::Https(stream) => ProxyStream::Secured(stream),
+            }
+        }
+
         match self {
             Self::Regular(inner) => {
                 let fut = inner.call(req);
-                Box::pin(async move {
-                    fut.await
-                        .map_err(|e| hyper_socks2::Error::Connector(e.into()))
-                })
+                Box::pin(async move { Ok(convert_stream(fut.await?)) })
+            }
+            Self::Socks5Proxy(inner) => {
+                let fut = inner.call(req);
+                Box::pin(async move { Ok(convert_stream(fut.await?)) })
+            }
+            Self::HttpProxy(inner) => {
+                let fut = inner.call(req);
+                Box::pin(async move { fut.await.map_err(Into::into) })
             }
-            Self::Proxied(inner) => Box::pin(inner.call(req)),
         }
     }
 }

server/svix-server/src/core/webhook_http_client.rs

@@ -901,7 +901,7 @@ pub async fn queue_handler(
         cfg.whitelist_subnets.clone(),
         Some(Arc::new(vec!["backend".to_owned()])),
         cfg.dangerous_disable_tls_verification,
-        cfg.proxy_config.clone(),
+        cfg.proxy_config.as_ref(),
     );
 
     tokio::spawn(

server/svix-server/src/worker.rs

@@ -17,21 +17,38 @@ async fn test_message_delivery_via_socks5() {
     use crate::utils::start_svix_server_with_cfg;
 
     let mut cfg = get_default_test_config();
-    cfg.proxy_config = Some(proxy_config());
+    cfg.proxy_config = Some(socks_proxy_config());
     let (client, _) = start_svix_server_with_cfg(&cfg).await;
-    run_socks5_test(&client).await;
+    run_proxy_test(&client).await;
 }
 
-fn proxy_config() -> ProxyConfig {
+fn socks_proxy_config() -> ProxyConfig {
     ProxyConfig {
         addr: ProxyAddr::new("socks5://localhost:1080").unwrap(),
     }
 }
 
-async fn run_socks5_test(client: &TestClient) {
+#[ignore] // requires an http proxy to be running
+#[tokio::test]
+async fn test_message_delivery_via_http_proxy() {
+    use crate::utils::start_svix_server_with_cfg;
+
+    let mut cfg = get_default_test_config();
+    cfg.proxy_config = Some(http_proxy_config());
+    let (client, _) = start_svix_server_with_cfg(&cfg).await;
+    run_proxy_test(&client).await;
+}
+
+fn http_proxy_config() -> ProxyConfig {
+    ProxyConfig {
+        addr: ProxyAddr::new("http://localhost:8888").unwrap(),
+    }
+}
+
+async fn run_proxy_test(client: &TestClient) {
     let mut receiver = TestReceiver::start(StatusCode::OK);
 
-    let app_id = create_test_app(client, "kafkaSinkTest").await.unwrap().id;
+    let app_id = create_test_app(client, "proxyTest").await.unwrap().id;
     create_test_endpoint(client, &app_id, &receiver.endpoint)
         .await
         .unwrap();