proxy: Add TLS identity to endpoint metadata and wire it through to `Connect::new` (#1008)

Depends on #1006. Depends on #1041.

This PR adds a `tls_identity` field to the endpoint `Metadata` struct, which
contains the `TlsIdentity` metadata sent by the control plane's Destination
service. 

I changed the `ctx::transport::Client` context struct to hold a `Metadata`,
rather than just the labels, so the TLS support determination is always
available. In addition, I've added it as an additional parameter to 
`transport::Connect::new`, so that when we create a new connection, the TLS
code will be able to determine whether or not TLS is supported and, if it is, 
how to verify the endpoint's identity.

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

proxy/benches/record.rs

@@ -7,6 +7,7 @@ extern crate test;
 
 use conduit_proxy::{
     ctx,
+    control::destination,
     telemetry::{
         event,
         metrics,
@@ -43,7 +44,11 @@ where
     L: IntoIterator<Item=(S, S)>,
     S: fmt::Display,
 {
-    ctx::transport::Client::new(&proxy, &addr(), metrics::DstLabels::new(labels))
+    ctx::transport::Client::new(
+        &proxy,
+        &addr(),
+        destination::Metadata::new(metrics::DstLabels::new(labels), None),
+    )
 }
 
 fn request(

proxy/src/bind.rs

@@ -214,12 +214,12 @@ where
         let client_ctx = ctx::transport::Client::new(
             &self.ctx,
             &addr,
-            ep.dst_labels().cloned(),
+            ep.metadata().clone(),
         );
 
         // Map a socket address to a connection.
         let connect = self.sensors.connect(
-            transport::Connect::new(addr),
+            transport::Connect::new(addr, ep.tls_identity()),
             &client_ctx,
         );
 

proxy/src/control/destination/background.rs

@@ -22,9 +22,13 @@ use tower_reconnect::Reconnect;
 use conduit_proxy_controller_grpc::common::{Destination, TcpAddress};
 use conduit_proxy_controller_grpc::destination::client::Destination as DestinationSvc;
 use conduit_proxy_controller_grpc::destination::update::Update as PbUpdate2;
-use conduit_proxy_controller_grpc::destination::{Update as PbUpdate, WeightedAddr};
+use conduit_proxy_controller_grpc::destination::{
+    Update as PbUpdate,
+    TlsIdentity as TlsIdentityPb,
+    WeightedAddr,
+};
 
-use super::{Metadata, ResolveRequest, Responder, Update};
+use super::{TlsIdentity, Metadata, ResolveRequest, Responder, Update};
 use control::{
     cache::{Cache, CacheChange, Exists},
     fully_qualified_authority::FullyQualifiedAuthority,
@@ -566,12 +570,27 @@ fn pb_to_addr_meta(
         .collect::<Vec<_>>();
     labels.sort_by(|(k0, _), (k1, _)| k0.cmp(k1));
 
-    let meta = Metadata {
-        metric_labels: DstLabels::new(labels.into_iter()),
-    };
+    let tls = pb.tls_identity.and_then(TlsIdentity::from_pb);
+
+    let meta = Metadata::new(DstLabels::new(labels.into_iter()), tls);
     Some((addr, meta))
 }
 
+impl TlsIdentity {
+    pub fn from_pb(pb: TlsIdentityPb) -> Option<Self> {
+        use conduit_proxy_controller_grpc::destination::tls_identity::Strategy;
+        pb.strategy.map(|strategy| match strategy {
+            Strategy::K8sPodNamespace(i) =>
+                TlsIdentity::K8sPodNamespace {
+                    controller_ns: i.controller_ns,
+                    pod_ns: i.pod_ns,
+                    pod_name: i.pod_name,
+                },
+        })
+    }
+}
+
+
 fn pb_to_sock_addr(pb: TcpAddress) -> Option<SocketAddr> {
     use conduit_proxy_controller_grpc::common::ip_address::Ip;
     use std::net::{Ipv4Addr, Ipv6Addr};

proxy/src/control/destination/endpoint.rs

@@ -1,7 +1,7 @@
 use std::net::SocketAddr;
 
 use telemetry::metrics::DstLabels;
-
+use super::{Metadata, TlsIdentity};
 
 /// An individual traffic target.
 ///
@@ -9,16 +9,16 @@ use telemetry::metrics::DstLabels;
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
 pub struct Endpoint {
     address: SocketAddr,
-    dst_labels: Option<DstLabels>,
+    metadata: Metadata,
 }
 
 // ==== impl Endpoint =====
 
 impl Endpoint {
-    pub fn new(address: SocketAddr, dst_labels: Option<DstLabels>) -> Self {
+    pub fn new(address: SocketAddr, metadata: Metadata) -> Self {
         Self {
             address,
-            dst_labels,
+            metadata,
         }
     }
 
@@ -26,8 +26,16 @@ impl Endpoint {
         self.address
     }
 
+    pub fn metadata(&self) -> &Metadata {
+        &self.metadata
+    }
+
     pub fn dst_labels(&self) -> Option<&DstLabels> {
-        self.dst_labels.as_ref()
+        self.metadata.dst_labels()
+    }
+
+    pub fn tls_identity(&self) -> Option<&TlsIdentity> {
+        self.metadata.tls_identity()
     }
 }
 
@@ -35,7 +43,7 @@ impl From<SocketAddr> for Endpoint {
     fn from(address: SocketAddr) -> Self {
         Self {
             address,
-            dst_labels: None,
+            metadata: Metadata::no_metadata()
         }
     }
 }

proxy/src/control/destination/mod.rs

@@ -88,11 +88,31 @@ pub struct Resolution<B> {
 
 /// Metadata describing an endpoint.
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
-struct Metadata {
+pub struct Metadata {
     /// A set of Prometheus metric labels describing the destination.
-    metric_labels: Option<DstLabels>,
+    dst_labels: Option<DstLabels>,
+
+    /// How to verify TLS for the endpoint.
+    tls_identity: Option<Arc<TlsIdentity>>,
 }
 
+/// How to verify TLS for an endpoint.
+///
+/// NOTE: This currently derives `Hash`, `PartialEq`, and `Eq`, which is not
+///       entirely correct, as domain name equality ought to be case
+///       insensitive. However, `Metadata` must be `Hash` + `Eq`, so this is at
+///       least better than having `Metadata` ignore the TLS identity when
+///       checking for equality
+#[derive(Debug, Hash, PartialEq, Eq)]
+pub enum TlsIdentity {
+    K8sPodNamespace {
+        controller_ns: String,
+        pod_ns: String,
+        pod_name: String,
+    },
+}
+
+
 #[derive(Debug, Clone)]
 enum Update {
     /// Indicates that an endpoint should be bound to `SocketAddr` with the
@@ -204,7 +224,7 @@ where
                     // by replacing the old endpoint with the new one, so
                     // insertions of new endpoints and metadata changes for
                     // existing ones can be handled in the same way.
-                    let endpoint = Endpoint::new(addr, meta.metric_labels.clone());
+                    let endpoint = Endpoint::new(addr, meta);
 
                     let service = self.bind.bind(&endpoint).map_err(|_| ())?;
 
@@ -229,9 +249,31 @@ impl Responder {
 // ===== impl Metadata =====
 
 impl Metadata {
-    fn no_metadata() -> Self {
+    /// Construct a Metadata struct representing an endpoint with no metadata.
+    pub fn no_metadata() -> Self {
         Metadata {
-            metric_labels: None,
+            dst_labels: None,
+            // If we have no metadata on an endpoint, assume it does not support TLS.
+            tls_identity: None,
         }
     }
+
+    pub fn new(
+        dst_labels: Option<DstLabels>,
+        tls_identity: Option<TlsIdentity>
+    ) -> Self {
+        Metadata {
+            dst_labels,
+            tls_identity: tls_identity.map(Arc::new),
+        }
+    }
+
+    /// Returns the endpoint's labels from the destination service, if it has them.
+    pub fn dst_labels(&self) -> Option<&DstLabels> {
+        self.dst_labels.as_ref()
+    }
+
+    pub fn tls_identity(&self) -> Option<&TlsIdentity> {
+        self.tls_identity.as_ref().map(Arc::as_ref)
+    }
 }

proxy/src/ctx/http.rs

@@ -2,6 +2,7 @@ use http;
 use std::sync::Arc;
 
 use ctx;
+use control::destination;
 use telemetry::metrics::DstLabels;
 
 /// Describes a stream's request headers.
@@ -53,8 +54,12 @@ impl Request {
         Arc::new(r)
     }
 
+    pub fn tls_identity(&self) -> Option<&destination::TlsIdentity> {
+        self.client.tls_identity()
+    }
+
     pub fn dst_labels(&self) -> Option<&DstLabels> {
-        self.client.dst_labels.as_ref()
+        self.client.dst_labels()
     }
 }
 
@@ -68,6 +73,10 @@ impl Response {
         Arc::new(r)
     }
 
+    pub fn tls_identity(&self) -> Option<&destination::TlsIdentity> {
+        self.request.tls_identity()
+    }
+
     pub fn dst_labels(&self) -> Option<&DstLabels> {
         self.request.dst_labels()
     }

proxy/src/ctx/mod.rs

@@ -86,6 +86,7 @@ pub mod test_util {
     };
 
     use ctx;
+    use control::destination;
     use telemetry::metrics::DstLabels;
 
     fn addr() -> SocketAddr {
@@ -108,8 +109,8 @@ pub mod test_util {
         L: IntoIterator<Item=(S, S)>,
         S: fmt::Display,
     {
-        let labels = DstLabels::new(labels);
-        ctx::transport::Client::new(&proxy, &addr(), labels)
+        let meta = destination::Metadata::new(DstLabels::new(labels), None);
+        ctx::transport::Client::new(&proxy, &addr(), meta)
     }
 
     pub fn request(

proxy/src/ctx/transport.rs

@@ -1,8 +1,8 @@
-use std::{cmp, hash};
 use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 
 use ctx;
+use control::destination;
 use telemetry::metrics::DstLabels;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -21,11 +21,11 @@ pub struct Server {
 }
 
 /// Identifies a connection from the proxy to another process.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Client {
     pub proxy: Arc<ctx::Proxy>,
     pub remote: SocketAddr,
-    pub dst_labels: Option<DstLabels>,
+    pub metadata: destination::Metadata,
 }
 
 impl Ctx {
@@ -82,35 +82,26 @@ impl Client {
     pub fn new(
         proxy: &Arc<ctx::Proxy>,
         remote: &SocketAddr,
-        dst_labels: Option<DstLabels>,
+        metadata: destination::Metadata,
     ) -> Arc<Client> {
         let c = Client {
             proxy: Arc::clone(proxy),
             remote: *remote,
-            dst_labels,
+            metadata,
         };
 
         Arc::new(c)
     }
-}
 
-impl hash::Hash for Client {
-    fn hash<H: hash::Hasher>(&self, state: &mut H) {
-        self.proxy.hash(state);
-        self.remote.hash(state);
-        // ignore dst_labels
+    pub fn tls_identity(&self) -> Option<&destination::TlsIdentity> {
+        self.metadata.tls_identity()
+    }
+
+    pub fn dst_labels(&self) -> Option<&DstLabels> {
+        self.metadata.dst_labels()
     }
 }
 
-impl cmp::PartialEq for Client {
-    fn eq(&self, other: &Self) -> bool {
-        self.proxy.eq(&other.proxy) &&
-        self.remote.eq(&other.remote)
-    }
-}
-
-impl cmp::Eq for Client {}
-
 impl From<Arc<Client>> for Ctx {
     fn from(c: Arc<Client>) -> Self {
         Ctx::Client(c)

proxy/src/transparency/tcp.rs

@@ -7,6 +7,7 @@ use futures::{future, Async, Future, Poll};
 use tokio_connect::Connect;
 use tokio::io::{AsyncRead, AsyncWrite};
 
+use control::destination;
 use ctx::transport::{Client as ClientCtx, Server as ServerCtx};
 use telemetry::Sensors;
 use timeout::Timeout;
@@ -57,10 +58,10 @@ impl Proxy {
         let client_ctx = ClientCtx::new(
             &srv_ctx.proxy,
             &orig_dst,
-            None,
+            destination::Metadata::no_metadata(),
         );
         let c = Timeout::new(
-            transport::Connect::new(orig_dst),
+            transport::Connect::new(orig_dst, None), // No TLS.
             self.connect_timeout,
         );
         let connect = self.sensors.connect(c, &client_ctx);

proxy/src/transport/connect.rs

@@ -8,6 +8,7 @@ use std::str::FromStr;
 use http;
 
 use connection;
+use control::destination;
 use dns;
 
 #[derive(Debug, Clone)]
@@ -95,7 +96,12 @@ impl fmt::Display for HostAndPort {
 
 impl Connect {
     /// Returns a `Connect` to `addr`.
-    pub fn new(addr: SocketAddr) -> Self {
+    pub fn new(
+        addr: SocketAddr,
+        tls_identity: Option<&destination::TlsIdentity>,
+    ) -> Self {
+        // TODO: this is currently unused.
+        let _ = tls_identity;
         Self {
             addr,
         }