feat: add support for HTTPS_PROXY env var (#665)

* feat: add support for HTTPS_PROXY env var * fix: hide features behind s3 feature gate * fix: remove panic when unwrapping uri
delta-io · Jul 5, 2022 · 9bfa8ab · 9bfa8ab
1 parent 0284cd1
commit 9bfa8ab
Show file tree

Hide file tree

Showing 4 changed files with 203 additions and 14 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
@@ -44,6 +44,9 @@ rusoto_s3 = { version = "0.48", default-features = false, optional = true }
 rusoto_sts = { version = "0.48", default-features = false, optional = true }
 rusoto_dynamodb = { version = "0.48", default-features = false, optional = true }
 maplit = { version = "1", optional = true }
+hyper = {  version = "0.14.19", default-features = false, optional = true}
+hyper-rustls = {  version = "0.23.0", default-features = false, optional = true, features = ["http2", "rustls-native-certs", "tokio-runtime"] } 
+hyper-proxy = {  version = "0.9.1", default-features = false, optional = true, features = ["rustls"] }
 
 # Glue
 rusoto_glue = { version = "0.48", default-features = false, optional = true }
@@ -91,6 +94,9 @@ s3 = [
     "rusoto_dynamodb/native-tls",
     "maplit",
     "dynamodb_lock/native-tls",
+    "hyper",
+    "hyper-rustls",
+    "hyper-proxy"
 ]
 s3-rustls = [
     "rusoto_core/rustls",
@@ -100,6 +106,9 @@ s3-rustls = [
     "rusoto_dynamodb/rustls",
     "maplit",
     "dynamodb_lock/rustls",
+    "hyper",
+    "hyper-rustls",
+    "hyper-proxy"
 ]
 gcs = ["async-stream", "tame-gcs", "tame-oauth", "reqwest"]
 glue = ["s3", "rusoto_glue"]

diff --git a/rust/src/storage/mod.rs b/rust/src/storage/mod.rs
@@ -1,5 +1,7 @@
 //! Object storage backend abstraction layer for Delta Table transaction logs and data
 
+#[cfg(any(feature = "s3", feature = "s3-rustls"))]
+use hyper::http::uri::InvalidUri;
 use std::fmt::Debug;
 use std::pin::Pin;
 
@@ -408,6 +410,15 @@ pub enum StorageError {
         /// Uri error details when the URI is invalid.
         source: UriError,
     },
+
+    /// Error returned when the URI is invalid.
+    #[cfg(any(feature = "s3", feature = "s3-rustls"))]
+    #[error("Invalid URI parsing")]
+    ParsingUri {
+        #[from]
+        /// Uri error details when the URI parsing is invalid.
+        source: InvalidUri,
+    },
 }
 
 impl StorageError {

diff --git a/rust/src/storage/s3/mod.rs b/rust/src/storage/s3/mod.rs
@@ -6,6 +6,7 @@ use std::{fmt, pin::Pin};
 
 use chrono::{DateTime, FixedOffset, Utc};
 use futures::Stream;
+
 use log::debug;
 use rusoto_core::{HttpClient, HttpConfig, Region, RusotoError};
 use rusoto_credential::AutoRefreshingProvider;
@@ -28,6 +29,12 @@ use uuid::Uuid;
 
 use dynamodb_lock::{LockClient, LockItem, DEFAULT_MAX_RETRY_ACQUIRE_LOCK_ATTEMPTS};
 
+use hyper::client::HttpConnector;
+use hyper_proxy::{Intercept, Proxy, ProxyConnector};
+use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
+
+use std::env;
+
 /// Lock data which stores an attempt to rename `source` into `destination`
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct LockData {
@@ -416,10 +423,32 @@ fn get_sts_assume_role_provider(
     Ok(AutoRefreshingProvider::new(provider)?)
 }
 
-fn create_http_client(pool_idle_timeout: Duration) -> Result<HttpClient, StorageError> {
+fn create_http_client(
+    pool_idle_timeout: Duration,
+) -> Result<HttpClient<ProxyConnector<HttpsConnector<HttpConnector>>>, StorageError> {
     let mut config = HttpConfig::new();
     config.pool_idle_timeout(pool_idle_timeout);
-    Ok(HttpClient::new_with_config(config)?)
+    let https_connector = HttpsConnectorBuilder::new()
+        .with_native_roots()
+        .https_or_http()
+        .enable_http2()
+        .build();
+    match env::var("HTTPS_PROXY") {
+        Ok(proxy_uri) => {
+            let proxy = Proxy::new(Intercept::All, proxy_uri.parse()?);
+            let proxy_connector = ProxyConnector::from_proxy(https_connector, proxy)?;
+            Ok(HttpClient::<ProxyConnector<HttpsConnector<HttpConnector>>>::from_connector_with_config(
+                proxy_connector,
+                config,
+            ))
+        }
+        Err(_) => Ok(
+            HttpClient::<ProxyConnector<HttpsConnector<HttpConnector>>>::from_connector_with_config(
+                ProxyConnector::new(https_connector)?,
+                config,
+            ),
+        ),
+    }
 }
 
 fn get_web_identity_provider() -> Result<AutoRefreshingProvider<WebIdentityProvider>, StorageError>