Proxy: Add TLS client infrastructure.

proxy/src/bind.rs

@@ -16,6 +16,7 @@ use ctx;
 use telemetry::{self, sensor};
 use transparency::{self, HttpBody, h1};
 use transport;
+use tls;
 
 /// Binds a `Service` from a `SocketAddr`.
 ///
@@ -205,18 +206,20 @@ where
     fn bind_stack(&self, ep: &Endpoint, protocol: &Protocol) -> Stack<B> {
         debug!("bind_stack endpoint={:?}, protocol={:?}", ep, protocol);
         let addr = ep.address();
+
+        let tls = tls::current_connection_config(ep.tls_identity(),
+                                                 &self.ctx.tls_client_config_watch());
+
         let client_ctx = ctx::transport::Client::new(
             &self.ctx,
             &addr,
             ep.metadata().clone(),
-            // TODO: when we can use TLS for client connections, indicate
-            //       whether or not the connection was TLS here.
-            ctx::transport::TlsStatus::Disabled,
+            tls.to_empty(),
         );
 
         // Map a socket address to a connection.
         let connect = self.sensors.connect(
-            transport::Connect::new(addr, ep.tls_identity().cloned()),
+            transport::Connect::new(addr, tls),
             &client_ctx,
         );
 
@@ -266,7 +269,9 @@ where
 
 
 impl<C, B> Bind<C, B> {
-    pub fn with_protocol(self, protocol: Protocol) -> BindProtocol<C, B> {
+    pub fn with_protocol(self, protocol: Protocol)
+        -> BindProtocol<C, B>
+    {
         BindProtocol {
             bind: self,
             protocol,

proxy/src/conditional.rs

@@ -0,0 +1,56 @@
+use std;
+
+/// Like `std::option::Option<C>` but `None` carries a reason why the value
+/// isn't available.
+#[derive(Clone, Debug)]
+pub enum Conditional<C, R> where
+    C: Clone + std::fmt::Debug,
+    R: Clone + std::fmt::Debug
+{
+    Some(C),
+    None(R)
+}
+
+impl<C, R> Copy for Conditional<C, R> where
+    C: Copy + Clone + std::fmt::Debug,
+    R: Copy + Clone + std::fmt::Debug {}
+
+impl<C, R> Eq for Conditional<C, R> where
+    C: Eq + Clone + std::fmt::Debug,
+    R: Eq + Clone + std::fmt::Debug {}
+
+impl<C, R> PartialEq for Conditional<C, R> where
+    C: PartialEq + Clone + std::fmt::Debug,
+    R: PartialEq + Clone + std::fmt::Debug {
+    fn eq(&self, other: &Conditional<C, R>) -> bool {
+        use self::Conditional::*;
+        match (self, other) {
+            (Some(a), Some(b)) => a.eq(b),
+            (None(a), None(b)) => a.eq(b),
+            _ => false,
+        }
+    }
+}
+
+impl<C, R> std::hash::Hash for Conditional<C, R> where
+    C: std::hash::Hash + Clone + std::fmt::Debug,
+    R: std::hash::Hash + Clone + std::fmt::Debug {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        match self {
+            Conditional::Some(c) => c.hash(state),
+            Conditional::None(r) => r.hash(state),
+        }
+    }
+}
+
+impl<C, R> Conditional<C, R> where
+    C: Clone + std::fmt::Debug,
+    R: Copy + Clone + std::fmt::Debug
+{
+    pub fn to_empty(&self) -> Conditional<(), R> {
+        match self {
+            Conditional::Some(_) => Conditional::Some(()),
+            Conditional::None(r) => Conditional::None(*r),
+        }
+    }
+}

proxy/src/connection.rs

@@ -9,32 +9,34 @@ use tokio::{
     net::{TcpListener, TcpStream, ConnectFuture},
     reactor::Handle,
 };
-
+use conditional::Conditional;
 use ctx::transport::TlsStatus;
 use config::Addr;
 use transport::{GetOriginalDst, Io, tls};
 
-pub type PlaintextSocket = TcpStream;
-
 pub struct BoundPort {
     inner: std::net::TcpListener,
     local_addr: SocketAddr,
 }
 
 /// Initiates a client connection to the given address.
-pub fn connect(addr: &SocketAddr) -> Connecting {
-    Connecting {
-        inner: PlaintextSocket::connect(addr),
-        // TODO: when we can open TLS client connections, this is where we will
-        //       indicate that for telemetry.
-        tls_status: TlsStatus::Disabled,
+pub fn connect(addr: &SocketAddr,
+               tls: tls::ConditionalConnectionConfig<tls::ClientConfig>)
+    -> Connecting
+{
+    Connecting::Plaintext {
+        connect: TcpStream::connect(addr),
+        tls: Some(tls),
     }
 }
 
 /// A socket that is in the process of connecting.
-pub struct Connecting {
-    inner: ConnectFuture,
-    tls_status: TlsStatus,
+pub enum Connecting {
+    Plaintext {
+        connect: ConnectFuture,
+        tls: Option<tls::ConditionalConnectionConfig<tls::ClientConfig>>,
+    },
+    UpgradeToTls(tls::UpgradeClientToTls),
 }
 
 /// Abstracts a plaintext socket vs. a TLS decorated one.
@@ -138,7 +140,7 @@ impl BoundPort {
                     // libraries don't have the necessary API for that, so just
                     // do it here.
                     set_nodelay_or_warn(&socket);
-                    let tls_status = if let Some((_identity, config_watch)) = &tls {
+                    let why_no_tls = if let Some((_identity, config_watch)) = &tls {
                         // TODO: use `identity` to differentiate between TLS
                         // that the proxy should terminate vs. TLS that should
                         // be passed through.
@@ -150,12 +152,12 @@ impl BoundPort {
                             return Either::A(f);
                         } else {
                             // No valid TLS configuration.
-                            TlsStatus::NoConfig
+                            tls::ReasonForNoTls::NoConfig
                         }
                     } else {
-                        TlsStatus::Disabled
+                        tls::ReasonForNoTls::Disabled
                     };
-                    let conn = Connection::new(socket, tls_status);
+                    let conn = Connection::plain(socket, why_no_tls);
                     Either::B(future::ok((conn, remote_addr)))
                 })
                 .then(|r| {
@@ -181,28 +183,50 @@ impl Future for Connecting {
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let socket = try_ready!(self.inner.poll());
-        set_nodelay_or_warn(&socket);
-        Ok(Async::Ready(Connection::new(socket, self.tls_status)))
+        loop {
+            let new_state = match self {
+                Connecting::Plaintext { connect, tls } => {
+                    let plaintext_stream = try_ready!(connect.poll());
+                    set_nodelay_or_warn(&plaintext_stream);
+                    match tls.take().expect("Polled after ready") {
+                        Conditional::Some(config) => {
+                            let upgrade_to_tls = tls::Connection::connect(
+                                plaintext_stream, &config.identity, config.config);
+                            Some(Connecting::UpgradeToTls(upgrade_to_tls))
+                        },
+                        Conditional::None(why) => {
+                            return Ok(Async::Ready(Connection::plain(plaintext_stream, why)));
+                        },
+                    }
+                },
+                Connecting::UpgradeToTls(upgrading) => {
+                    let tls_stream = try_ready!(upgrading.poll());
+                    return Ok(Async::Ready(Connection::tls(tls_stream)));
+                },
+            };
+            if let Some(s) = new_state {
+                std::mem::replace(self, s);
+            }
+        }
     }
 }
 
 // ===== impl Connection =====
 
 impl Connection {
-    fn new<I: Io + 'static>(io: I, tls_status: TlsStatus) -> Self {
+    fn plain(io: TcpStream, why_no_tls: tls::ReasonForNoTls) -> Self {
         Connection {
             io: Box::new(io),
             peek_buf: BytesMut::new(),
-            tls_status,
+            tls_status: Conditional::None(why_no_tls),
         }
     }
 
-    fn tls(tls: tls::Connection) -> Self {
-            Connection {
+    fn tls<S: tls::Session + std::fmt::Debug + 'static>(tls: tls::Connection<S>) -> Self {
+        Connection {
             io: Box::new(tls),
             peek_buf: BytesMut::new(),
-            tls_status: TlsStatus::Success,
+            tls_status: Conditional::Some(()),
         }
     }
 
@@ -312,7 +336,7 @@ impl<T: Peek> Future for PeekFuture<T> {
 
 // Misc.
 
-fn set_nodelay_or_warn(socket: &PlaintextSocket) {
+fn set_nodelay_or_warn(socket: &TcpStream) {
     if let Err(e) = socket.set_nodelay(true) {
         warn!(
             "could not set TCP_NODELAY on {:?}/{:?}: {}",

proxy/src/ctx/mod.rs

@@ -10,16 +10,20 @@
 use config;
 use std::time::SystemTime;
 use std::sync::Arc;
+use transport::tls;
+
 pub mod http;
 pub mod transport;
 
 /// Describes a single running proxy instance.
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug)]
 pub struct Process {
     /// Identifies the Kubernetes namespace in which this proxy is process.
     pub scheduled_namespace: String,
 
     pub start_time: SystemTime,
+
+    tls_client_config: tls::ClientConfigWatch,
 }
 
 /// Indicates the orientation of traffic, relative to a sidecar proxy.
@@ -29,7 +33,7 @@ pub struct Process {
 ///   local instance.
 /// - The  _outbound_ proxy receives traffic from the local instance and forwards it to a
 ///   remove service.
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug)]
 pub enum Proxy {
     Inbound(Arc<Process>),
     Outbound(Arc<Process>),
@@ -41,15 +45,17 @@ impl Process {
         Arc::new(Self {
             scheduled_namespace: ns.into(),
             start_time: SystemTime::now(),
+            tls_client_config: tls::ClientConfig::no_tls(),
         })
     }
 
     /// Construct a new `Process` from environment variables.
-    pub fn new(config: &config::Config) -> Arc<Self> {
+    pub fn new(config: &config::Config, tls_client_config: tls::ClientConfigWatch) -> Arc<Self> {
         let start_time = SystemTime::now();
         Arc::new(Self {
             scheduled_namespace: config.namespaces.pod.clone(),
             start_time,
+            tls_client_config,
         })
     }
 }
@@ -73,6 +79,12 @@ impl Proxy {
     pub fn is_outbound(&self) -> bool {
         !self.is_inbound()
     }
+
+    pub fn tls_client_config_watch(&self) -> &tls::ClientConfigWatch {
+        match self {
+            Proxy::Inbound(process) | Proxy::Outbound(process) => &process.tls_client_config
+        }
+    }
 }
 
 #[cfg(test)]
@@ -82,7 +94,6 @@ pub mod test_util {
         fmt,
         net::SocketAddr,
         sync::Arc,
-        time::SystemTime,
     };
 
     use ctx;
@@ -94,10 +105,7 @@ pub mod test_util {
     }
 
     pub fn process() -> Arc<ctx::Process> {
-        Arc::new(ctx::Process {
-            scheduled_namespace: "test".into(),
-            start_time: SystemTime::now(),
-        })
+        ctx::Process::test("test")
     }
 
     pub fn server(

proxy/src/ctx/transport.rs

@@ -5,6 +5,7 @@ use ctx;
 use control::destination;
 use telemetry::metrics::DstLabels;
 use transport::tls;
+use conditional::Conditional;
 
 #[derive(Debug)]
 pub enum Ctx {
@@ -33,18 +34,7 @@ pub struct Client {
 
 /// Identifies whether or not a connection was secured with TLS,
 /// and, if it was not, the reason why.
-#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
-pub enum TlsStatus {
-    /// The TLS handshake was successful.
-    Success,
-    /// TLS was not enabled for this connection.
-    Disabled,
-    /// TLS was enabled for this connection, but we have no valid
-    /// config.
-    NoConfig,
-    // TODO: When the proxy falls back to plaintext on handshake
-    // failures, we'll want to add a variant for that here as well.
-}
+pub type TlsStatus = Conditional<(), tls::ReasonForNoTls>;
 
 impl Ctx {
     pub fn proxy(&self) -> &Arc<ctx::Proxy> {

proxy/src/inbound.rs

@@ -109,6 +109,8 @@ mod tests {
     use super::Inbound;
     use bind::{self, Bind, Host};
     use ctx;
+    use conditional::Conditional;
+    use tls;
 
     fn new_inbound(default: Option<net::SocketAddr>, ctx: &Arc<ctx::Proxy>) -> Inbound<()> {
         let bind = Bind::new().with_ctx(ctx.clone());
@@ -123,6 +125,9 @@ mod tests {
         (addr, protocol)
     }
 
+    const TLS_DISABLED: Conditional<(), tls::ReasonForNoTls> =
+        Conditional::None(tls::ReasonForNoTls::Disabled);
+
     quickcheck! {
         fn recognize_orig_dst(
             orig_dst: net::SocketAddr,
@@ -134,9 +139,7 @@ mod tests {
             let inbound = new_inbound(None, &ctx);
 
             let srv_ctx = ctx::transport::Server::new(
-                &ctx, &local, &remote, &Some(orig_dst),
-                ctx::transport::TlsStatus::Disabled,
-            );
+                &ctx, &local, &remote, &Some(orig_dst), TLS_DISABLED);
 
             let rec = srv_ctx.orig_dst_if_not_local().map(make_key_http1);
 
@@ -163,7 +166,7 @@ mod tests {
                     &local,
                     &remote,
                     &None,
-                    ctx::transport::TlsStatus::Disabled,
+                    TLS_DISABLED,
                 ));
 
             inbound.recognize(&req) == default.map(make_key_http1)
@@ -195,7 +198,7 @@ mod tests {
                     &local,
                     &remote,
                     &Some(local),
-                    ctx::transport::TlsStatus::Disabled,
+                    TLS_DISABLED,
                 ));
 
             inbound.recognize(&req) == default.map(make_key_http1)