Proxy: Allow non-Conduit-bound TLS and non-TLS through. (#1203)

On the server (accept) side of TLS, if the traffic isn't targetting the
proxy (as determined by the TLS ClientHello SNI), or if the traffic
isn't TLS, then pass it through.

Signed-off-by: Brian Smith <brian@briansmith.org>

proxy/src/connection.rs

@@ -30,6 +30,18 @@ pub fn connect(addr: &SocketAddr,
     }
 }
 
+/// A server socket that is in the process of conditionally upgrading to TLS.
+enum ConditionallyUpgradeServerToTls {
+    Plaintext(Option<ConditionallyUpgradeServerToTlsInner>),
+    UpgradeToTls(tls::UpgradeServerToTls),
+}
+
+struct ConditionallyUpgradeServerToTlsInner {
+    socket: TcpStream,
+    tls: tls::ConnectionConfig<tls::ServerConfig>,
+    peek_buf: BytesMut,
+}
+
 /// A socket that is in the process of connecting.
 pub enum Connecting {
     Plaintext {
@@ -154,19 +166,10 @@ impl BoundPort {
                         Conditional::None(r) => Conditional::None(*r),
                     };
                     let conn = match tls {
-                        Conditional::Some(config) => {
-                            // TODO: use `config.identity` to differentiate
-                            // between TLS that the proxy should terminate vs.
-                            // TLS that should be passed through.
-                            let f = tls::Connection::accept(socket, config.config)
-                                .map(move |tls| Connection::tls(tls));
-                            Either::A(f)
-                        },
-                        Conditional::None(why_no_tls) => {
-                            let f = future::ok(socket)
-                                .map(move |plain| Connection::plain(plain, why_no_tls));
-                            Either::B(f)
-                        },
+                        Conditional::Some(tls) =>
+                            Either::A(ConditionallyUpgradeServerToTls::new(socket, tls)),
+                        Conditional::None(why_no_tls) =>
+                            Either::B(future::ok(Connection::plain(socket, why_no_tls))),
                     };
                     conn.map(move |conn| (conn, remote_addr))
                 })
@@ -186,6 +189,61 @@ impl BoundPort {
     }
 }
 
+// ===== impl ConditionallyUpgradeServerToTls =====
+
+impl ConditionallyUpgradeServerToTls {
+    fn new(socket: TcpStream, tls: tls::ConnectionConfig<tls::ServerConfig>) -> Self {
+        ConditionallyUpgradeServerToTls::Plaintext(Some(ConditionallyUpgradeServerToTlsInner {
+            socket,
+            tls,
+            peek_buf: BytesMut::with_capacity(8192),
+        }))
+    }
+}
+
+impl Future for ConditionallyUpgradeServerToTls {
+    type Item = Connection;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            *self = match self {
+                ConditionallyUpgradeServerToTls::Plaintext(ref mut inner) => {
+                    let r = {
+                        let inner = inner.as_mut().unwrap();
+                        try_ready!(inner.socket.read_buf(&mut inner.peek_buf));
+                        tls::conditional_accept::match_client_hello(
+                            inner.peek_buf.as_ref(), &inner.tls.identity)
+                    };
+                    match r {
+                        tls::conditional_accept::Match::Matched => {
+                            trace!("upgrading accepted connection to TLS");
+                            let inner = inner.take().expect("Polled after ready");
+                            let upgrade_to_tls = tls::Connection::accept(
+                                inner.socket, inner.peek_buf.freeze(), inner.tls.config);
+                            ConditionallyUpgradeServerToTls::UpgradeToTls(upgrade_to_tls)
+                        },
+                        tls::conditional_accept::Match::NotMatched => {
+                            trace!("passing through accepted connection without TLS");
+                            let inner = inner.take().expect("Polled after ready");
+                            let conn = Connection::plain_with_peek_buf(
+                                inner.socket, inner.peek_buf, tls::ReasonForNoTls::NotProxyTls);
+                            return Ok(Async::Ready(conn));
+                        },
+                        tls::conditional_accept::Match::Incomplete => {
+                            return Ok(Async::NotReady);
+                        },
+                    }
+                },
+                ConditionallyUpgradeServerToTls::UpgradeToTls(upgrading) => {
+                    let tls_stream = try_ready!(upgrading.poll());
+                    return Ok(Async::Ready(Connection::tls(BoxedIo::new(tls_stream))));
+                }
+            }
+        }
+    }
+}
+
 // ===== impl Connecting =====
 
 impl Future for Connecting {
@@ -211,7 +269,7 @@ impl Future for Connecting {
                 },
                 Connecting::UpgradeToTls(upgrading) => {
                     let tls_stream = try_ready!(upgrading.poll());
-                    return Ok(Async::Ready(Connection::tls(tls_stream)));
+                    return Ok(Async::Ready(Connection::tls(BoxedIo::new(tls_stream))));
                 },
             };
         }
@@ -222,16 +280,22 @@ impl Future for Connecting {
 
 impl Connection {
     fn plain(io: TcpStream, why_no_tls: tls::ReasonForNoTls) -> Self {
+        Self::plain_with_peek_buf(io, BytesMut::new(), why_no_tls)
+    }
+
+    fn plain_with_peek_buf(io: TcpStream, peek_buf: BytesMut, why_no_tls: tls::ReasonForNoTls)
+        -> Self
+    {
         Connection {
             io: BoxedIo::new(io),
-            peek_buf: BytesMut::new(),
+            peek_buf,
             tls_status: Conditional::None(why_no_tls),
         }
     }
 
-    fn tls<S: tls::Session + std::fmt::Debug + 'static>(tls: tls::Connection<S>) -> Self {
+    fn tls(io: BoxedIo) -> Self {
         Connection {
-            io: BoxedIo::new(tls),
+            io: io,
             peek_buf: BytesMut::new(),
             tls_status: Conditional::Some(()),
         }

proxy/src/telemetry/metrics/labels.rs

@@ -383,7 +383,8 @@ impl fmt::Display for ctx::transport::TlsStatus {
             Conditional::None(tls::ReasonForNoTls::NoConfig) => f.pad(",tls=\"no_config\""),
             Conditional::None(tls::ReasonForNoTls::Disabled) |
             Conditional::None(tls::ReasonForNoTls::InternalTraffic) |
-            Conditional::None(tls::ReasonForNoTls::NoIdentity(_)) => Ok(()),
+            Conditional::None(tls::ReasonForNoTls::NoIdentity(_)) |
+            Conditional::None(tls::ReasonForNoTls::NotProxyTls) => Ok(()),
         }
     }
 }

proxy/src/transport/prefixed.rs

@@ -1,5 +1,3 @@
-#![allow(dead_code)] // TODO: Actually use this.
-
 use std::{cmp, fmt::Debug, io, net::SocketAddr};
 
 use super::io::internal::Io;

proxy/src/transport/tls/conditional_accept.rs

@@ -1,5 +1,3 @@
-#![allow(dead_code)] // TODO: Use this.
-
 use super::{Identity, untrusted};
 
 #[derive(Debug, Eq, PartialEq)]

proxy/src/transport/tls/config.rs

@@ -96,6 +96,10 @@ pub enum ReasonForNoTls {
 
     /// The connection is between the proxy and the service
     InternalTraffic,
+
+    /// The connection isn't TLS or it is TLS but not intended to be handled
+    /// by the proxy.
+    NotProxyTls,
 }
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]

proxy/src/transport/tls/connection.rs

@@ -6,7 +6,7 @@ use futures::Future;
 use tokio::prelude::*;
 use tokio::net::TcpStream;
 
-use transport::{AddrInfo, io::internal::Io};
+use transport::{AddrInfo, io::internal::Io, prefixed::Prefixed};
 
 use super::{
     identity::Identity,
@@ -19,22 +19,24 @@ use super::{
 use std::fmt::Debug;
 
 pub use self::rustls::Session;
+use bytes::Bytes;
 
 // In theory we could replace `TcpStream` with `Io`. However, it is likely that
 // in the future we'll need to do things specific to `TcpStream`, so optimize
 // for that unless/until there is some benefit to doing otherwise.
 #[derive(Debug)]
-pub struct Connection<S: Session>(TlsStream<TcpStream, S>);
+pub struct Connection<S, C>(TlsStream<S, C>) where S: Debug, C: Debug;
 
-pub struct UpgradeToTls<S, F>(F)
-    where S: Session,
-          F: Future<Item = TlsStream<TcpStream, S>, Error = io::Error>;
+pub struct UpgradeToTls<S, C, F>(F)
+    where C: Session,
+          F: Future<Item = TlsStream<S, C>, Error = io::Error>;
 
-impl<S, F> Future for UpgradeToTls<S, F>
-    where S: Session,
-          F: Future<Item = TlsStream<TcpStream, S>, Error = io::Error>
+impl<C, S, F> Future for UpgradeToTls<S, C, F>
+    where S: Debug,
+          C: Session + Debug,
+          F: Future<Item = TlsStream<S, C>, Error = io::Error>
 {
-    type Item = Connection<S>;
+    type Item = Connection<S, C>;
     type Error = io::Error;
 
     fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> {
@@ -44,12 +46,15 @@ impl<S, F> Future for UpgradeToTls<S, F>
 }
 
 pub type UpgradeClientToTls =
-    UpgradeToTls<rustls::ClientSession, tokio_rustls::ConnectAsync<TcpStream>>;
+    UpgradeToTls<TcpStream, rustls::ClientSession, tokio_rustls::ConnectAsync<TcpStream>>;
 
 pub type UpgradeServerToTls =
-    UpgradeToTls<rustls::ServerSession, tokio_rustls::AcceptAsync<TcpStream>>;
+    UpgradeToTls<
+        Prefixed<TcpStream>,
+        rustls::ServerSession,
+        tokio_rustls::AcceptAsync<Prefixed<TcpStream>>>;
 
-impl Connection<rustls::ClientSession> {
+impl Connection<TcpStream, rustls::ClientSession> {
     pub fn connect(socket: TcpStream, identity: &Identity, ClientConfig(config): ClientConfig)
         -> UpgradeClientToTls
     {
@@ -57,26 +62,36 @@ impl Connection<rustls::ClientSession> {
     }
 }
 
-impl Connection<rustls::ServerSession> {
-    pub fn accept(socket: TcpStream, ServerConfig(config): ServerConfig) -> UpgradeServerToTls
+impl Connection<Prefixed<TcpStream>, rustls::ServerSession> {
+    pub fn accept(socket: TcpStream, prefix: Bytes, ServerConfig(config): ServerConfig)
+                  -> UpgradeServerToTls
     {
-        UpgradeToTls(config.accept_async(socket))
+        UpgradeToTls(config.accept_async(Prefixed::new(prefix, socket)))
     }
 }
 
-impl<S: Session> io::Read for Connection<S> {
+impl<S, C> io::Read for Connection<S, C>
+    where S: Debug + io::Read + io::Write,
+          C: Session + Debug
+{
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         self.0.read(buf)
     }
 }
 
-impl<S: Session> AsyncRead for Connection<S> {
+impl<S, C> AsyncRead for Connection<S, C>
+    where S: AsyncRead + AsyncWrite + Debug + io::Read + io::Write,
+          C: Session + Debug
+{
     unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
         self.0.prepare_uninitialized_buffer(buf)
     }
 }
 
-impl<S: Session> io::Write for Connection<S> {
+impl<S, C> io::Write for Connection<S, C>
+    where S: Debug + io::Read + io::Write,
+          C: Session + Debug
+{
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         self.0.write(buf)
     }
@@ -86,7 +101,10 @@ impl<S: Session> io::Write for Connection<S> {
     }
 }
 
-impl<S: Session> AsyncWrite for Connection<S> {
+impl<S, C> AsyncWrite for Connection<S, C>
+    where S: AsyncRead + AsyncWrite + Debug + io::Read + io::Write,
+          C: Session + Debug
+{
     fn shutdown(&mut self) -> Poll<(), io::Error> {
         self.0.shutdown()
     }
@@ -96,7 +114,10 @@ impl<S: Session> AsyncWrite for Connection<S> {
     }
 }
 
-impl<S: Session + Debug> AddrInfo for Connection<S> {
+impl<S, C> AddrInfo for Connection<S, C>
+    where S: AddrInfo + Debug,
+          C: Session + Debug
+{
     fn local_addr(&self) -> Result<SocketAddr, io::Error> {
         self.0.get_ref().0.local_addr()
     }
@@ -106,7 +127,10 @@ impl<S: Session + Debug> AddrInfo for Connection<S> {
     }
 }
 
-impl<S: Session + Debug> Io for Connection<S> {
+impl<S, C> Io for Connection<S, C>
+    where S: Io + Debug,
+          C: Session + Debug
+{
     fn shutdown_write(&mut self) -> Result<(), io::Error> {
         self.0.get_mut().0.shutdown_write()
     }

proxy/src/transport/tls/mod.rs

@@ -4,7 +4,7 @@ extern crate tokio_rustls;
 extern crate untrusted;
 extern crate webpki;
 
-mod conditional_accept;
+pub mod conditional_accept;
 mod config;
 mod cert_resolver;
 mod connection;
@@ -25,7 +25,12 @@ pub use self::{
         ServerConfigWatch,
         watch_for_config_changes,
     },
-    connection::{Connection, Session, UpgradeClientToTls},
+    connection::{
+        Connection,
+        Session,
+        UpgradeClientToTls,
+        UpgradeServerToTls
+    },
     dns_name::{DnsName, InvalidDnsName},
     identity::Identity,
 };