feat(s2n-tls-hyper): Allow plain HTTP connections

bindings/rust/s2n-tls-hyper/src/connector.rs

@@ -27,6 +27,7 @@ use tower_service::Service;
 pub struct HttpsConnector<Http, ConnBuilder = Config> {
     http: Http,
     conn_builder: ConnBuilder,
+    insecure_http: bool,
 }
 
 impl<ConnBuilder> HttpsConnector<HttpConnector, ConnBuilder>
@@ -101,7 +102,11 @@ where
     /// configured on `conn_builder` with APIs like
     /// `s2n_tls::config::Builder::set_application_protocol_preference()` will be ignored.
     pub fn builder_with_http(http: Http, conn_builder: ConnBuilder) -> Builder<Http, ConnBuilder> {
-        Builder { http, conn_builder }
+        Builder {
+            http,
+            conn_builder,
+            insecure_http: false,
+        }
     }
 }
 
@@ -110,14 +115,22 @@ where
 pub struct Builder<Http, ConnBuilder> {
     http: Http,
     conn_builder: ConnBuilder,
+    insecure_http: bool,
 }
 
 impl<Http, ConnBuilder> Builder<Http, ConnBuilder> {
+    /// Allows communication with insecure HTTP endpoints in addition to secure HTTPS endpoints.
+    pub fn with_insecure_http(&mut self) -> &mut Self {
+        self.insecure_http = true;
+        self
+    }
+
     /// Builds a new `HttpsConnector`.
     pub fn build(self) -> HttpsConnector<Http, ConnBuilder> {
         HttpsConnector {
             http: self.http,
             conn_builder: self.conn_builder,
+            insecure_http: self.insecure_http,
         }
     }
 }
@@ -155,10 +168,18 @@ where
     }
 
     fn call(&mut self, req: Uri) -> Self::Future {
-        // Currently, the only supported stream type is TLS. If the application attempts to
-        // negotiate HTTP over plain TCP, return an error.
-        if req.scheme() == Some(&http::uri::Scheme::HTTP) {
-            return Box::pin(async move { Err(Error::InvalidScheme) });
+        match req.scheme() {
+            Some(scheme) if scheme == &http::uri::Scheme::HTTPS => (),
+            Some(scheme) if scheme == &http::uri::Scheme::HTTP && self.insecure_http => {
+                let call = self.http.call(req);
+                return Box::pin(async move {
+                    let tcp = call.await.map_err(|e| Error::HttpError(e.into()))?;
+                    Ok(MaybeHttpsStream::Http(tcp))
+                });
+            }
+            _ => {
+                return Box::pin(async move { Err(Error::InvalidScheme) });
+            }
         }
 
         // Attempt to negotiate HTTP/2 by including it in the ALPN extension. Other supported HTTP
@@ -235,15 +256,16 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_unsecure_http() -> Result<(), Box<dyn StdError>> {
+    async fn test_invalid_scheme() -> Result<(), Box<dyn StdError>> {
         let connector = HttpsConnector::new(Config::default());
         let client: Client<_, Empty<Bytes>> =
             Client::builder(TokioExecutor::new()).build(connector);
 
-        let uri = Uri::from_str("http://www.amazon.com")?;
+        // Attempt to make a request with an arbitrary invalid scheme.
+        let uri = Uri::from_str("notascheme://www.amazon.com")?;
         let error = client.get(uri).await.unwrap_err();
 
-        // Ensure that an InvalidScheme error is returned when HTTP over TCP is attempted.
+        // Ensure that an InvalidScheme error is returned.
         let error = error.source().unwrap().downcast_ref::<Error>().unwrap();
         assert!(matches!(error, Error::InvalidScheme));
 

bindings/rust/s2n-tls-hyper/src/stream.rs

@@ -15,11 +15,8 @@ use std::{
 };
 
 /// `MaybeHttpsStream` is a wrapper over a hyper TCP stream, `Transport`, allowing for TLS to be
-/// negotiated over the TCP stream.
-///
-/// While not currently implemented, the `MaybeHttpsStream` enum will provide an `Http` type
-/// corresponding to the plain TCP stream, allowing for HTTP to be negotiated in addition to HTTPS
-/// when the HTTP scheme is used.
+/// negotiated over the TCP stream via the `Https` type. The `Http` type bypasses TLS to optionally
+/// allow for communication with HTTP endpoints over plain TCP.
 ///
 /// This struct is used to implement `tower_service::Service` for `HttpsConnector`, and shouldn't
 /// need to be used directly.
@@ -38,6 +35,7 @@ where
     // traits to tokio's. This allows the `Read` and `Write` implementations for `MaybeHttpsStream`
     // to simply call the `TokioIo` `poll` functions.
     Https(TokioIo<TlsStream<TokioIo<Transport>, Builder::Output>>),
+    Http(Transport),
 }
 
 impl<Transport, Builder> HyperConnection for MaybeHttpsStream<Transport, Builder>
@@ -48,7 +46,7 @@ where
 {
     fn connected(&self) -> Connected {
         match self {
-            MaybeHttpsStream::Https(stream) => {
+            Self::Https(stream) => {
                 let connected = stream.inner().get_ref().connected();
                 let conn = stream.inner().as_ref();
                 match conn.application_protocol() {
@@ -57,6 +55,7 @@ where
                     _ => connected,
                 }
             }
+            Self::Http(stream) => stream.connected(),
         }
     }
 }
@@ -74,6 +73,7 @@ where
     ) -> Poll<Result<(), Error>> {
         match Pin::get_mut(self) {
             Self::Https(stream) => Pin::new(stream).poll_read(cx, buf),
+            Self::Http(stream) => Pin::new(stream).poll_read(cx, buf),
         }
     }
 }
@@ -91,18 +91,21 @@ where
     ) -> Poll<Result<usize, Error>> {
         match Pin::get_mut(self) {
             Self::Https(stream) => Pin::new(stream).poll_write(cx, buf),
+            Self::Http(stream) => Pin::new(stream).poll_write(cx, buf),
         }
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
         match Pin::get_mut(self) {
-            MaybeHttpsStream::Https(stream) => Pin::new(stream).poll_flush(cx),
+            Self::Https(stream) => Pin::new(stream).poll_flush(cx),
+            Self::Http(stream) => Pin::new(stream).poll_flush(cx),
         }
     }
 
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
         match Pin::get_mut(self) {
-            MaybeHttpsStream::Https(stream) => Pin::new(stream).poll_shutdown(cx),
+            Self::Https(stream) => Pin::new(stream).poll_shutdown(cx),
+            Self::Http(stream) => Pin::new(stream).poll_shutdown(cx),
         }
     }
 }

bindings/rust/s2n-tls-hyper/tests/common/echo.rs

@@ -11,7 +11,7 @@ use s2n_tls_tokio::TlsAcceptor;
 use std::{error::Error, future::Future};
 use tokio::net::TcpListener;
 
-async fn echo(
+pub async fn echo(
     req: Request<hyper::body::Incoming>,
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
     Ok(Response::new(req.into_body().boxed()))

bindings/rust/s2n-tls-hyper/tests/http.rs

@@ -1,11 +1,14 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::common::echo::echo;
 use crate::common::InsecureAcceptAllCertificatesHandler;
 use bytes::Bytes;
 use common::echo::serve_echo;
 use http::{Method, Request, Uri, Version};
 use http_body_util::{BodyExt, Empty, Full};
+use hyper::service::service_fn;
+use hyper_util::rt::TokioIo;
 use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use s2n_tls::{
     callbacks::{ClientHelloCallback, ConnectionFuture},
@@ -14,8 +17,12 @@ use s2n_tls::{
     security::DEFAULT_TLS13,
 };
 use s2n_tls_hyper::connector::HttpsConnector;
+use s2n_tls_hyper::error;
 use std::{error::Error, pin::Pin, str::FromStr};
-use tokio::{net::TcpListener, task::JoinHandle};
+use tokio::{
+    net::TcpListener,
+    task::{JoinHandle, JoinSet},
+};
 
 pub mod common;
 
@@ -330,3 +337,65 @@ async fn config_alpn_ignored() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn insecure_http() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let addr = listener.local_addr()?;
+
+    let mut tasks: JoinSet<Result<(), Box<dyn Error + Send + Sync>>> = JoinSet::new();
+    tasks.spawn(async move {
+        // Listen for HTTP requests on a plain TCP stream.
+        let (tcp_stream, _) = listener.accept().await.unwrap();
+        let server = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
+        server
+            .serve_connection(TokioIo::new(tcp_stream), service_fn(echo))
+            .await?;
+
+        Ok(())
+    });
+
+    tasks.spawn(async move {
+        for enable_insecure_http in [false, true] {
+            let connector = {
+                let config = common::config()?.build()?;
+                let mut builder = HttpsConnector::builder(config);
+                if enable_insecure_http {
+                    builder.with_insecure_http();
+                }
+                builder.build()
+            };
+
+            let client: Client<_, Empty<Bytes>> =
+                Client::builder(TokioExecutor::new()).build(connector);
+            let uri = Uri::from_str(format!("http://127.0.0.1:{}", addr.port()).as_str())?;
+            let response = client.get(uri).await;
+
+            if enable_insecure_http {
+                // If insecure HTTP is enabled, the request should succeed.
+                let response = response.unwrap();
+                assert_eq!(response.status(), 200);
+            } else {
+                // By default, insecure HTTP is disabled, and the request should error.
+                let error = response.unwrap_err();
+
+                // Ensure an InvalidScheme error is produced.
+                let error = error
+                    .source()
+                    .unwrap()
+                    .downcast_ref::<error::Error>()
+                    .unwrap();
+                assert!(matches!(error, error::Error::InvalidScheme));
+                assert!(!error.to_string().is_empty());
+            }
+        }
+
+        Ok(())
+    });
+
+    while let Some(res) = tasks.join_next().await {
+        res.unwrap()?;
+    }
+
+    Ok(())
+}