Configure inbound HTTP routes via gRPC (#1814)

The policy controller serves inbound server configuration. Recent
changes have updated the inbound proxy to support applying HTTP
route-specific policies, but only a single default route configuration
was used.

This change updates the proxy to use a (not yet released) new proxy API
version that includes HTTP routes in server responses. This change adds
protobuf conversions from these types to the proxy's HTTP route types.

If the proxy receives a router filter of an unknown type (i.e., because
the controller is running a later version of the API that includes new
types), then the proxy will FAIL all requests on that route with an
internal server error. It's considered safer to fail hard in this case,
rather than to silently ignore a configured policy that could
potentially be security-sensitive.

Signed-off-by: Oliver Gould <[email protected]>

Author: olix0r

Cargo.lock

@@ -870,6 +870,7 @@ dependencies = [
  "linkerd-tonic-watch",
  "linkerd-tracing",
  "linkerd2-proxy-api",
+ "once_cell",
  "parking_lot",
  "thiserror",
  "tokio",
@@ -1145,6 +1146,8 @@ name = "linkerd-http-route"
 version = "0.1.0"
 dependencies = [
  "http",
+ "linkerd2-proxy-api",
+ "maplit",
  "rand",
  "regex",
  "thiserror",
@@ -1697,9 +1700,9 @@ dependencies = [
 
 [[package]]
 name = "linkerd2-proxy-api"
-version = "0.5.0"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12c93aba8dbdc8f465de51ef08c5e1938790ea0ae7390d66b3f4d2d36c297d0b"
+checksum = "6af1b0893c92d50e1af9a9342df7bd1ba73b9ef2abce700e170f2b2d4c84ac72"
 dependencies = [
  "h2",
  "http",
@@ -1739,6 +1742,12 @@ dependencies = [
  "linked-hash-map",
 ]
 
+[[package]]
+name = "maplit"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
+
 [[package]]
 name = "match_cfg"
 version = "0.1.0"

linkerd/app/core/src/errors/respond.rs

@@ -91,11 +91,15 @@ where
 
 impl SyntheticHttpResponse {
     pub fn unexpected_error() -> Self {
+        Self::internal_error("unexpected error")
+    }
+
+    pub fn internal_error(msg: impl Into<Cow<'static, str>>) -> Self {
         Self {
             close_connection: true,
             http_status: http::StatusCode::INTERNAL_SERVER_ERROR,
             grpc_status: tonic::Code::Internal,
-            message: Cow::Borrowed("unexpected error"),
+            message: msg.into(),
             location: None,
         }
     }

linkerd/app/inbound/Cargo.toml

@@ -18,7 +18,8 @@ linkerd-cache = { path = "../../cache" }
 linkerd-http-access-log = { path = "../../http-access-log" }
 linkerd-server-policy = { path = "../../server-policy", features = ["proto"] }
 linkerd-tonic-watch = { path = "../../tonic-watch" }
-linkerd2-proxy-api = { version = "0.5", features = ["inbound"] }
+linkerd2-proxy-api = { version = "0.6", features = ["inbound"] }
+once_cell = "1"
 parking_lot = "0.12"
 thiserror = "1"
 tokio = { version = "1", features = ["sync"] }

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

@@ -144,6 +144,11 @@ impl errors::HttpRescue<Error> for ServerRescue {
         {
             return Ok(errors::SyntheticHttpResponse::redirect(*status, location));
         }
+        if let Some(cause) = errors::cause_ref::<policy::HttpInvalidPolicy>(&*error) {
+            return Ok(errors::SyntheticHttpResponse::internal_error(
+                cause.to_string(),
+            ));
+        }
 
         if let Some(cause) = errors::cause_ref::<crate::GatewayDomainInvalid>(&*error) {
             return Ok(errors::SyntheticHttpResponse::not_found(cause));

linkerd/app/inbound/src/policy.rs

@@ -9,8 +9,8 @@ pub(crate) use self::store::Store;
 pub use self::{
     config::Config,
     http::{
-        HttpRouteInvalidRedirect, HttpRouteNotFound, HttpRouteRedirect, HttpRouteUnauthorized,
-        NewHttpPolicy,
+        HttpInvalidPolicy, HttpRouteInvalidRedirect, HttpRouteNotFound, HttpRouteRedirect,
+        HttpRouteUnauthorized, NewHttpPolicy,
     },
     tcp::NewTcpPolicy,
 };

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

@@ -10,10 +10,12 @@ use linkerd_app_core::{
 };
 use linkerd_server_policy::ServerPolicy;
 use linkerd_tonic_watch::StreamWatch;
+use std::time;
 
 #[derive(Clone, Debug)]
 pub(super) struct Api<S> {
     workload: String,
+    detect_timeout: time::Duration,
     client: Client<S>,
 }
 
@@ -22,15 +24,20 @@ pub(super) struct GrpcRecover(ExponentialBackoff);
 
 pub(super) type Watch<S> = StreamWatch<GrpcRecover, Api<S>>;
 
+/// If an invalid policy is encountered, then this will be updated to hold a
+/// default, invalid policy.
+static INVALID_POLICY: once_cell::sync::OnceCell<ServerPolicy> = once_cell::sync::OnceCell::new();
+
 impl<S> Api<S>
 where
     S: tonic::client::GrpcService<tonic::body::BoxBody, Error = Error> + Clone,
     S::ResponseBody:
         http::HttpBody<Data = tonic::codegen::Bytes, Error = Error> + Default + Send + 'static,
 {
-    pub(super) fn new(workload: String, client: S) -> Self {
+    pub(super) fn new(workload: String, detect_timeout: time::Duration, client: S) -> Self {
         Self {
             workload,
+            detect_timeout,
             client: Client::new(client),
         }
     }
@@ -65,20 +72,24 @@ where
             port: port.into(),
             workload: self.workload.clone(),
         };
+        let detect_timeout = self.detect_timeout;
         let mut client = self.client.clone();
         Box::pin(async move {
             let rsp = client.watch_port(tonic::Request::new(req)).await?;
             Ok(rsp.map(|updates| {
                 updates
-                    .map(|up| {
-                        let policy = up?.try_into().map_err(|e| {
-                            tonic::Status::invalid_argument(&*format!(
-                                "received invalid policy: {}",
-                                e
-                            ))
-                        })?;
+                    .map_ok(move |up| {
+                        // If the server returned an invalid server policy, we
+                        // default to using an invalid policy that causes all
+                        // requests to report an internal error.
+                        let policy = ServerPolicy::try_from(up).unwrap_or_else(|error| {
+                            tracing::warn!(%error, "Server misconfigured");
+                            INVALID_POLICY
+                                .get_or_init(|| ServerPolicy::invalid(detect_timeout))
+                                .clone()
+                        });
                         tracing::debug!(?policy);
-                        Ok(policy)
+                        policy
                     })
                     .boxed()
             }))

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

@@ -1,4 +1,4 @@
-use super::{api::Api, DefaultPolicy, GetPolicy, ServerPolicy, Store};
+use super::{api::Api, DefaultPolicy, GetPolicy, Protocol, ServerPolicy, Store};
 use linkerd_app_core::{control, dns, identity, metrics, svc::NewService};
 use std::collections::{HashMap, HashSet};
 use tokio::time::Duration;
@@ -49,8 +49,15 @@ impl Config {
             } => {
                 let watch = {
                     let backoff = control.connect.backoff;
-                    let c = control.build(dns, metrics, identity).new_service(());
-                    Api::new(workload, c).into_watch(backoff)
+                    let client = control.build(dns, metrics, identity).new_service(());
+                    let detect_timeout = match default {
+                        DefaultPolicy::Allow(ServerPolicy {
+                            protocol: Protocol::Detect { timeout, .. },
+                            ..
+                        }) => timeout,
+                        _ => Duration::from_secs(10),
+                    };
+                    Api::new(workload, detect_timeout, client).into_watch(backoff)
                 };
                 Store::spawn_discover(default, cache_max_idle_age, watch, ports)
             }

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

@@ -2,7 +2,7 @@ use linkerd_app_core::{IpNet, Ipv4Net, Ipv6Net};
 use linkerd_server_policy::{
     authz::Suffix, http, Authentication, Authorization, Meta, Protocol, ServerPolicy,
 };
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 pub fn all_authenticated(timeout: Duration) -> ServerPolicy {
     mk("all-authenticated", all_nets(), authenticated(), timeout)
@@ -62,18 +62,22 @@ fn mk(
     authentication: Authentication,
     timeout: Duration,
 ) -> ServerPolicy {
-    let authorizations = std::sync::Arc::new([Authorization {
+    let authorizations = Arc::new([Authorization {
+        meta: Meta::new_default(name),
         networks: nets.into_iter().map(Into::into).collect(),
         authentication,
-        meta: Meta::new_default(name),
     }]);
-    let http = std::sync::Arc::new([http::default(authorizations.clone())]);
+
+    // The default policy supports protocol detection and uses the default
+    // authorization policy on a default route that matches all requests.
+    let protocol = Protocol::Detect {
+        timeout,
+        http: Arc::new([http::default(authorizations.clone())]),
+        tcp_authorizations: authorizations,
+    };
+
     ServerPolicy {
-        protocol: Protocol::Detect {
-            timeout,
-            http,
-            tcp_authorizations: authorizations,
-        },
         meta: Meta::new_default(name),
+        protocol,
     }
 }

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

@@ -79,6 +79,10 @@ pub struct GrpcRouteInjectedFailure {
     pub message: Arc<str>,
 }
 
+#[derive(Debug, thiserror::Error)]
+#[error("invalid server policy: {0}")]
+pub struct HttpInvalidPolicy(&'static str);
+
 // === impl NewHttpPolicy ===
 
 impl<N> NewHttpPolicy<N> {
@@ -285,6 +289,10 @@ fn apply_http_filters<B>(
             http::Filter::RequestHeaders(rh) => {
                 rh.apply(req.headers_mut());
             }
+
+            http::Filter::InternalError(msg) => {
+                return Err(HttpInvalidPolicy(msg).into());
+            }
         }
     }
 
@@ -303,6 +311,10 @@ fn apply_grpc_filters<B>(route: &grpc::Policy, req: &mut ::http::Request<B>) ->
             grpc::Filter::RequestHeaders(rh) => {
                 rh.apply(req.headers_mut());
             }
+
+            grpc::Filter::InternalError(msg) => {
+                return Err(HttpInvalidPolicy(msg).into());
+            }
         }
     }