inbound: Discover policies from the control plane (#1205)

We've recently introduced support for authorization policies. These
policies are configured statically from the proxy's environment. This
change adds an optional new mode to support dynamic configuration from
the control plane.

The proxy now supports new environment variables,
`LINKERD2_PROXY_POLICY_SVC_{ADDR,NAME}` that may be set with the address
and identity of the policy controller. When this is set, other static
inbound port configurations (requiring identity, marking ports as
opaque, etc) are ignored in favor of the results returned by the API.
Instead, the proxy uses the `LINKER2_PROXY_INBOUND_PORTS` environment
variable to discover the list of ports configured in the pod's spec and
discovers the policies for each of these ports before admitting inbound
connections. When a connection is received for a port not in this list,
the `LINKERD2_PROXY_INBOUND_DEFAULT_POLICY` configuration is used to
determine whether the connection may be permitted or whether it should
be refused.

This change modifies the set of default policies to support only `deny`,
`all-authenticated` and `all-unauthenticated` (removing
`all-mtls-unauthenticated`). Instead a
`LINKERD2_PROXY_INBOUND_PORTS_REQUIRE_TLS` configuration is introduced
to enable tls requirements for individual pods. This static
configuration should only be necessary for the identity controller, in
which case we need to use an unauthenticated default policy to permit
kubernetes healtchecking probes; and we can use this new configuration
to enforce a default policy on the gRPC API port.

Author: olix0r

.github/workflows/advisory.yml

@@ -47,7 +47,7 @@ jobs:
   # needed for release builds. This job builds the proxy in release-mode to ensure the build isn't
   # OOM-killed.
   release-binary:
-    timeout-minutes: 15
+    timeout-minutes: 20
     runs-on: ubuntu-latest
     permissions:
       contents: read

Cargo.lock

@@ -745,10 +745,13 @@ dependencies = [
  "linkerd-app-test",
  "linkerd-io",
  "linkerd-server-policy",
+ "linkerd-tonic-watch",
  "linkerd-tracing",
+ "linkerd2-proxy-api",
  "thiserror",
  "tokio",
  "tokio-test",
+ "tonic",
  "tower",
  "tracing",
 ]

linkerd/app/core/src/addr_match.rs

@@ -28,14 +28,21 @@ impl AddrMatch {
         }
     }
 
+    #[inline]
     pub fn names(&self) -> &NameMatch {
         &self.names
     }
 
+    #[inline]
     pub fn nets(&self) -> &IpMatch {
         &self.nets
     }
 
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.names.is_empty() && self.nets.is_empty()
+    }
+
     #[inline]
     pub fn matches(&self, addr: &Addr) -> bool {
         match addr {
@@ -89,6 +96,11 @@ impl FromIterator<Suffix> for NameMatch {
 }
 
 impl NameMatch {
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
     #[inline]
     pub fn matches(&self, name: &Name) -> bool {
         self.0.iter().any(|sfx| sfx.contains(name))
@@ -108,6 +120,11 @@ impl IpMatch {
         Self(Arc::new(nets.into_iter().collect()))
     }
 
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
     #[inline]
     pub fn matches(&self, addr: IpAddr) -> bool {
         self.0.iter().any(|net| match (net, addr) {

linkerd/app/core/src/control.rs

@@ -38,19 +38,16 @@ type BalanceBody =
 
 type RspBody = linkerd_http_metrics::requests::ResponseBody<BalanceBody, classify::Eos>;
 
-pub type Client<B> = svc::Buffer<http::Request<B>, http::Response<RspBody>, Error>;
+pub type Client = svc::Buffer<http::Request<tonic::body::BoxBody>, http::Response<RspBody>, Error>;
 
 impl Config {
-    pub fn build<B, L>(
+    pub fn build<L>(
         self,
         dns: dns::Resolver,
         metrics: metrics::ControlHttp,
         identity: Option<L>,
-    ) -> svc::BoxNewService<(), Client<B>>
+    ) -> svc::BoxNewService<(), Client>
     where
-        B: http::HttpBody + Send + 'static,
-        B::Data: Send,
-        B::Error: Into<Error> + Send + Sync + 'static,
         L: Clone + svc::Param<tls::client::Config> + Send + Sync + 'static,
     {
         let addr = self.addr;

linkerd/app/core/src/serve.rs

@@ -75,7 +75,8 @@ pub async fn serve<M, S, I, A>(
                 }
             }
         }
-    };
+    }
+    .in_current_span();
 
     // Stop the accept loop when the shutdown signal fires.
     //

linkerd/app/inbound/Cargo.toml

@@ -15,8 +15,11 @@ http = "0.2"
 futures = { version = "0.3", default-features = false }
 linkerd-app-core = { path = "../core" }
 linkerd-server-policy = { path = "../../server-policy" }
+linkerd-tonic-watch = { path = "../../tonic-watch" }
+linkerd2-proxy-api = { version = "0.2", features = ["client", "inbound"] }
 thiserror = "1.0"
 tokio = { version = "1", features = ["sync"] }
+tonic = { version = "0.5", default-features = false }
 tower = { version = "0.4.8", features = ["util"] }
 tracing = "0.1.26"
 

linkerd/app/inbound/src/accept.rs

@@ -1,4 +1,7 @@
-use crate::{port_policies::AllowPolicy, Inbound};
+use crate::{
+    policy::{AllowPolicy, CheckPolicy},
+    Inbound,
+};
 use linkerd_app_core::{
     io, svc,
     transport::addrs::{ClientAddr, OrigDstAddr, Remote},
@@ -7,8 +10,8 @@ use linkerd_app_core::{
 use std::fmt::Debug;
 use tracing::info_span;
 
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct Accept {
+#[derive(Clone, Debug)]
+pub(crate) struct Accept {
     client_addr: Remote<ClientAddr>,
     orig_dst_addr: OrigDstAddr,
     policy: AllowPolicy,
@@ -20,9 +23,10 @@ impl<N> Inbound<N> {
     /// Builds a stack that accepts connections. Connections to the proxy port are diverted to the
     /// 'direct' stack; otherwise connections are associated with a policy and passed to the inner
     /// stack.
-    pub fn push_accept<T, I, NSvc, D, DSvc>(
+    pub(crate) fn push_accept<T, I, NSvc, D, DSvc>(
         self,
         proxy_port: u16,
+        policies: impl CheckPolicy + Clone + Send + Sync + 'static,
         direct: D,
     ) -> Inbound<svc::BoxNewTcp<T, I>>
     where
@@ -40,8 +44,7 @@ impl<N> Inbound<N> {
         DSvc::Error: Into<Error>,
         DSvc::Future: Send,
     {
-        self.map_stack(|cfg, rt, accept| {
-            let port_policies = cfg.port_policies.clone();
+        self.map_stack(|_, rt, accept| {
             accept
                 .push_switch(
                     // Switch to the `direct` stack when a connection's original destination is the
@@ -52,10 +55,12 @@ impl<N> Inbound<N> {
                         if addr.port() == proxy_port {
                             return Ok(svc::Either::B(t));
                         }
-                        let policy = port_policies.check_allowed(t.param(), t.param())?;
+                        let orig_dst_addr = t.param();
+                        let policy = policies.check_policy(orig_dst_addr)?;
+                        tracing::debug!(?policy, "Accepted");
                         Ok(svc::Either::A(Accept {
                             client_addr: t.param(),
-                            orig_dst_addr: t.param(),
+                            orig_dst_addr,
                             policy,
                         }))
                     },
@@ -101,7 +106,10 @@ impl svc::Param<AllowPolicy> for Accept {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{test_util, DefaultPolicy, PortPolicies};
+    use crate::{
+        policy::{DefaultPolicy, Store},
+        test_util,
+    };
     use futures::future;
     use linkerd_app_core::{
         svc::{NewService, ServiceExt},
@@ -112,18 +120,21 @@ mod tests {
     #[tokio::test(flavor = "current_thread")]
     async fn default_allow() {
         let (io, _) = io::duplex(1);
-        let allow = ServerPolicy {
-            protocol: linkerd_server_policy::Protocol::Opaque,
-            authorizations: vec![Authorization {
-                authentication: Authentication::Unauthenticated,
-                networks: vec![Default::default()],
+        let policies = Store::fixed(
+            ServerPolicy {
+                protocol: linkerd_server_policy::Protocol::Opaque,
+                authorizations: vec![Authorization {
+                    authentication: Authentication::Unauthenticated,
+                    networks: vec![Default::default()],
+                    labels: Default::default(),
+                }],
                 labels: Default::default(),
-            }],
-            labels: Default::default(),
-        };
-        inbound(allow)
+            },
+            None,
+        );
+        inbound()
             .with_stack(new_ok())
-            .push_accept(999, new_panic("direct stack must not be built"))
+            .push_accept(999, policies, new_panic("direct stack must not be built"))
             .into_inner()
             .new_service(Target(1000))
             .oneshot(io)
@@ -133,10 +144,11 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn default_deny() {
+        let policies = Store::fixed(DefaultPolicy::Deny, None);
         let (io, _) = io::duplex(1);
-        inbound(DefaultPolicy::Deny)
+        inbound()
             .with_stack(new_ok())
-            .push_accept(999, new_panic("direct stack must not be built"))
+            .push_accept(999, policies, new_panic("direct stack must not be built"))
             .into_inner()
             .new_service(Target(1000))
             .oneshot(io)
@@ -146,21 +158,20 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn direct() {
+        let policies = Store::fixed(DefaultPolicy::Deny, None);
         let (io, _) = io::duplex(1);
-        inbound(DefaultPolicy::Deny)
+        inbound()
             .with_stack(new_panic("detect stack must not be built"))
-            .push_accept(999, new_ok())
+            .push_accept(999, policies, new_ok())
             .into_inner()
             .new_service(Target(999))
             .oneshot(io)
             .await
             .expect("should succeed");
     }
 
-    fn inbound(port_policies: impl Into<PortPolicies>) -> Inbound<()> {
-        let mut c = test_util::default_config();
-        c.port_policies = port_policies.into();
-        Inbound::new(c, test_util::runtime().0)
+    fn inbound() -> Inbound<()> {
+        Inbound::new(test_util::default_config(), test_util::runtime().0)
     }
 
     fn new_panic<T>(msg: &'static str) -> svc::BoxNewTcp<T, io::DuplexStream> {