outbound: Prevent loops (#525)

This change modifies the outbound proxy to fail to build services
targetting localhost:4140 (where 4140 is the outbound port). This
prevents looping and will result in 502s.

Author: olix0r

Cargo.lock

@@ -633,6 +633,7 @@ dependencies = [
  "indexmap",
  "libc",
  "linkerd2-addr",
+ "linkerd2-admit",
  "linkerd2-box",
  "linkerd2-buffer",
  "linkerd2-cache",
@@ -700,7 +701,6 @@ dependencies = [
  "futures",
  "http",
  "indexmap",
- "linkerd2-admit",
  "linkerd2-app-core",
  "quickcheck",
  "tokio",

linkerd/app/core/Cargo.toml

@@ -21,6 +21,7 @@ hyper = "0.12"
 futures = "0.1"
 indexmap = "1.0"
 linkerd2-addr = { path = "../../addr" }
+linkerd2-admit = { path = "../../admit" }
 linkerd2-cache = { path = "../../cache" }
 linkerd2-box = { path = "../../box" }
 linkerd2-buffer = { path = "../../buffer" }

linkerd/app/core/src/lib.rs

@@ -10,6 +10,7 @@
 #![deny(warnings, rust_2018_idioms)]
 
 pub use linkerd2_addr::{self as addr, Addr, NameAddr};
+pub use linkerd2_admit as admit;
 pub use linkerd2_cache as cache;
 pub use linkerd2_conditional::Conditional;
 pub use linkerd2_drain as drain;

linkerd/app/inbound/Cargo.toml

@@ -13,7 +13,6 @@ bytes = "0.4"
 http = "0.1"
 futures = "0.1"
 indexmap = "1.0"
-linkerd2-admit = { path = "../../admit" }
 linkerd2-app-core = { path = "../core" }
 tokio = "0.1.14"
 tower = "0.1"

linkerd/app/inbound/src/lib.rs

@@ -10,9 +10,8 @@ pub use self::endpoint::{
 };
 use self::require_identity_for_ports::RequireIdentityForPorts;
 use futures::future;
-use linkerd2_admit as admit;
 use linkerd2_app_core::{
-    classify,
+    admit, classify,
     config::{ProxyConfig, ServerConfig},
     drain, dst, errors, metric_labels,
     opencensus::proto::trace::v1 as oc,

linkerd/app/inbound/src/require_identity_for_ports.rs

@@ -1,6 +1,5 @@
 use indexmap::IndexSet;
-use linkerd2_admit as admit;
-use linkerd2_app_core::transport::tls;
+use linkerd2_app_core::{admit, transport::tls};
 use std::sync::Arc;
 
 /// A connection policy that drops

linkerd/app/outbound/src/lib.rs

@@ -12,7 +12,7 @@ pub use self::endpoint::{
 use ::http::header::HOST;
 use futures::future;
 use linkerd2_app_core::{
-    classify,
+    admit, classify,
     config::{ProxyConfig, ServerConfig},
     dns, drain, dst, errors, metric_labels,
     opencensus::proto::trace::v1 as oc,
@@ -40,9 +40,11 @@ mod add_remote_ip_on_rsp;
 mod add_server_id_on_rsp;
 mod endpoint;
 mod orig_proto_upgrade;
+mod prevent_loop;
 mod require_identity_on_endpoint;
 
 use self::orig_proto_upgrade::OrigProtoUpgradeLayer;
+use self::prevent_loop::PreventLoop;
 use self::require_identity_on_endpoint::MakeRequireIdentityLayer;
 
 const EWMA_DEFAULT_RTT: Duration = Duration::from_millis(30);
@@ -156,6 +158,7 @@ impl Config {
                         let backoff = connect.backoff.clone();
                         move |_| Ok(backoff.stream())
                     }))
+                    .push(admit::AdmitLayer::new(PreventLoop::new(listen_addr.port())))
                     .push(observability.clone())
                     .push(identity_headers.clone())
                     .push(http::override_authority::Layer::new(vec![HOST.as_str(), CANONICAL_DST_HEADER]))

linkerd/app/outbound/src/prevent_loop.rs

@@ -0,0 +1,44 @@
+use super::endpoint::{HttpEndpoint, Target};
+use linkerd2_app_core::admit;
+
+/// A connection policy that drops
+#[derive(Copy, Clone, Debug)]
+pub struct PreventLoop {
+    port: u16,
+}
+
+#[derive(Copy, Clone, Debug)]
+pub struct LoopPrevented {
+    port: u16,
+}
+
+impl PreventLoop {
+    pub fn new(port: u16) -> Self {
+        Self { port }
+    }
+}
+
+impl admit::Admit<Target<HttpEndpoint>> for PreventLoop {
+    type Error = LoopPrevented;
+
+    fn admit(&mut self, ep: &Target<HttpEndpoint>) -> Result<(), Self::Error> {
+        tracing::debug!(addr = %ep.inner.addr, self.port);
+        if ep.inner.addr.ip().is_loopback() && ep.inner.addr.port() == self.port {
+            return Err(LoopPrevented { port: self.port });
+        }
+
+        Ok(())
+    }
+}
+
+impl std::fmt::Display for LoopPrevented {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "outbound requests must not target localhost:{}",
+            self.port
+        )
+    }
+}
+
+impl std::error::Error for LoopPrevented {}