feat: allow unencrypted HTTP connections to AWS via flag (#3916)

This is required for the test bench.

clap_blocks/src/object_store.rs

@@ -145,6 +145,10 @@ Possible values (case insensitive):
     #[clap(long = "--aws-session-token", env = "AWS_SESSION_TOKEN")]
     pub aws_session_token: Option<String>,
 
+    /// Allow unencrypted HTTP connection to AWS.
+    #[clap(long = "--aws-allow-http", env = "AWS_ALLOW_HTTP")]
+    pub aws_allow_http: bool,
+
     /// When using Google Cloud Storage as the object store, set this to the
     /// path to the JSON file that contains the Google credentials.
     ///
@@ -272,6 +276,7 @@ impl TryFrom<&ObjectStoreConfig> for ObjectStore {
                             endpoint,
                             session_token,
                             config.object_store_connection_limit,
+                            config.aws_allow_http,
                         )
                         .context(InvalidS3ConfigSnafu)
                     }

object_store/src/aws.rs

@@ -372,6 +372,7 @@ impl ObjectStoreApi for AmazonS3 {
 ///
 /// Note do not expose the AmazonS3::new() function to allow it to be
 /// swapped out when the aws feature is not enabled
+#[allow(clippy::too_many_arguments)]
 pub(crate) fn new_s3(
     access_key_id: Option<impl Into<String>>,
     secret_access_key: Option<impl Into<String>>,
@@ -380,6 +381,7 @@ pub(crate) fn new_s3(
     endpoint: Option<impl Into<String>>,
     session_token: Option<impl Into<String>>,
     max_connections: NonZeroUsize,
+    allow_http: bool,
 ) -> Result<AmazonS3> {
     let region = region.into();
     let region: rusoto_core::Region = match endpoint {
@@ -393,23 +395,21 @@ pub(crate) fn new_s3(
     let mut builder = HyperBuilder::default();
     builder.pool_max_idle_per_host(max_connections.get());
 
-    // For testing purposes, allow connections to HTTP endpoints.
-    #[cfg(test)]
-    let connector = hyper_rustls::HttpsConnectorBuilder::new()
-        .with_webpki_roots()
-        .https_or_http()
-        .enable_http1()
-        .enable_http2()
-        .build();
-    // In production, we should not allow plain-text connections when pushing
-    // parquet files to object storage, so only TLS connections are allowed.
-    #[cfg(not(test))]
-    let connector = hyper_rustls::HttpsConnectorBuilder::new()
-        .with_webpki_roots()
-        .https_only()
-        .enable_http1()
-        .enable_http2()
-        .build();
+    let connector = if allow_http {
+        hyper_rustls::HttpsConnectorBuilder::new()
+            .with_webpki_roots()
+            .https_or_http()
+            .enable_http1()
+            .enable_http2()
+            .build()
+    } else {
+        hyper_rustls::HttpsConnectorBuilder::new()
+            .with_webpki_roots()
+            .https_only()
+            .enable_http1()
+            .enable_http2()
+            .build()
+    };
 
     let http_client = rusoto_core::request::HttpClient::from_builder(builder, connector);
 
@@ -452,6 +452,7 @@ pub(crate) fn new_failing_s3() -> Result<AmazonS3> {
         None as Option<&str>,
         None as Option<&str>,
         NonZeroUsize::new(16).unwrap(),
+        true,
     )
 }
 
@@ -767,6 +768,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 
@@ -786,6 +788,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 
@@ -828,6 +831,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 
@@ -864,6 +868,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 
@@ -901,6 +906,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 
@@ -925,6 +931,7 @@ mod tests {
             config.endpoint,
             config.token,
             NonZeroUsize::new(16).unwrap(),
+            true,
         )
         .expect("Valid S3 config");
 

object_store/src/dummy.rs

@@ -83,7 +83,7 @@ impl ObjectStoreApi for DummyObjectStore {
 }
 
 /// Stub when s3 is not configured
-#[allow(dead_code)]
+#[allow(dead_code, clippy::too_many_arguments)]
 pub(crate) fn new_s3(
     _access_key_id: Option<impl Into<String>>,
     _secret_access_key: Option<impl Into<String>>,
@@ -92,6 +92,7 @@ pub(crate) fn new_s3(
     _endpoint: Option<impl Into<String>>,
     _session_token: Option<impl Into<String>>,
     _max_connections: NonZeroUsize,
+    _allow_http: bool,
 ) -> Result<DummyObjectStore> {
     NotSupportedSnafu { name: "aws" }.fail()
 }

object_store/src/lib.rs

@@ -117,6 +117,7 @@ pub struct ObjectStore {
 
 impl ObjectStore {
     /// Configure a connection to Amazon S3.
+    #[allow(clippy::too_many_arguments)]
     pub fn new_amazon_s3(
         access_key_id: Option<impl Into<String>>,
         secret_access_key: Option<impl Into<String>>,
@@ -125,6 +126,7 @@ impl ObjectStore {
         endpoint: Option<impl Into<String>>,
         session_token: Option<impl Into<String>>,
         max_connections: NonZeroUsize,
+        allow_http: bool,
     ) -> Result<Self> {
         let s3 = aws::new_s3(
             access_key_id,
@@ -134,6 +136,7 @@ impl ObjectStore {
             endpoint,
             session_token,
             max_connections,
+            allow_http,
         )?;
         Ok(Self {
             integration: ObjectStoreIntegration::AmazonS3(s3),