feat(app/inbound): introduce Permitted<T> target (#4119)

this commit introduces a new target type to the inbound proxy's
policy enforcement middleware.

for context, this middleware is used by "lifting" a
`NewService<T, Service = S>` into an
`NewService<(HttpRoutePermit, T), Service = S>`,
where `S: svc::Service<http::Request<B>>`. this _lazily_ enforces
policy, allowing a connection to progress before eventually enforcing
authorization policy at the request-level, i.e. when a `S`-typed
`Service<Request<_>>` is yielded.

this is helpful in part because it allows us to implement generic
connection-level middleware across `T`-typed targets, while affixing
permits with information about the policy authorizing a request across
these connections.

this also allows us to inject policy enforcement in the stack before we
deal with routing traffic to `Logical` request targets.

in subsequent work, we intend to introduce `linkerd_http_prom` metrics
middleware to the inbound proxy at this level of the stack.

broadly, our middleware layers make use of our `Param<T>` and
`ExtractParam<P, T>` traits to abstract over field access from target
types. this is useful at the protocol- and connection-level layers of
our networking stack because it allows `NewService<T>` middleware like
`NewCountRequests` to be agnostic to its `T`-typed target, so long as it
(a) is provided an extractor to find a single time series from the
labeled metric family to increment given that `T`, and (b) wraps a
`NewService<T>` that yields a `Service<Request<B>`.

rust's type system does not support generic specialization, which means
that our `T: Param<U>`-shaped bounds do not play well with a `(_, T)`
tuple, which the outer layers of our inbound proxy's http router deals
with.

to facilitate the introduction of telemetry middleware wrapping our http
router, this commit introduces a `Permitted<T>` type to replace these
`(HttpRoutePermit, T)` tuples:

```rust
/// Describes an authorized `T`-typed target.
pub struct Permitted<T> {
    pub permit: HttpRoutePermit,
    pub target: T,
}
```

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

linkerd/app/admin/src/stack.rs

@@ -114,7 +114,7 @@ impl Config {
                     .to_layer::<classify::Response, _, Permitted>(),
             )
             .push(classify::NewClassify::layer_default())
-            .push_map_target(|(permit, http)| Permitted { permit, http })
+            .push_map_target(Permitted::from)
             .push(inbound::policy::NewHttpPolicy::layer(
                 metrics.http_authz.clone(),
             ))
@@ -281,6 +281,17 @@ impl Param<metrics::EndpointLabels> for Permitted {
     }
 }
 
+impl From<inbound::policy::Permitted<Http>> for Permitted {
+    fn from(
+        inbound::policy::Permitted { permit, target }: inbound::policy::Permitted<Http>,
+    ) -> Self {
+        Self {
+            permit,
+            http: target,
+        }
+    }
+}
+
 // === TlsParams ===
 
 impl<T> ExtractParam<tls::server::Timeout, T> for TlsParams {

linkerd/app/gateway/src/http.rs

@@ -1,5 +1,5 @@
 use super::Gateway;
-use inbound::{GatewayAddr, GatewayDomainInvalid};
+use inbound::{policy::Permitted, GatewayAddr, GatewayDomainInvalid};
 use linkerd_app_core::{
     metrics::ServerLabel,
     profiles,
@@ -81,50 +81,57 @@ impl Gateway {
         R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
         R::Resolution: Unpin,
     {
-        let http = self
-            .outbound
-            .clone()
-            .with_stack(inner)
-            .push_http_cached(resolve)
-            .into_stack()
-            // Discard `T` and its associated client-specific metadata.
-            .push_map_target(Target::discard_parent)
-            .push(svc::ArcNewService::layer())
-            // Add headers to prevent loops.
-            .push(NewHttpGateway::layer(
-                self.inbound.identity().local_id().clone(),
-            ))
-            .push_on_service(svc::LoadShed::layer())
-            .lift_new()
-            .push(svc::ArcNewService::layer())
-            // After protocol-downgrade, we need to build an inner stack for
-            // each request-level HTTP version.
-            .push(svc::NewOneshotRoute::layer_via(|t: &Target<T>| {
-                ByRequestVersion(t.clone())
-            }))
-            // Only permit gateway traffic to endpoints for which we have
-            // discovery information.
-            .push_filter(|(_, parent): (_, T)| -> Result<_, GatewayDomainInvalid> {
-                let routes = {
-                    let mut profile =
-                        svc::Param::<Option<watch::Receiver<profiles::Profile>>>::param(&parent)
+        let http =
+            self.outbound
+                .clone()
+                .with_stack(inner)
+                .push_http_cached(resolve)
+                .into_stack()
+                // Discard `T` and its associated client-specific metadata.
+                .push_map_target(Target::discard_parent)
+                .push(svc::ArcNewService::layer())
+                // Add headers to prevent loops.
+                .push(NewHttpGateway::layer(
+                    self.inbound.identity().local_id().clone(),
+                ))
+                .push_on_service(svc::LoadShed::layer())
+                .lift_new()
+                .push(svc::ArcNewService::layer())
+                // After protocol-downgrade, we need to build an inner stack for
+                // each request-level HTTP version.
+                .push(svc::NewOneshotRoute::layer_via(|t: &Target<T>| {
+                    ByRequestVersion(t.clone())
+                }))
+                // Only permit gateway traffic to endpoints for which we have
+                // discovery information.
+                .push_filter(
+                    |Permitted {
+                         permit: _,
+                         target: parent,
+                     }: Permitted<T>|
+                     -> Result<_, GatewayDomainInvalid> {
+                        let routes = {
+                            let mut profile = svc::Param::<
+                                Option<watch::Receiver<profiles::Profile>>,
+                            >::param(&parent)
                             .ok_or(GatewayDomainInvalid)?;
-                    let init =
-                        mk_routes(&profile.borrow_and_update()).ok_or(GatewayDomainInvalid)?;
-                    outbound::http::spawn_routes(profile, init, mk_routes)
-                };
-
-                Ok(Target {
-                    routes,
-                    addr: parent.param(),
-                    version: parent.param(),
-                    parent,
-                })
-            })
-            .push(svc::ArcNewService::layer())
-            // Authorize requests to the gateway.
-            .push(self.inbound.authorize_http())
-            .arc_new_clone_http();
+                            let init = mk_routes(&profile.borrow_and_update())
+                                .ok_or(GatewayDomainInvalid)?;
+                            outbound::http::spawn_routes(profile, init, mk_routes)
+                        };
+
+                        Ok(Target {
+                            routes,
+                            addr: parent.param(),
+                            version: parent.param(),
+                            parent,
+                        })
+                    },
+                )
+                .push(svc::ArcNewService::layer())
+                // Authorize requests to the gateway.
+                .push(self.inbound.authorize_http())
+                .arc_new_clone_http();
 
         self.inbound
             .clone()

linkerd/app/inbound/src/http/router.rs

@@ -236,12 +236,12 @@ impl<C> Inbound<C> {
                 .push_on_service(svc::LoadShed::layer())
                 .push(svc::NewMapErr::layer_from_target::<LogicalError, _>())
                 .lift_new()
-                .check_new_new::<(policy::HttpRoutePermit, T), Logical>()
+                .check_new_new::<policy::Permitted<T>, Logical>()
                 .push(svc::ArcNewService::layer())
-                .push(svc::NewOneshotRoute::layer_via(|(permit, t): &(policy::HttpRoutePermit, T)| {
-                    LogicalPerRequest::from((permit.clone(), t.clone()))
+                .push(svc::NewOneshotRoute::layer_via(|t: &policy::Permitted<T>| {
+                    LogicalPerRequest::from(t)
                 }))
-                .check_new_service::<(policy::HttpRoutePermit, T), http::Request<http::BoxBody>>()
+                .check_new_service::<policy::Permitted<T>, http::Request<http::BoxBody>>()
                 .push(svc::ArcNewService::layer())
                 .push(policy::NewHttpPolicy::layer(rt.metrics.http_authz.clone()))
                 // Used by tap.
@@ -254,12 +254,12 @@ impl<C> Inbound<C> {
 
 // === impl LogicalPerRequest ===
 
-impl<T> From<(policy::HttpRoutePermit, T)> for LogicalPerRequest
+impl<'a, T> From<&'a policy::Permitted<T>> for LogicalPerRequest
 where
     T: Param<Remote<ServerAddr>>,
     T: Param<tls::ConditionalServerTls>,
 {
-    fn from((permit, t): (policy::HttpRoutePermit, T)) -> Self {
+    fn from(policy::Permitted { permit, target }: &'a policy::Permitted<T>) -> Self {
         let labels = [
             ("srv", &permit.labels.route.server.0),
             ("route", &permit.labels.route.route),
@@ -276,9 +276,9 @@ where
         .collect::<std::collections::BTreeMap<_, _>>();
 
         Self {
-            server: t.param(),
-            tls: t.param(),
-            permit,
+            server: target.param(),
+            tls: target.param(),
+            permit: permit.clone(),
             labels: labels.into(),
         }
     }

linkerd/app/inbound/src/policy.rs

@@ -12,7 +12,7 @@ pub use self::{
     config::Config,
     http::{
         HttpInvalidPolicy, HttpRouteInvalidRedirect, HttpRouteNotFound, HttpRouteRedirect,
-        HttpRouteUnauthorized, NewHttpPolicy,
+        HttpRouteUnauthorized, NewHttpPolicy, Permitted,
     },
     tcp::NewTcpPolicy,
 };

linkerd/app/inbound/src/policy/http.rs

@@ -46,6 +46,13 @@ struct ConnectionMeta {
     tls: tls::ConditionalServerTls,
 }
 
+/// A `T`-typed target with policy enforced by a [`NewHttpPolicy<N>`] layer.
+#[derive(Debug)]
+pub struct Permitted<T> {
+    pub permit: HttpRoutePermit,
+    pub target: T,
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("no route found for request")]
 pub struct HttpRouteNotFound(());
@@ -138,7 +145,7 @@ macro_rules! try_fut {
 impl<B, T, N, S> svc::Service<::http::Request<B>> for HttpPolicyService<T, N>
 where
     T: Clone,
-    N: svc::NewService<(HttpRoutePermit, T), Service = S>,
+    N: svc::NewService<Permitted<T>, Service = S>,
     S: svc::Service<::http::Request<B>>,
     S::Error: Into<Error>,
 {
@@ -175,7 +182,10 @@ where
 
         future::Either::Left(
             self.inner
-                .new_service((permit, self.target.clone()))
+                .new_service(Permitted {
+                    permit,
+                    target: self.target.clone(),
+                })
                 .oneshot(req)
                 .err_into::<Error>(),
         )
@@ -383,3 +393,17 @@ fn apply_grpc_filters<B>(route: &grpc::Policy, req: &mut ::http::Request<B>) ->
 
     Ok(())
 }
+
+// === impl Permitted ===
+
+/// An authorized `T`-typed target can produce `P`-typed parameters.
+impl<T, P> svc::Param<P> for Permitted<T>
+where
+    T: svc::Param<P>,
+{
+    fn param(&self) -> P {
+        let Self { target, .. } = self;
+
+        target.param()
+    }
+}

linkerd/app/inbound/src/policy/http/tests.rs

@@ -39,7 +39,7 @@ macro_rules! new_svc {
             policy,
             connection: $conn,
             metrics: HttpAuthzMetrics::default(),
-            inner: |(permit, _): (HttpRoutePermit, ())| {
+            inner: |Permitted { permit, target: () }: Permitted<()>| {
                 let f = $rsp;
                 svc::mk(move |req: ::http::Request<BoxBody>| {
                     futures::future::ready((f)(permit.clone(), req))