policy: add filters for TCP and TLS policy

linkerd/app/outbound/src/opaq/logical/route.rs

@@ -2,12 +2,16 @@ use super::{super::Concrete, Logical};
 use crate::RouteRef;
 use linkerd_app_core::{io, svc, Error};
 use linkerd_distribute as distribute;
-use std::{fmt::Debug, hash::Hash};
+use linkerd_proxy_client_policy as policy;
+use std::{fmt::Debug, hash::Hash, sync::Arc};
+
+mod filters;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub(crate) struct Backend<T> {
     pub(crate) route_ref: RouteRef,
     pub(crate) concrete: Concrete<T>,
+    pub(super) filters: Arc<[policy::opaq::Filter]>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -20,6 +24,7 @@ pub(crate) struct Route<T> {
     pub(super) parent: T,
     pub(super) logical: Logical,
     pub(super) route_ref: RouteRef,
+    pub(super) filters: Arc<[policy::opaq::Filter]>,
     pub(super) distribution: BackendDistribution<T>,
 }
 
@@ -42,6 +47,7 @@ impl<T: Clone> Clone for Backend<T> {
         Self {
             route_ref: self.route_ref.clone(),
             concrete: self.concrete.clone(),
+            filters: self.filters.clone(),
         }
     }
 }
@@ -71,12 +77,16 @@ where
             svc::stack(inner)
                 .push_map_target(|t| t)
                 .push_map_target(|b: Backend<T>| b.concrete)
+                // apply backend filters
+                .push_filter(filters::apply)
                 .lift_new()
                 .push(NewDistribute::layer())
                 // The router does not take the backend's availability into
                 // consideration, so we must eagerly fail requests to prevent
                 // leaking tasks onto the runtime.
                 .push_on_service(svc::LoadShed::layer())
+                // apply route level filters
+                .push_filter(filters::apply)
                 .push(svc::NewMapErr::layer_with(|rt: &Self| {
                     let route = rt.params.route_ref.clone();
                     move |source| RouteError {
@@ -95,3 +105,15 @@ impl<T: Clone> svc::Param<BackendDistribution<T>> for MatchedRoute<T> {
         self.params.distribution.clone()
     }
 }
+
+impl<T: Clone> svc::Param<Arc<[policy::opaq::Filter]>> for MatchedRoute<T> {
+    fn param(&self) -> Arc<[policy::opaq::Filter]> {
+        self.params.filters.clone()
+    }
+}
+
+impl<T: Clone> svc::Param<Arc<[policy::opaq::Filter]>> for Backend<T> {
+    fn param(&self) -> Arc<[policy::opaq::Filter]> {
+        self.filters.clone()
+    }
+}

linkerd/app/outbound/src/opaq/logical/route/filters.rs

@@ -0,0 +1,44 @@
+use linkerd_app_core::{svc, Error};
+use linkerd_proxy_client_policy::opaq;
+use std::{fmt::Debug, sync::Arc};
+
+pub(crate) fn apply<T>(t: T) -> Result<T, Error>
+where
+    T: Clone,
+    T: svc::Param<Arc<[opaq::Filter]>>,
+{
+    let filters: &[opaq::Filter] = &t.param();
+    if let Some(filter) = filters.iter().next() {
+        match filter {
+            opaq::Filter::ForbiddenRoute => {
+                return Err(errors::TCPForbiddenRoute.into());
+            }
+
+            opaq::Filter::InvalidBackend(message) => {
+                return Err(errors::TCPInvalidBackend(message.clone()).into());
+            }
+
+            opaq::Filter::InternalError(message) => {
+                return Err(errors::TCPInvalidPolicy(message).into());
+            }
+        }
+    }
+
+    Ok(t)
+}
+
+pub mod errors {
+    use super::*;
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("forbidden TCP route")]
+    pub struct TCPForbiddenRoute;
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("invalid TCP backend: {0}")]
+    pub struct TCPInvalidBackend(pub Arc<str>);
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("invalid client policy: {0}")]
+    pub struct TCPInvalidPolicy(pub &'static str);
+}

linkerd/app/outbound/src/opaq/logical/router.rs

@@ -100,6 +100,7 @@ where
                 let concrete = mk_dispatch(&rb.backend);
                 route::Backend {
                     route_ref: route_ref.clone(),
+                    filters: rb.filters.clone(),
                     concrete,
                 }
             };
@@ -122,21 +123,24 @@ where
                 }
             };
 
-        let mk_policy =
-            |policy::RoutePolicy::<policy::opaq::Filter, ()> {
-                 meta, distribution, ..
-             }| {
-                let route_ref = RouteRef(meta);
-                let logical = logical.clone();
-
-                let distribution = mk_distribution(&route_ref, &distribution);
-                route::Route {
-                    logical,
-                    parent: parent.clone(),
-                    route_ref,
-                    distribution,
-                }
-            };
+        let mk_policy = |policy::RoutePolicy::<policy::opaq::Filter, ()> {
+                             meta,
+                             distribution,
+                             filters,
+                             ..
+                         }| {
+            let route_ref = RouteRef(meta);
+            let logical = logical.clone();
+
+            let distribution = mk_distribution(&route_ref, &distribution);
+            route::Route {
+                logical,
+                parent: parent.clone(),
+                route_ref,
+                filters,
+                distribution,
+            }
+        };
 
         let routes = routes.as_ref().map(|route| opaq_route::Route {
             policy: mk_policy(route.policy.clone()),

linkerd/app/outbound/src/tls/logical/route.rs

@@ -2,13 +2,17 @@ use super::super::Concrete;
 use crate::{ParentRef, RouteRef};
 use linkerd_app_core::{io, svc, Addr, Error};
 use linkerd_distribute as distribute;
+use linkerd_proxy_client_policy as policy;
 use linkerd_tls_route as tls_route;
-use std::{fmt::Debug, hash::Hash};
+use std::{fmt::Debug, hash::Hash, sync::Arc};
+
+mod filters;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub(crate) struct Backend<T> {
     pub(crate) route_ref: RouteRef,
     pub(crate) concrete: Concrete<T>,
+    pub(super) filters: Arc<[policy::tls::Filter]>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -23,6 +27,7 @@ pub(crate) struct Route<T> {
     pub(super) addr: Addr,
     pub(super) parent_ref: ParentRef,
     pub(super) route_ref: RouteRef,
+    pub(super) filters: Arc<[policy::tls::Filter]>,
     pub(super) distribution: BackendDistribution<T>,
 }
 
@@ -45,6 +50,7 @@ impl<T: Clone> Clone for Backend<T> {
         Self {
             route_ref: self.route_ref.clone(),
             concrete: self.concrete.clone(),
+            filters: self.filters.clone(),
         }
     }
 }
@@ -74,12 +80,16 @@ where
             svc::stack(inner)
                 .push_map_target(|t| t)
                 .push_map_target(|b: Backend<T>| b.concrete)
+                // apply backend filters
+                .push_filter(filters::apply)
                 .lift_new()
                 .push(NewDistribute::layer())
                 // The router does not take the backend's availability into
                 // consideration, so we must eagerly fail requests to prevent
                 // leaking tasks onto the runtime.
                 .push_on_service(svc::LoadShed::layer())
+                // apply route level filters
+                .push_filter(filters::apply)
                 .push(svc::NewMapErr::layer_with(|rt: &Self| {
                     let route = rt.params.route_ref.clone();
                     move |source| RouteError {
@@ -98,3 +108,15 @@ impl<T: Clone> svc::Param<BackendDistribution<T>> for MatchedRoute<T> {
         self.params.distribution.clone()
     }
 }
+
+impl<T: Clone> svc::Param<Arc<[policy::tls::Filter]>> for MatchedRoute<T> {
+    fn param(&self) -> Arc<[policy::tls::Filter]> {
+        self.params.filters.clone()
+    }
+}
+
+impl<T: Clone> svc::Param<Arc<[policy::tls::Filter]>> for Backend<T> {
+    fn param(&self) -> Arc<[policy::tls::Filter]> {
+        self.filters.clone()
+    }
+}

linkerd/app/outbound/src/tls/logical/route/filters.rs

@@ -0,0 +1,44 @@
+use linkerd_app_core::{svc, Error};
+use linkerd_proxy_client_policy::tls;
+use std::{fmt::Debug, sync::Arc};
+
+pub(crate) fn apply<T>(t: T) -> Result<T, Error>
+where
+    T: Clone,
+    T: svc::Param<Arc<[tls::Filter]>>,
+{
+    let filters: &[tls::Filter] = &t.param();
+    if let Some(filter) = filters.iter().next() {
+        match filter {
+            tls::Filter::ForbiddenRoute => {
+                return Err(errors::TlSForbiddenRoute.into());
+            }
+
+            tls::Filter::InvalidBackend(message) => {
+                return Err(errors::TLSInvalidBackend(message.clone()).into());
+            }
+
+            tls::Filter::InternalError(message) => {
+                return Err(errors::TLSInvalidPolicy(message).into());
+            }
+        }
+    }
+
+    Ok(t)
+}
+
+pub mod errors {
+    use super::*;
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("forbidden TLS route")]
+    pub struct TlSForbiddenRoute;
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("invalid TLS backend: {0}")]
+    pub struct TLSInvalidBackend(pub Arc<str>);
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("invalid client policy: {0}")]
+    pub struct TLSInvalidPolicy(pub &'static str);
+}

linkerd/app/outbound/src/tls/logical/router.rs

@@ -105,6 +105,7 @@ where
                 let concrete = mk_dispatch(&rb.backend);
                 route::Backend {
                     route_ref: route_ref.clone(),
+                    filters: rb.filters.clone(),
                     concrete,
                 }
             };
@@ -127,22 +128,25 @@ where
                 }
             };
 
-        let mk_policy =
-            |policy::RoutePolicy::<policy::tls::Filter, ()> {
-                 meta, distribution, ..
-             }| {
-                let route_ref = RouteRef(meta);
-                let parent_ref = parent_ref.clone();
-
-                let distribution = mk_distribution(&route_ref, &distribution);
-                route::Route {
-                    addr: addr.clone(),
-                    parent: parent.clone(),
-                    parent_ref: parent_ref.clone(),
-                    route_ref,
-                    distribution,
-                }
-            };
+        let mk_policy = |policy::RoutePolicy::<policy::tls::Filter, ()> {
+                             meta,
+                             distribution,
+                             filters,
+                             ..
+                         }| {
+            let route_ref = RouteRef(meta);
+            let parent_ref = parent_ref.clone();
+
+            let distribution = mk_distribution(&route_ref, &distribution);
+            route::Route {
+                addr: addr.clone(),
+                parent: parent.clone(),
+                parent_ref: parent_ref.clone(),
+                route_ref,
+                filters,
+                distribution,
+            }
+        };
 
         let routes = routes
             .iter()

linkerd/proxy/client-policy/src/lib.rs

@@ -172,9 +172,20 @@ impl ClientPolicy {
                     routes: HTTP_ROUTES.clone(),
                     failure_accrual: Default::default(),
                 },
-                // TODO(ver): Include a route with a filter that emits errors
-                // for connections.
-                opaque: opaq::Opaque { routes: None },
+
+                opaque: opaq::Opaque {
+                    routes: Some(opaq::Route {
+                        policy: opaq::Policy {
+                            meta: META.clone(),
+                            filters: std::iter::once(opaq::Filter::InternalError(
+                                "invalid client policy configuration",
+                            ))
+                            .collect(),
+                            distribution: RouteDistribution::Empty,
+                            params: (),
+                        },
+                    }),
+                },
             },
             backends: BACKENDS.clone(),
         }