Transport opaque connections over mTLS (#785)

The proxy supports transporting "opaque" TCP streams, but it cannot do
so with mTLS. Because we cannot perform protocol detection (including
mTLS discovery) on server-first or otherwise idle TCP streams, we have
no reliable way to instrument mTLS on these connections.

This change leverages a new discovery API that annotates an
`OpaqueTransport` hint for targets that are marked opaque but have a
Linkerd proxy. This hint includes the inbound port of the target proxy
and, when specified, configures the outbound proxy to:

1. Connect directly to the inbound port (instead of the original target
   port).
2. Write an "opaque transport header"--a special protocol marker and a
   length-delimited protobuf message including the original target port.

This allows the inbound proxy to perform TLS discovery for these
streams; and then the inbound proxy is able to route the connection to
the proper application port as informed by the connection header.

This change, effectively, wraps arbitrary, opaque TCP streams in a
client-first TCP protocol so that inbound proxies can perform mTLS
detection.

This change does **not** extend the proxy's gateway mode to transport
these connections. This change will be done in a followup.

Author: olix0r

Cargo.lock

@@ -632,7 +632,7 @@ dependencies = [
 [[package]]
 name = "http-body"
 version = "0.4.0"
-source = "git+https://github.com/hyperium/http-body#5e434739e747c0b6611ec41020740b17f735d25a"
+source = "git+https://github.com/hyperium/http-body?branch=master#5e434739e747c0b6611ec41020740b17f735d25a"
 dependencies = [
  "bytes 0.6.0",
  "http",
@@ -876,6 +876,7 @@ dependencies = [
  "linkerd2-http-classify",
  "linkerd2-http-metrics",
  "linkerd2-metrics",
+ "linkerd2-opaque-transport",
  "linkerd2-opencensus",
  "linkerd2-proxy-api",
  "linkerd2-proxy-api-resolve",
@@ -988,9 +989,11 @@ dependencies = [
  "linkerd2-retry",
  "pin-project 0.4.22",
  "tokio 0.3.5",
+ "tokio-test 0.3.0",
  "tower",
  "tracing",
  "tracing-futures",
+ "tracing-subscriber",
 ]
 
 [[package]]
@@ -1249,6 +1252,23 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd2-opaque-transport"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "bytes 0.6.0",
+ "linkerd2-dns-name",
+ "linkerd2-error",
+ "linkerd2-io",
+ "linkerd2-proxy-transport",
+ "prost",
+ "prost-build",
+ "tokio 0.3.5",
+ "tokio-test 0.3.0",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd2-opencensus"
 version = "0.1.0"
@@ -1281,8 +1301,8 @@ dependencies = [
 
 [[package]]
 name = "linkerd2-proxy-api"
-version = "0.1.15"
-source = "git+https://github.com/linkerd/linkerd2-proxy-api?rev=259628840ba613c2e5673fc6a39b946e1b06f09a#259628840ba613c2e5673fc6a39b946e1b06f09a"
+version = "0.1.16"
+source = "git+https://github.com/linkerd/linkerd2-proxy-api?tag=v0.1.16#5d60650505652c804b589e575a9302e75e772ef1"
 dependencies = [
  "h2 0.2.6",
  "http",
@@ -2028,7 +2048,7 @@ dependencies = [
 [[package]]
 name = "prost-derive"
 version = "0.6.1"
-source = "git+https://github.com/danburkert/prost#278e8ab695f4acb415ab6f780a5c85f0a1a3b0ec"
+source = "git+https://github.com/danburkert/prost#423f5ec5bd165a7007a388edfb2b485d5bbf40c7"
 dependencies = [
  "anyhow",
  "itertools",

Cargo.toml

@@ -15,6 +15,7 @@ members = [
     "linkerd/channel",
     "linkerd/concurrency-limit",
     "linkerd/conditional",
+    "linkerd/opaque-transport",
     "linkerd/dns/name",
     "linkerd/dns",
     "linkerd/drain",
@@ -78,4 +79,4 @@ tonic = { git = "https://github.com/hawkw/tonic", branch = "eliza/tokio-0.3" }
 tonic-build = { git = "https://github.com/hawkw/tonic", branch = "eliza/tokio-0.3" }
 prost = { git = "https://github.com/danburkert/prost" }
 prost-build = { git = "https://github.com/danburkert/prost" }
-prost-types = { git = "https://github.com/danburkert/prost" }
+prost-types = { git = "https://github.com/danburkert/prost" }

linkerd/app/core/Cargo.toml

@@ -27,6 +27,7 @@ linkerd2-cache = { path = "../../cache" }
 linkerd2-buffer = { path = "../../buffer" }
 linkerd2-concurrency-limit = { path = "../../concurrency-limit" }
 linkerd2-conditional = { path = "../../conditional" }
+linkerd2-opaque-transport = { path = "../../opaque-transport" }
 linkerd2-dns = { path = "../../dns" }
 linkerd2-drain = { path = "../../drain" }
 linkerd2-duplex = { path = "../../duplex" }
@@ -40,7 +41,7 @@ linkerd2-http-metrics = { path = "../../http-metrics" }
 linkerd2-metrics = { path = "../../metrics" }
 linkerd2-opencensus = { path = "../../opencensus" }
 linkerd2-proxy-core = { path = "../../proxy/core" }
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a" }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.16" }
 linkerd2-proxy-api-resolve = { path = "../../proxy/api-resolve" }
 linkerd2-proxy-discover = { path = "../../proxy/discover" }
 linkerd2-proxy-identity = { path = "../../proxy/identity" }
@@ -85,5 +86,5 @@ libc = "0.2"
 procinfo = "0.4.2"
 
 [dev-dependencies]
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a", features = ["arbitrary"] }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.16", features = ["arbitrary"] }
 prost-types = "0.6.0"

linkerd/app/core/src/lib.rs

@@ -17,6 +17,7 @@ pub use linkerd2_drain as drain;
 pub use linkerd2_error::{Error, Never, Recover};
 pub use linkerd2_exp_backoff as exp_backoff;
 pub use linkerd2_http_metrics as http_metrics;
+pub use linkerd2_opaque_transport as opaque_transport;
 pub use linkerd2_opencensus as opencensus;
 pub use linkerd2_reconnect as reconnect;
 pub use linkerd2_service_profiles as profiles;

linkerd/app/inbound/src/endpoint.rs

@@ -1,7 +1,9 @@
 use indexmap::IndexMap;
 use linkerd2_app_core::{
     classify, dst, http_request_authority_addr, http_request_host_addr,
-    http_request_l5d_override_dst_addr, metrics, profiles,
+    http_request_l5d_override_dst_addr, metrics,
+    opaque_transport::Header,
+    profiles,
     proxy::{http, identity, tap},
     stack_tracing, svc,
     transport::{self, listen, tls},
@@ -140,6 +142,15 @@ impl From<TcpAccept> for TcpEndpoint {
     }
 }
 
+impl From<(Option<Header>, TcpAccept)> for TcpEndpoint {
+    fn from((hdr, tcp): (Option<Header>, TcpAccept)) -> Self {
+        match hdr {
+            Some(Header { port, .. }) => Self { port },
+            None => tcp.into(),
+        }
+    }
+}
+
 impl Into<SocketAddr> for TcpEndpoint {
     fn into(self) -> SocketAddr {
         (&self).into()

linkerd/app/inbound/src/lib.rs

@@ -16,6 +16,7 @@ use linkerd2_app_core::{
     classify,
     config::{ProxyConfig, ServerConfig},
     drain, dst, errors, metrics,
+    opaque_transport::DetectHeader,
     opencensus::proto::trace::v1 as oc,
     profiles,
     proxy::{
@@ -115,6 +116,7 @@ impl Config {
             .into_inner();
 
         let accept = self.build_accept(
+            prevent_loop,
             tcp_forward.clone(),
             http_router,
             metrics.clone(),
@@ -273,11 +275,8 @@ impl Config {
 
         // If the traffic is targeted at the inbound port, send it through
         // the loopback service (i.e. as a gateway).
-        let switch_loopback = svc::stack(loopback).push_switch(prevent_loop, profile);
-
-        // Attempts to resolve the target as a service profile or, if that
-        // fails, skips that stack to forward to the local endpoint.
-        svc::stack(switch_loopback)
+        svc::stack(profile)
+            .push_switch(prevent_loop, loopback)
             .check_new_service::<Target, http::Request<http::boxed::BoxBody>>()
             .push_on_response(
                 svc::layers()
@@ -300,6 +299,7 @@ impl Config {
 
     pub fn build_accept<I, F, A, H, S>(
         &self,
+        prevent_loop: impl Into<PreventLoop>,
         tcp_forward: F,
         http_router: H,
         metrics: metrics::Proxy,
@@ -319,10 +319,17 @@ impl Config {
            + 'static
     where
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + Unpin + Send + 'static,
-        F: svc::NewService<TcpEndpoint, Service = A> + Unpin + Clone + Send + 'static,
+        F: svc::NewService<TcpEndpoint, Service = A> + Unpin + Clone + Send + Sync + 'static,
         A: tower::Service<io::PrefixedIo<I>, Response = ()> + Clone + Send + Sync + 'static,
-        A::Error: Into<Error>,
-        A::Future: Send,
+        <A as tower::Service<io::PrefixedIo<I>>>::Error: Into<Error>,
+        <A as tower::Service<io::PrefixedIo<I>>>::Future: Send,
+        A: tower::Service<io::PrefixedIo<io::PrefixedIo<I>>, Response = ()>
+            + Clone
+            + Send
+            + Sync
+            + 'static,
+        <A as tower::Service<io::PrefixedIo<io::PrefixedIo<I>>>>::Error: Into<Error>,
+        <A as tower::Service<io::PrefixedIo<io::PrefixedIo<I>>>>::Future: Send,
         H: svc::NewService<Target, Service = S> + Unpin + Clone + Send + Sync + 'static,
         S: tower::Service<
                 http::Request<http::boxed::BoxBody>,
@@ -384,13 +391,31 @@ impl Config {
             .check_new_service::<(http::Version, TcpAccept), http::Request<_>>()
             .into_inner();
 
-        svc::stack(http::NewServeHttp::new(h2_settings, http_server, drain))
-            .push(svc::stack::NewOptional::layer(
+        // When HTTP detection fails, forward the connection to the application
+        // as an opaque TCP stream.
+        let tcp = svc::stack(tcp_forward.clone())
+            .push_map_target(TcpEndpoint::from)
+            .push_switch(
+                prevent_loop.into(),
+                // If the connection targets the inbound port, try to detect an
+                // opaque transport header and rewrite the target port
+                // accordingly. If there was no opaque transport header, the
+                // forwarding will fail when the tcp connect stack applies loop
+                // prevention.
                 svc::stack(tcp_forward)
                     .push_map_target(TcpEndpoint::from)
-                    .into_inner(),
-            ))
-            .check_new_clone::<(Option<http::Version>, TcpAccept)>()
+                    .push(transport::NewDetectService::layer(
+                        transport::detect::DetectTimeout::new(
+                            self.proxy.detect_protocol_timeout,
+                            DetectHeader::default(),
+                        ),
+                    )),
+            )
+            .push_on_response(drain::Retain::layer(drain.clone()))
+            .into_inner();
+
+        svc::stack(http::NewServeHttp::new(h2_settings, http_server, drain))
+            .push(svc::stack::NewOptional::layer(tcp))
             .push_cache(cache_max_idle_age)
             .push(transport::NewDetectService::layer(
                 transport::detect::DetectTimeout::new(

linkerd/app/inbound/src/prevent_loop.rs

@@ -42,7 +42,7 @@ where
     fn use_primary(&self, target: &T) -> bool {
         let addr = target.into();
         tracing::debug!(%addr, self.port);
-        addr.port() == self.port
+        addr.port() != self.port
     }
 }
 

linkerd/app/integration/Cargo.toml

@@ -27,7 +27,7 @@ hyper = "0.14.0-dev"
 linkerd2-app = { path = "..", features = ["mock-orig-dst"] }
 linkerd2-app-core = { path = "../core", features = ["mock-orig-dst"] }
 linkerd2-metrics = { path = "../../metrics", features = ["test_util"] }
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a", features = ["arbitrary"] }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.16", features = ["arbitrary"] }
 linkerd2-app-test = { path = "../test" }
 regex = "0.1"
 socket2 = "0.3.12"

linkerd/app/integration/src/controller.rs

@@ -426,6 +426,7 @@ pub fn destination_add_labeled(
         Hint::Unknown => None,
         Hint::H2 => Some(pb::ProtocolHint {
             protocol: Some(pb::protocol_hint::Protocol::H2(pb::protocol_hint::H2 {})),
+            ..Default::default()
         }),
     };
     pb::Update {

linkerd/app/integration/src/tests/transparency.rs

@@ -355,7 +355,7 @@ macro_rules! http1_tests {
         }
 
         #[tokio::test]
-        async fn http1_removes_connection_headers() {
+        async fn http1_removes_opaque_transports() {
             let _trace = trace_init();
 
             let srv = server::http1()