Decouple TLS detection from TCP connections (#818)

olix0r · web-flow · commit f3ec8b726561 · 2021-01-04T14:03:37.000-08:00
The TLS accept stack--the `DetectTls` type--is currently coupled to the
`TcpStream` type; but the `TcpStream` type requires an actual OS-level
TCP connection, which isn't ideal for testing.

This change introduces a new trait `tls::accept::Detectable` and an
implementation for `TcpStream`. This will permit us to use alternate
implementations (e.g., for `io::DuplexStream`).

This change also updates the `DetectTls` type name to `NewDetectTls`,
and `AcceptTls` to `DetectTls`, to fit our more recent idioms.
diff --git a/linkerd/app/inbound/src/lib.rs b/linkerd/app/inbound/src/lib.rs
@@ -384,37 +384,35 @@ impl Config {
             .into_inner()
     }
 
-    pub fn build_tls_accept<D, DSvc, F, FSvc>(
+    pub fn build_tls_accept<I, D, DSvc, F, FSvc>(
         self,
         detect: D,
         tcp_forward: F,
         identity: tls::Conditional<identity::Local>,
         metrics: metrics::Proxy,
     ) -> impl svc::NewService<
         listen::Addrs,
-        Service = impl svc::Service<TcpStream, Response = (), Error = Error, Future = impl Send>,
+        Service = impl svc::Service<I, Response = (), Error = Error, Future = impl Send>,
     > + Clone
     where
+        I: tls::accept::Detectable + Send + 'static,
         D: svc::NewService<TcpAccept, Service = DSvc> + Clone + Send + 'static,
-        DSvc: svc::Service<SensorIo<tls::accept::Io>, Response = ()> + Send + 'static,
+        DSvc: svc::Service<SensorIo<tls::accept::Io<I>>, Response = ()> + Send + 'static,
         DSvc::Error: Into<Error>,
         DSvc::Future: Send,
         F: svc::NewService<TcpEndpoint, Service = FSvc> + Clone + 'static,
-        FSvc: svc::Service<SensorIo<TcpStream>, Response = ()> + 'static,
+        FSvc: svc::Service<SensorIo<I>, Response = ()> + 'static,
         FSvc::Error: Into<Error>,
         FSvc::Future: Send,
     {
-        let ProxyConfig {
-            detect_protocol_timeout,
-            ..
-        } = self.proxy;
-        let require_identity = self.require_identity_for_inbound_ports;
-
         svc::stack(detect)
-            .push_request_filter(require_identity)
+            .push_request_filter(self.require_identity_for_inbound_ports)
             .push(metrics.transport.layer_accept())
             .push_map_target(TcpAccept::from)
-            .push(tls::DetectTls::layer(identity, detect_protocol_timeout))
+            .push(tls::NewDetectTls::layer(
+                identity,
+                self.proxy.detect_protocol_timeout,
+            ))
             .push_switch(
                 self.disable_protocol_detection_for_ports,
                 svc::stack(tcp_forward)
diff --git a/linkerd/app/src/admin.rs b/linkerd/app/src/admin.rs
@@ -33,7 +33,7 @@ impl Config {
 
         let (ready, latch) = admin::Readiness::new();
         let admin = admin::Admin::new(report, ready, shutdown, trace);
-        let accept = tls::DetectTls::new(
+        let accept = tls::NewDetectTls::new(
             identity,
             admin.into_accept(),
             std::time::Duration::from_secs(1),
diff --git a/linkerd/app/src/tap.rs b/linkerd/app/src/tap.rs
@@ -51,7 +51,7 @@ impl Config {
 
                 let service =
                     tap::AcceptPermittedClients::new(permitted_peer_identities.into(), server);
-                let accept = tls::DetectTls::new(
+                let accept = tls::NewDetectTls::new(
                     identity,
                     move |meta: tls::accept::Meta| {
                         let service = service.clone();
diff --git a/linkerd/proxy/tap/src/accept.rs b/linkerd/proxy/tap/src/accept.rs
@@ -9,10 +9,13 @@ use linkerd2_proxy_transport::{
     io,
     tls::{accept::Connection, Conditional, ReasonForNoPeerName},
 };
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::Arc;
-use std::task::{Context, Poll};
+use std::{
+    future::Future,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::net::TcpStream;
 use tower::Service;
 
 #[derive(Clone, Debug)]
@@ -63,7 +66,7 @@ impl AcceptPermittedClients {
     }
 }
 
-impl Service<Connection> for AcceptPermittedClients {
+impl Service<Connection<TcpStream>> for AcceptPermittedClients {
     type Response = ServeFuture;
     type Error = Error;
     type Future = future::Ready<Result<Self::Response, Self::Error>>;
@@ -72,7 +75,7 @@ impl Service<Connection> for AcceptPermittedClients {
         Poll::Ready(Ok(()))
     }
 
-    fn call(&mut self, (meta, io): Connection) -> Self::Future {
+    fn call(&mut self, (meta, io): Connection<TcpStream>) -> Self::Future {
         future::ok(match meta.peer_identity {
             Conditional::Some(ref peer) => {
                 if self.permitted_client_ids.contains(peer) {
diff --git a/linkerd/proxy/transport/src/tls/accept.rs b/linkerd/proxy/transport/src/tls/accept.rs
@@ -27,6 +27,28 @@ pub trait HasConfig {
     fn tls_server_config(&self) -> Arc<Config>;
 }
 
+/// Must be implemented for I/O types like `TcpStream` on which TLS is
+/// transparently detected.
+///
+/// This is necessary so that we can be generic over the I/O type but still use
+/// `TcpStream::peek` to avoid allocating for mTLS SNI detection.
+#[async_trait::async_trait]
+pub trait Detectable {
+    /// Attempts to detect a `ClientHello` message from the underlying transport
+    /// and, if its SNI matches `local_name`, initiates a TLS server handshake to
+    /// decrypt the stream.
+    ///
+    /// Returns the client's identity, if one exists, and an optionally decrypted
+    /// transport.
+    async fn detected(
+        self,
+        config: Arc<Config>,
+        local_name: identity::Name,
+    ) -> io::Result<(PeerIdentity, Io<Self>)>
+    where
+        Self: Sized;
+}
+
 /// Produces a server config that fails to handshake all connections.
 pub fn empty_config() -> Arc<Config> {
     let verifier = rustls::NoClientAuth::new();
@@ -40,12 +62,12 @@ pub struct Meta {
     pub addrs: Addrs,
 }
 
-pub type Io = EitherIo<PrefixedIo<TcpStream>, TlsStream<PrefixedIo<TcpStream>>>;
+pub type Io<T> = EitherIo<PrefixedIo<T>, TlsStream<PrefixedIo<T>>>;
 
-pub type Connection = (Meta, Io);
+pub type Connection<T> = (Meta, Io<T>);
 
 #[derive(Clone, Debug)]
-pub struct DetectTls<I, A> {
+pub struct NewDetectTls<I, A> {
     local_identity: Conditional<I>,
     inner: A,
     timeout: Duration,
@@ -55,7 +77,7 @@ pub struct DetectTls<I, A> {
 pub struct DetectTimeout(());
 
 #[derive(Clone, Debug)]
-pub struct AcceptTls<I, N> {
+pub struct DetectTls<I, N> {
     addrs: Addrs,
     local_identity: Conditional<I>,
     inner: N,
@@ -70,7 +92,7 @@ const PEEK_CAPACITY: usize = 512;
 // insufficient. This is the same value used in HTTP detection.
 const BUFFER_CAPACITY: usize = 8192;
 
-impl<I: HasConfig, N> DetectTls<I, N> {
+impl<I: HasConfig, N> NewDetectTls<I, N> {
     pub fn new(local_identity: Conditional<I>, inner: N, timeout: Duration) -> Self {
         Self {
             local_identity,
@@ -90,15 +112,15 @@ impl<I: HasConfig, N> DetectTls<I, N> {
     }
 }
 
-impl<I, N> NewService<Addrs> for DetectTls<I, N>
+impl<I, N> NewService<Addrs> for NewDetectTls<I, N>
 where
     I: HasConfig + Clone,
     N: NewService<Meta> + Clone,
 {
-    type Service = AcceptTls<I, N>;
+    type Service = DetectTls<I, N>;
 
     fn new_service(&mut self, addrs: Addrs) -> Self::Service {
-        AcceptTls {
+        DetectTls {
             addrs,
             local_identity: self.local_identity.clone(),
             inner: self.inner.clone(),
@@ -107,12 +129,14 @@ where
     }
 }
 
-impl<I: HasConfig, N, A> tower::Service<TcpStream> for AcceptTls<I, N>
+impl<T, I, N, NSvc> tower::Service<T> for DetectTls<I, N>
 where
-    N: NewService<Meta, Service = A> + Clone + Send + 'static,
-    A: tower::Service<Io, Response = ()> + Send + 'static,
-    A::Error: Into<Error>,
-    A::Future: Send,
+    T: Detectable + Send + 'static,
+    I: HasConfig,
+    N: NewService<Meta, Service = NSvc> + Clone + Send + 'static,
+    NSvc: tower::Service<Io<T>, Response = ()> + Send + 'static,
+    NSvc::Error: Into<Error>,
+    NSvc::Future: Send,
 {
     type Response = ();
     type Error = Error;
@@ -122,7 +146,7 @@ where
         Poll::Ready(Ok(()))
     }
 
-    fn call(&mut self, tcp: TcpStream) -> Self::Future {
+    fn call(&mut self, tcp: T) -> Self::Future {
         let addrs = self.addrs.clone();
         let mut new_accept = self.inner.clone();
 
@@ -134,7 +158,7 @@ where
 
                 Box::pin(async move {
                     let (peer_identity, io) = tokio::select! {
-                        res = detect(config, name, tcp) => { res? }
+                        res = tcp.detected(config, name) => { res? }
                         () = timeout => {
                             return Err(DetectTimeout(()).into());
                         }
@@ -163,71 +187,74 @@ where
     }
 }
 
-pub async fn detect(
-    tls_config: Arc<Config>,
-    local_id: identity::Name,
-    mut tcp: TcpStream,
-) -> io::Result<(PeerIdentity, Io)> {
-    const NO_TLS_META: PeerIdentity = Conditional::None(ReasonForNoPeerName::NoTlsFromRemote);
-
-    // First, try to use MSG_PEEK to read the SNI from the TLS ClientHello.
-    // Because peeked data does not need to be retained, we use a static
-    // buffer to prevent needless heap allocation.
-    //
-    // Anecdotally, the ClientHello sent by Linkerd proxies is <300B. So a
-    // ~500B byte buffer is more than enough.
-    let mut buf = [0u8; PEEK_CAPACITY];
-    let sz = tcp.peek(&mut buf).await?;
-    debug!(sz, "Peeked bytes from TCP stream");
-    match conditional_accept::match_client_hello(&buf, &local_id) {
-        conditional_accept::Match::Matched => {
-            trace!("Identified matching SNI via peek");
-            // Terminate the TLS stream.
-            let (peer_id, tls) = handshake(tls_config, PrefixedIo::from(tcp)).await?;
-            return Ok((peer_id, EitherIo::Right(tls)));
-        }
-
-        conditional_accept::Match::NotMatched => {
-            trace!("Not a matching TLS ClientHello");
-            return Ok((NO_TLS_META, EitherIo::Left(tcp.into())));
-        }
-
-        conditional_accept::Match::Incomplete => {}
-    }
-
-    // Peeking didn't return enough data, so instead we'll allocate more
-    // capacity and try reading data from the socket.
-    debug!("Attempting to buffer TLS ClientHello after incomplete peek");
-    let mut buf = BytesMut::with_capacity(BUFFER_CAPACITY);
-    debug!(buf.capacity = %buf.capacity(), "Reading bytes from TCP stream");
-    while tcp.read_buf(&mut buf).await? != 0 {
-        debug!(buf.len = %buf.len(), "Read bytes from TCP stream");
-        match conditional_accept::match_client_hello(buf.as_ref(), &local_id) {
+#[async_trait::async_trait]
+impl Detectable for TcpStream {
+    async fn detected(
+        mut self,
+        tls_config: Arc<Config>,
+        local_id: identity::Name,
+    ) -> io::Result<(PeerIdentity, Io<Self>)> {
+        const NO_TLS_META: PeerIdentity = Conditional::None(ReasonForNoPeerName::NoTlsFromRemote);
+
+        // First, try to use MSG_PEEK to read the SNI from the TLS ClientHello.
+        // Because peeked data does not need to be retained, we use a static
+        // buffer to prevent needless heap allocation.
+        //
+        // Anecdotally, the ClientHello sent by Linkerd proxies is <300B. So a
+        // ~500B byte buffer is more than enough.
+        let mut buf = [0u8; PEEK_CAPACITY];
+        let sz = self.peek(&mut buf).await?;
+        debug!(sz, "Peeked bytes from TCP stream");
+        match conditional_accept::match_client_hello(&buf, &local_id) {
             conditional_accept::Match::Matched => {
-                trace!("Identified matching SNI via buffered read");
+                trace!("Identified matching SNI via peek");
                 // Terminate the TLS stream.
-                let (peer_id, tls) =
-                    handshake(tls_config.clone(), PrefixedIo::new(buf.freeze(), tcp)).await?;
+                let (peer_id, tls) = handshake(tls_config, PrefixedIo::from(self)).await?;
                 return Ok((peer_id, EitherIo::Right(tls)));
             }
 
-            conditional_accept::Match::NotMatched => break,
+            conditional_accept::Match::NotMatched => {
+                trace!("Not a matching TLS ClientHello");
+                return Ok((NO_TLS_META, EitherIo::Left(self.into())));
+            }
 
-            conditional_accept::Match::Incomplete => {
-                if buf.capacity() == 0 {
-                    // If we can't buffer an entire TLS ClientHello, it
-                    // almost definitely wasn't initiated by another proxy,
-                    // at least.
-                    warn!("Buffer insufficient for TLS ClientHello");
-                    break;
+            conditional_accept::Match::Incomplete => {}
+        }
+
+        // Peeking didn't return enough data, so instead we'll allocate more
+        // capacity and try reading data from the socket.
+        debug!("Attempting to buffer TLS ClientHello after incomplete peek");
+        let mut buf = BytesMut::with_capacity(BUFFER_CAPACITY);
+        debug!(buf.capacity = %buf.capacity(), "Reading bytes from TCP stream");
+        while self.read_buf(&mut buf).await? != 0 {
+            debug!(buf.len = %buf.len(), "Read bytes from TCP stream");
+            match conditional_accept::match_client_hello(buf.as_ref(), &local_id) {
+                conditional_accept::Match::Matched => {
+                    trace!("Identified matching SNI via buffered read");
+                    // Terminate the TLS stream.
+                    let (peer_id, tls) =
+                        handshake(tls_config.clone(), PrefixedIo::new(buf.freeze(), self)).await?;
+                    return Ok((peer_id, EitherIo::Right(tls)));
+                }
+
+                conditional_accept::Match::NotMatched => break,
+
+                conditional_accept::Match::Incomplete => {
+                    if buf.capacity() == 0 {
+                        // If we can't buffer an entire TLS ClientHello, it
+                        // almost definitely wasn't initiated by another proxy,
+                        // at least.
+                        warn!("Buffer insufficient for TLS ClientHello");
+                        break;
+                    }
                 }
             }
         }
-    }
 
-    trace!("Could not read TLS ClientHello via buffering");
-    let io = EitherIo::Left(PrefixedIo::new(buf.freeze(), tcp));
-    Ok((NO_TLS_META, io))
+        trace!("Could not read TLS ClientHello via buffering");
+        let io = EitherIo::Left(PrefixedIo::new(buf.freeze(), self));
+        Ok((NO_TLS_META, io))
+    }
 }
 
 async fn handshake<T>(
diff --git a/linkerd/proxy/transport/src/tls/mod.rs b/linkerd/proxy/transport/src/tls/mod.rs
@@ -6,7 +6,7 @@ pub mod accept;
 pub mod client;
 mod conditional_accept;
 
-pub use self::accept::DetectTls;
+pub use self::accept::NewDetectTls;
 pub use self::client::Client;
 
 /// Describes whether or not a connection was secured with TLS and, if it was
diff --git a/linkerd/proxy/transport/tests/tls_accept.rs b/linkerd/proxy/transport/tests/tls_accept.rs
@@ -8,11 +8,7 @@
 use futures::prelude::*;
 use linkerd2_error::Never;
 use linkerd2_identity::{test_util, CrtKey, Name};
-use linkerd2_proxy_transport::tls::{
-    self,
-    accept::{self, DetectTls},
-    Conditional,
-};
+use linkerd2_proxy_transport::tls::{self, Conditional};
 use linkerd2_proxy_transport::{
     io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
     BindTcp, ConnectTcp,
@@ -115,7 +111,7 @@ where
     CF: Future<Output = Result<CR, io::Error>> + Send + 'static,
     CR: Send + 'static,
     // Server
-    S: Fn(accept::Connection) -> SF + Clone + Send + 'static,
+    S: Fn(tls::accept::Connection<TcpStream>) -> SF + Clone + Send + 'static,
     SF: Future<Output = Result<SR, io::Error>> + Send + 'static,
     SR: Send + 'static,
 {
@@ -148,9 +144,9 @@ where
 
         let (listen_addr, listen) = BindTcp::new(addr, None).bind().expect("must bind");
 
-        let mut detect = DetectTls::new(
+        let mut detect = tls::NewDetectTls::new(
             server_tls,
-            move |meta: accept::Meta| {
+            move |meta: tls::accept::Meta| {
                 let server = server.clone();
                 let sender = sender.clone();
                 let peer_identity = Some(meta.peer_identity.clone());
