add rustls support for actix-http and awc

actix-http/Cargo.toml

@@ -27,6 +27,7 @@ default = []
 
 # openssl
 ssl = ["openssl", "actix-connect/ssl"]
+rust-tls = ["rustls", "webpki-roots", "actix-connect/rust-tls"]
 
 # brotli encoding, requires c compiler
 brotli = ["brotli2"]
@@ -93,6 +94,8 @@ flate2 = { version="1.0.7", optional = true, default-features = false }
 # optional deps
 failure = { version = "0.1.5", optional = true }
 openssl = { version="0.10", optional = true }
+rustls = { version = "0.15.2", optional = true }
+webpki-roots = { version = "0.16", optional = true }
 chrono = "0.4.6"
 
 [dev-dependencies]

actix-http/src/client/connector.rs

@@ -19,7 +19,14 @@ use super::Connect;
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslConnector;
 
-#[cfg(not(feature = "ssl"))]
+#[cfg(feature = "rust-tls")]
+use rustls::{Session, ClientConfig};
+#[cfg(feature = "rust-tls")]
+use std::sync::Arc;
+#[cfg(feature = "rust-tls")]
+type SslConnector = Arc<ClientConfig>;
+
+#[cfg(all(not(feature = "ssl"), not(feature = "rust-tls")))]
 type SslConnector = ();
 
 /// Manages http client network connectivity
@@ -67,7 +74,15 @@ impl Connector<(), ()> {
                     .map_err(|e| error!("Can not set alpn protocol: {:?}", e));
                 ssl.build()
             }
-            #[cfg(not(feature = "ssl"))]
+            #[cfg(feature = "rust-tls")]
+            {
+                let protos = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
+                let mut config = ClientConfig::new();
+                config.set_protocols(&protos);
+                config.root_store.add_server_trust_anchors(&webpki_roots::TLS_SERVER_ROOTS);
+                Arc::new(config)
+            }
+            #[cfg(all(not(feature = "ssl"), not(feature = "rust-tls")))]
             {}
         };
 
@@ -182,7 +197,7 @@ where
         self,
     ) -> impl Service<Request = Connect, Response = impl Connection, Error = ConnectError>
                  + Clone {
-        #[cfg(not(feature = "ssl"))]
+        #[cfg(all(not(feature = "ssl"), not(feature = "rust-tls")))]
         {
             let connector = TimeoutService::new(
                 self.timeout,
@@ -255,6 +270,71 @@ where
                 TimeoutError::Timeout => ConnectError::Timeout,
             });
 
+            connect_impl::InnerConnector {
+                tcp_pool: ConnectionPool::new(
+                    tcp_service,
+                    self.conn_lifetime,
+                    self.conn_keep_alive,
+                    None,
+                    self.limit,
+                ),
+                ssl_pool: ConnectionPool::new(
+                    ssl_service,
+                    self.conn_lifetime,
+                    self.conn_keep_alive,
+                    Some(self.disconnect_timeout),
+                    self.limit,
+                ),
+            }
+        }
+        #[cfg(feature = "rust-tls")]
+        {
+            const H2: &[u8] = b"h2";
+            use actix_connect::ssl::RustlsConnector;
+
+            let ssl_service = TimeoutService::new(
+                self.timeout,
+                apply_fn(self.connector.clone(), |msg: Connect, srv| {
+                    srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
+                })
+                .map_err(ConnectError::from)
+                .and_then(
+                    RustlsConnector::service(self.ssl)
+                        .map_err(ConnectError::from)
+                        .map(|stream| {
+                            let sock = stream.into_parts().0;
+                            let h2 = sock
+                                .get_ref()
+                                .1
+                                .get_alpn_protocol()
+                                .map(|protos| protos.windows(2).any(|w| w == H2))
+                                .unwrap_or(false);
+                            if h2 {
+                                (sock, Protocol::Http2)
+                            } else {
+                                (sock, Protocol::Http1)
+                            }
+                        }),
+                ),
+            )
+            .map_err(|e| match e {
+                TimeoutError::Service(e) => e,
+                TimeoutError::Timeout => ConnectError::Timeout,
+            });
+
+            let tcp_service = TimeoutService::new(
+                self.timeout,
+                apply_fn(self.connector.clone(), |msg: Connect, srv| {
+                    srv.call(TcpConnect::new(msg.uri).set_addr(msg.addr))
+                })
+                .map_err(ConnectError::from)
+                .map(|stream| (stream.into_parts().0, Protocol::Http1)),
+            )
+            .map_err(|e| match e {
+                TimeoutError::Service(e) => e,
+                TimeoutError::Timeout => ConnectError::Timeout,
+            });
+
             connect_impl::InnerConnector {
                 tcp_pool: ConnectionPool::new(
                     tcp_service,
@@ -275,7 +355,7 @@ where
     }
 }
 
-#[cfg(not(feature = "ssl"))]
+#[cfg(all(not(feature = "ssl"), not(feature = "rust-tls")))]
 mod connect_impl {
     use futures::future::{err, Either, FutureResult};
     use futures::Poll;
@@ -479,3 +559,146 @@ mod connect_impl {
         }
     }
 }
+
+#[cfg(feature = "rust-tls")]
+mod connect_impl {
+    use std::marker::PhantomData;
+
+    use futures::future::{Either, FutureResult};
+    use futures::{Async, Future, Poll};
+
+    use super::*;
+    use crate::client::connection::EitherConnection;
+
+    pub(crate) struct InnerConnector<T1, T2, Io1, Io2>
+    where
+        Io1: AsyncRead + AsyncWrite + 'static,
+        Io2: AsyncRead + AsyncWrite + 'static,
+        T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>,
+        T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>,
+    {
+        pub(crate) tcp_pool: ConnectionPool<T1, Io1>,
+        pub(crate) ssl_pool: ConnectionPool<T2, Io2>,
+    }
+
+    impl<T1, T2, Io1, Io2> Clone for InnerConnector<T1, T2, Io1, Io2>
+    where
+        Io1: AsyncRead + AsyncWrite + 'static,
+        Io2: AsyncRead + AsyncWrite + 'static,
+        T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+        T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+    {
+        fn clone(&self) -> Self {
+            InnerConnector {
+                tcp_pool: self.tcp_pool.clone(),
+                ssl_pool: self.ssl_pool.clone(),
+            }
+        }
+    }
+
+    impl<T1, T2, Io1, Io2> Service for InnerConnector<T1, T2, Io1, Io2>
+    where
+        Io1: AsyncRead + AsyncWrite + 'static,
+        Io2: AsyncRead + AsyncWrite + 'static,
+        T1: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+        T2: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+    {
+        type Request = Connect;
+        type Response = EitherConnection<Io1, Io2>;
+        type Error = ConnectError;
+        type Future = Either<
+            FutureResult<Self::Response, Self::Error>,
+            Either<
+                InnerConnectorResponseA<T1, Io1, Io2>,
+                InnerConnectorResponseB<T2, Io1, Io2>,
+            >,
+        >;
+
+        fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+            self.tcp_pool.poll_ready()
+        }
+
+        fn call(&mut self, req: Connect) -> Self::Future {
+            match req.uri.scheme_str() {
+                Some("https") | Some("wss") => {
+                    Either::B(Either::B(InnerConnectorResponseB {
+                        fut: self.ssl_pool.call(req),
+                        _t: PhantomData,
+                    }))
+                }
+                _ => Either::B(Either::A(InnerConnectorResponseA {
+                    fut: self.tcp_pool.call(req),
+                    _t: PhantomData,
+                })),
+            }
+        }
+    }
+
+    pub(crate) struct InnerConnectorResponseA<T, Io1, Io2>
+    where
+        Io1: AsyncRead + AsyncWrite + 'static,
+        T: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+    {
+        fut: <ConnectionPool<T, Io1> as Service>::Future,
+        _t: PhantomData<Io2>,
+    }
+
+    impl<T, Io1, Io2> Future for InnerConnectorResponseA<T, Io1, Io2>
+    where
+        T: Service<Request = Connect, Response = (Io1, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+        Io1: AsyncRead + AsyncWrite + 'static,
+        Io2: AsyncRead + AsyncWrite + 'static,
+    {
+        type Item = EitherConnection<Io1, Io2>;
+        type Error = ConnectError;
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            match self.fut.poll()? {
+                Async::NotReady => Ok(Async::NotReady),
+                Async::Ready(res) => Ok(Async::Ready(EitherConnection::A(res))),
+            }
+        }
+    }
+
+    pub(crate) struct InnerConnectorResponseB<T, Io1, Io2>
+    where
+        Io2: AsyncRead + AsyncWrite + 'static,
+        T: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+    {
+        fut: <ConnectionPool<T, Io2> as Service>::Future,
+        _t: PhantomData<Io1>,
+    }
+
+    impl<T, Io1, Io2> Future for InnerConnectorResponseB<T, Io1, Io2>
+    where
+        T: Service<Request = Connect, Response = (Io2, Protocol), Error = ConnectError>
+            + Clone
+            + 'static,
+        Io1: AsyncRead + AsyncWrite + 'static,
+        Io2: AsyncRead + AsyncWrite + 'static,
+    {
+        type Item = EitherConnection<Io1, Io2>;
+        type Error = ConnectError;
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            match self.fut.poll()? {
+                Async::NotReady => Ok(Async::NotReady),
+                Async::Ready(res) => Ok(Async::Ready(EitherConnection::B(res))),
+            }
+        }
+    }
+}

awc/Cargo.toml

@@ -29,6 +29,9 @@ default = ["brotli", "flate2-zlib"]
 # openssl
 ssl = ["openssl", "actix-http/ssl"]
 
+# rustls
+rust-tls = ["rustls", "actix-http/rust-tls"]
+
 # brotli encoding, requires c compiler
 brotli = ["actix-http/brotli"]
 
@@ -55,6 +58,7 @@ serde_json = "1.0"
 serde_urlencoded = "0.5.3"
 tokio-timer = "0.2.8"
 openssl = { version="0.10", optional = true }
+rustls = { version = "0.15.2", optional = true }
 
 [dev-dependencies]
 actix-rt = "0.2.2"