inbound, outbound: Param-ify listen::Addrs (#955)

This PR changes the `Service`s constructed by the inbound and outbound
stacks from always taking the `listen::Addrs` type as a target, to being
parameterized over a target type that implements `Param` for various
address types.

This required changing some middleware layers that were previously
hardcoded to `listen::Addrs`, and adding `Param` impls to
`listen::Addrs` for various address types. I also added a new
`TcpAccept::from_local_addr` constructor, which sets the target address
based on the server's listen address, rather than from the
SO_ORIGINAL_DST address. This is used by the admin server, which doesn't
configure its listening socket to include original dst addresses.
Previously, the `From` implementation for `TcpAccept` just always tried
the orig dst addr and fell back to the listen addr if it was not
present. Now, we have separate behavior for the admin server and the
proxy servers --- admin always uses the local addr, and the proxies only
try SO_ORIGINAL_DST or fail.

There's no functional change here, but this will set up for a future
change to make stacks generic over SO_ORIGINAL_DST detection, allowing
us to remove the currently quite awkward feature flagging.

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

linkerd/app/inbound/src/direct.rs

@@ -4,7 +4,7 @@ use linkerd_app_core::{
     proxy::identity::LocalCrtKey,
     svc::{self, Param},
     tls,
-    transport::{self, listen, metrics::SensorIo, ClientAddr, Remote},
+    transport::{self, metrics::SensorIo, ClientAddr, OrigDstAddr, Remote},
     transport_header::{self, NewTransportHeaderServer, SessionProtocol, TransportHeader},
     Conditional, Error, NameAddr, Never,
 };
@@ -52,7 +52,7 @@ pub struct ClientInfo {
 type FwdIo<I> = io::PrefixedIo<SensorIo<tls::server::Io<I>>>;
 pub type GatewayIo<I> = io::EitherIo<FwdIo<I>, SensorIo<tls::server::Io<I>>>;
 
-impl<T> Inbound<T> {
+impl<N> Inbound<N> {
     /// Builds a stack that handles connections that target the proxy's inbound port
     /// (i.e. without an SO_ORIGINAL_DST setting). This port behaves differently from
     /// the main proxy stack:
@@ -61,22 +61,23 @@ impl<T> Inbound<T> {
     /// 2. TLS is required;
     /// 3. A transport header is expected. It's not strictly required, as
     ///    gateways may need to accept HTTP requests from older proxy versions
-    pub fn push_direct<I, TSvc, G, GSvc>(
+    pub fn push_direct<T, I, NSvc, G, GSvc>(
         self,
         gateway: G,
     ) -> Inbound<
         impl svc::NewService<
-                listen::Addrs,
+                T,
                 Service = impl svc::Service<I, Response = (), Error = Error, Future = impl Send>,
             > + Clone,
     >
     where
+        T: Param<Remote<ClientAddr>> + Param<Option<OrigDstAddr>> + Clone + Send + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
         I: Debug + Send + Sync + Unpin + 'static,
-        T: svc::NewService<TcpEndpoint, Service = TSvc> + Clone + Send + Sync + Unpin + 'static,
-        TSvc: svc::Service<FwdIo<I>, Response = ()> + Clone + Send + Sync + Unpin + 'static,
-        TSvc::Error: Into<Error>,
-        TSvc::Future: Send + Unpin,
+        N: svc::NewService<TcpEndpoint, Service = NSvc> + Clone + Send + Sync + Unpin + 'static,
+        NSvc: svc::Service<FwdIo<I>, Response = ()> + Clone + Send + Sync + Unpin + 'static,
+        NSvc::Error: Into<Error>,
+        NSvc::Future: Send + Unpin,
         G: svc::NewService<GatewayConnection, Service = GSvc>
             + Clone
             + Send
@@ -158,7 +159,7 @@ impl<T> Inbound<T> {
                 rt.identity.clone().map(WithTransportHeaderAlpn),
                 detect_timeout,
             ))
-            .check_new_service::<listen::Addrs, I>();
+            .check_new_service::<T, I>();
 
         Inbound {
             config,
@@ -170,23 +171,35 @@ impl<T> Inbound<T> {
 
 // === impl ClientInfo ===
 
-impl TryFrom<(tls::ConditionalServerTls, listen::Addrs)> for ClientInfo {
-    type Error = RefusedNoIdentity;
+impl<T> TryFrom<(tls::ConditionalServerTls, T)> for ClientInfo
+where
+    T: Param<Option<OrigDstAddr>>,
+    T: Param<Remote<ClientAddr>>,
+{
+    type Error = Error;
 
-    fn try_from(
-        (tls, addrs): (tls::ConditionalServerTls, listen::Addrs),
-    ) -> Result<Self, Self::Error> {
+    fn try_from((tls, addrs): (tls::ConditionalServerTls, T)) -> Result<Self, Self::Error> {
         match tls {
             Conditional::Some(tls::ServerTls::Established {
                 client_id: Some(client_id),
                 negotiated_protocol,
-            }) => Ok(Self {
-                client_id,
-                alpn: negotiated_protocol,
-                client_addr: addrs.client(),
-                local_addr: addrs.target_addr(),
-            }),
-            _ => Err(RefusedNoIdentity(())),
+            }) => {
+                let local: Option<OrigDstAddr> = addrs.param();
+                let OrigDstAddr(local_addr) = local.ok_or_else(|| {
+                    tracing::warn!("No SO_ORIGINAL_DST address found!");
+                    std::io::Error::new(
+                        std::io::ErrorKind::NotFound,
+                        "No SO_ORIGINAL_DST address found",
+                    )
+                })?;
+                Ok(Self {
+                    client_id,
+                    alpn: negotiated_protocol,
+                    client_addr: addrs.param(),
+                    local_addr,
+                })
+            }
+            _ => Err(RefusedNoIdentity(()).into()),
         }
     }
 }

linkerd/app/inbound/src/lib.rs

@@ -27,10 +27,10 @@ use linkerd_app_core::{
     serve,
     svc::{self, Param},
     tls,
-    transport::{self, listen, Remote, ServerAddr},
+    transport::{self, ClientAddr, OrigDstAddr, Remote, ServerAddr},
     Error, NameMatch, ProxyRuntime,
 };
-use std::{fmt::Debug, future::Future, net::SocketAddr, time::Duration};
+use std::{convert::TryFrom, fmt::Debug, future::Future, net::SocketAddr, time::Duration};
 use tracing::{debug_span, info_span};
 
 #[derive(Clone, Debug)]
@@ -208,16 +208,17 @@ where
         }
     }
 
-    pub fn into_server<I, G, GSvc, P>(
+    pub fn into_server<T, I, G, GSvc, P>(
         self,
         server_port: u16,
         profiles: P,
         gateway: G,
     ) -> impl svc::NewService<
-        listen::Addrs,
+        T,
         Service = impl svc::Service<I, Response = (), Error = Error, Future = impl Send>,
     > + Clone
     where
+        T: Param<Remote<ClientAddr>> + Param<Option<OrigDstAddr>> + Clone + Send + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
         I: Debug + Send + Sync + Unpin + 'static,
         G: svc::NewService<direct::GatewayConnection, Service = GSvc>,
@@ -253,7 +254,7 @@ where
             ))
             .push_request_filter(require_id)
             .push(self.runtime.metrics.transport.layer_accept())
-            .push_map_target(TcpAccept::from)
+            .push_request_filter(TcpAccept::try_from)
             .push(tls::NewDetectTls::layer(
                 self.runtime.identity.clone(),
                 config.detect_protocol_timeout,
@@ -266,21 +267,28 @@ where
                     .stack
                     .push_map_target(TcpEndpoint::from)
                     .push(self.runtime.metrics.transport.layer_accept())
-                    .push_map_target(TcpAccept::port_skipped)
-                    .instrument(|_: &_| debug_span!("forward"))
+                    .push_request_filter(TcpAccept::port_skipped)
+                    .check_new_service::<T, _>()
+                    .instrument(|_: &T| debug_span!("forward"))
                     .into_inner(),
             )
-            .check_new_service::<listen::Addrs, I>()
+            .check_new_service::<T, I>()
             .push_switch(
-                PreventLoop::from(server_port),
+                PreventLoop::from(server_port).to_switch(),
                 self.push_tcp_forward(server_port)
                     .push_direct(gateway)
                     .stack
                     .instrument(|_: &_| debug_span!("direct"))
                     .into_inner(),
             )
-            .instrument(|a: &listen::Addrs| info_span!("server", port = %a.target_addr().port()))
-            .check_new_service::<listen::Addrs, I>()
+            .instrument(|a: &T| {
+                if let Some(OrigDstAddr(target_addr)) = a.param() {
+                    info_span!("server", port = target_addr.port())
+                } else {
+                    info_span!("server", port = %"<no original dst>")
+                }
+            })
+            .check_new_service::<T, I>()
             .into_inner()
     }
 }
@@ -293,11 +301,21 @@ impl From<indexmap::IndexSet<u16>> for SkipByPort {
     }
 }
 
-impl svc::Predicate<listen::Addrs> for SkipByPort {
-    type Request = svc::Either<listen::Addrs, listen::Addrs>;
+impl<T> svc::Predicate<T> for SkipByPort
+where
+    T: Param<Option<OrigDstAddr>>,
+{
+    type Request = svc::Either<T, T>;
 
-    fn check(&mut self, t: listen::Addrs) -> Result<Self::Request, Error> {
-        if !self.0.contains(&t.target_addr().port()) {
+    fn check(&mut self, t: T) -> Result<Self::Request, Error> {
+        let OrigDstAddr(addr) = t.param().ok_or_else(|| {
+            tracing::warn!("No SO_ORIGINAL_DST address found!");
+            std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                "No SO_ORIGINAL_DST address found",
+            )
+        })?;
+        if !self.0.contains(&addr.port()) {
             Ok(svc::Either::A(t))
         } else {
             Ok(svc::Either::B(t))

linkerd/app/inbound/src/prevent_loop.rs

@@ -1,7 +1,7 @@
 use crate::TcpEndpoint;
 use linkerd_app_core::{
-    svc::stack::{Either, Predicate},
-    transport::listen::Addrs,
+    svc::stack::{Either, Param, Predicate},
+    transport::addrs::OrigDstAddr,
     Error,
 };
 
@@ -10,6 +10,10 @@ use linkerd_app_core::{
 pub struct PreventLoop {
     port: u16,
 }
+#[derive(Copy, Clone, Debug)]
+pub struct SwitchLoop {
+    port: u16,
+}
 
 #[derive(Copy, Clone, Debug)]
 pub struct LoopPrevented {
@@ -34,11 +38,23 @@ impl Predicate<TcpEndpoint> for PreventLoop {
     }
 }
 
-impl Predicate<Addrs> for PreventLoop {
-    type Request = Either<Addrs, Addrs>;
+impl PreventLoop {
+    pub fn to_switch(self) -> SwitchLoop {
+        SwitchLoop { port: self.port }
+    }
+}
+
+impl<T: Param<Option<OrigDstAddr>>> Predicate<T> for SwitchLoop {
+    type Request = Either<T, T>;
 
-    fn check(&mut self, addrs: Addrs) -> Result<Either<Addrs, Addrs>, Error> {
-        let addr = addrs.target_addr();
+    fn check(&mut self, addrs: T) -> Result<Either<T, T>, Error> {
+        let OrigDstAddr(addr) = addrs.param().ok_or_else(|| {
+            tracing::warn!("No SO_ORIGINAL_DST address found!");
+            std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                "No SO_ORIGINAL_DST address found",
+            )
+        })?;
         tracing::debug!(%addr, self.port);
         if addr.port() != self.port {
             Ok(Either::A(addrs))

linkerd/app/inbound/src/target.rs

@@ -6,11 +6,17 @@ use linkerd_app_core::{
     stack_tracing,
     svc::{self, Param},
     tls,
-    transport::{self, addrs::*, listen},
+    transport::{self, addrs::*},
     transport_header::TransportHeader,
     Addr, Conditional, Error, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
 };
-use std::{convert::TryInto, net::SocketAddr, str::FromStr, sync::Arc};
+use std::{
+    convert::{TryFrom, TryInto},
+    io,
+    net::SocketAddr,
+    str::FromStr,
+    sync::Arc,
+};
 use tracing::debug;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -60,25 +66,57 @@ pub struct RequestTarget {
 // === impl TcpAccept ===
 
 impl TcpAccept {
-    pub fn port_skipped(tcp: listen::Addrs) -> Self {
-        Self {
-            target_addr: tcp.target_addr(),
-            client_addr: tcp.client(),
+    pub fn port_skipped<T>(tcp: T) -> Result<Self, io::Error>
+    where
+        T: Param<Remote<ClientAddr>> + Param<Option<OrigDstAddr>>,
+    {
+        let orig_dst: Option<OrigDstAddr> = tcp.param();
+        let OrigDstAddr(target_addr) = orig_dst.ok_or_else(|| {
+            tracing::warn!("No SO_ORIGINAL_DST address found!");
+            io::Error::new(io::ErrorKind::NotFound, "No SO_ORIGINAL_DST address found")
+        })?;
+        Ok(Self {
+            target_addr,
+            client_addr: tcp.param(),
             tls: Conditional::None(tls::NoServerTls::PortSkipped),
-        }
+        })
     }
-}
 
-impl From<tls::server::Meta<listen::Addrs>> for TcpAccept {
-    fn from((tls, addrs): tls::server::Meta<listen::Addrs>) -> Self {
+    /// Returns a `TcpAccept` for the provided TLS metadata and addresses,
+    /// determining the target address from the server's local listener address
+    /// rather than a `SO_ORIGINAL_DST` address.
+    pub fn from_local_addr<T>((tls, addrs): tls::server::Meta<T>) -> Self
+    where
+        T: Param<Remote<ClientAddr>> + Param<Local<ServerAddr>>,
+    {
+        let Local(ServerAddr(target_addr)) = addrs.param();
         Self {
-            target_addr: addrs.target_addr(),
-            client_addr: addrs.client(),
+            target_addr,
+            client_addr: addrs.param(),
             tls,
         }
     }
 }
 
+impl<T> TryFrom<tls::server::Meta<T>> for TcpAccept
+where
+    T: Param<Remote<ClientAddr>> + Param<Option<OrigDstAddr>>,
+{
+    type Error = io::Error;
+    fn try_from((tls, addrs): tls::server::Meta<T>) -> Result<Self, Self::Error> {
+        let orig_dst: Option<OrigDstAddr> = addrs.param();
+        let OrigDstAddr(target_addr) = orig_dst.ok_or_else(|| {
+            tracing::warn!("No SO_ORIGINAL_DST address found!");
+            io::Error::new(io::ErrorKind::NotFound, "No SO_ORIGINAL_DST address found")
+        })?;
+        Ok(Self {
+            target_addr,
+            client_addr: addrs.param(),
+            tls,
+        })
+    }
+}
+
 impl Param<SocketAddr> for TcpAccept {
     fn param(&self) -> SocketAddr {
         self.target_addr

linkerd/app/outbound/src/discover.rs

@@ -1,7 +1,8 @@
 use crate::{tcp, Outbound};
 use linkerd_app_core::{
-    discovery_rejected, io, profiles, svc,
-    transport::{listen, metrics::SensorIo},
+    discovery_rejected, io, profiles,
+    svc::{self, stack::Param},
+    transport::{metrics::SensorIo, OrigDstAddr},
     Error, IpMatch,
 };
 use std::convert::TryFrom;
@@ -11,16 +12,17 @@ impl<N> Outbound<N> {
     /// Discovers the profile for a TCP endpoint.
     ///
     /// Resolved services are cached and buffered.
-    pub fn push_discover<I, NSvc, P>(
+    pub fn push_discover<T, I, NSvc, P>(
         self,
         profiles: P,
     ) -> Outbound<
         impl svc::NewService<
-            listen::Addrs,
+            T,
             Service = impl svc::Service<I, Response = (), Error = Error, Future = impl Send> + Clone,
         >,
     >
     where
+        T: Param<Option<OrigDstAddr>>,
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + std::fmt::Debug + Send + Unpin + 'static,
         N: svc::NewService<tcp::Logical, Service = NSvc> + Clone + Send + 'static,
         NSvc: svc::Service<SensorIo<I>, Response = (), Error = Error> + Send + 'static,
@@ -56,8 +58,8 @@ impl<N> Outbound<N> {
             .push(rt.metrics.transport.layer_accept())
             .push_cache(config.proxy.cache_max_idle_age)
             .instrument(|a: &tcp::Accept| info_span!("server", orig_dst = %a.orig_dst))
-            .push_request_filter(tcp::Accept::try_from)
-            .check_new_service::<listen::Addrs, I>();
+            .push_request_filter(|addrs: T| tcp::Accept::try_from(addrs.param()))
+            .check_new_service::<T, I>();
 
         Outbound {
             config,