admin: Enforce policy on admin requests (#1250)

Previously, the admin server admitted all requests. Now the admin server
will apply policies as discovered from the control plane. The proxy's
default policy is used before policy has been synced from the
controller.

Author: olix0r

linkerd/app/admin/src/stack.rs

@@ -38,6 +38,7 @@ struct UnexpectedSni(tls::ServerId, Remote<ClientAddr>);
 
 #[derive(Clone, Debug)]
 struct Tcp {
+    policy: inbound::policy::AllowPolicy,
     addr: Local<ServerAddr>,
     client: Remote<ClientAddr>,
     tls: tls::ConditionalServerTls,
@@ -49,6 +50,12 @@ struct Http {
     version: http::Version,
 }
 
+#[derive(Clone, Debug)]
+struct Permitted {
+    permit: inbound::policy::Permit,
+    http: Http,
+}
+
 #[derive(Clone)]
 struct TlsParams {
     identity: Option<LocalCrtKey>,
@@ -63,27 +70,36 @@ impl Config {
     pub fn build<B, R>(
         self,
         bind: B,
+        policy: impl inbound::policy::CheckPolicy,
         identity: Option<LocalCrtKey>,
         report: R,
         metrics: inbound::Metrics,
         trace: trace::Handle,
         drain: drain::Watch,
         shutdown: mpsc::UnboundedSender<()>,
-    ) -> Result<Task, Error>
+    ) -> Result<Task>
     where
         R: FmtMetrics + Clone + Send + Sync + Unpin + 'static,
         B: Bind<ServerConfig>,
         B::Addrs: svc::Param<Remote<ClientAddr>> + svc::Param<Local<ServerAddr>>,
     {
         let (listen_addr, listen) = bind.bind(&self.server)?;
 
+        // Get the policy for the admin server.
+        let policy = policy.check_policy(OrigDstAddr(listen_addr.into()))?;
+
         let (ready, latch) = crate::server::Readiness::new();
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);
         let admin = svc::stack(move |_| admin.clone())
-            .push(metrics.proxy.http_endpoint.to_layer::<classify::Response, _, Http>())
+            .push(metrics.proxy.http_endpoint.to_layer::<classify::Response, _, Permitted>())
+            .push_map_target(|(permit, http)| Permitted { permit, http })
+            .push(inbound::policy::NewAuthorizeHttp::layer(metrics.http_authz.clone()))
             .push_on_service(
                 svc::layers()
                     .push(errors::NewRespond::layer(|error: Error| -> Result<_> {
+                        if error.is::<inbound::policy::DeniedUnauthorized> () {
+                            return Ok(errors::SyntheticHttpResponse::permission_denied(error));
+                        }
                         tracing::warn!(%error, "Unexpected error");
                         Ok(errors::SyntheticHttpResponse::unexpected_error())
                     }))
@@ -135,12 +151,11 @@ impl Config {
             .push(detect::NewDetectService::layer(detect::Config::<http::DetectHttp>::from_timeout(DETECT_TIMEOUT)))
             .push(transport::metrics::NewServer::layer(metrics.proxy.transport))
             .push_map_target(move |(tls, addrs): (tls::ConditionalServerTls, B::Addrs)| {
-                // TODO(ver): We should enforce policy here; but we need to permit liveness probes
-                // for destination pods to startup...
                 Tcp {
                     tls,
                     client: addrs.param(),
                     addr: addrs.param(),
+                    policy: policy.clone(),
                 }
             })
             .push(svc::BoxNewService::layer())
@@ -165,8 +180,7 @@ impl Param<transport::labels::Key> for Tcp {
         transport::labels::Key::inbound_server(
             self.tls.clone(),
             self.addr.into(),
-            // TODO(ver) enforce policies on the proxy's admin port.
-            metrics::ServerLabel("default:admin".to_string()),
+            self.policy.server_label(),
         )
     }
 }
@@ -185,22 +199,39 @@ impl Param<OrigDstAddr> for Http {
     }
 }
 
+impl Param<Remote<ClientAddr>> for Http {
+    fn param(&self) -> Remote<ClientAddr> {
+        self.tcp.client
+    }
+}
+
+impl Param<tls::ConditionalServerTls> for Http {
+    fn param(&self) -> tls::ConditionalServerTls {
+        self.tcp.tls.clone()
+    }
+}
+
+impl Param<inbound::policy::AllowPolicy> for Http {
+    fn param(&self) -> inbound::policy::AllowPolicy {
+        self.tcp.policy.clone()
+    }
+}
+
 impl Param<metrics::ServerLabel> for Http {
     fn param(&self) -> metrics::ServerLabel {
-        metrics::ServerLabel("default:admin".to_string())
+        self.tcp.policy.server_label()
     }
 }
 
-impl Param<metrics::EndpointLabels> for Http {
+// === impl Permitted ===
+
+impl Param<metrics::EndpointLabels> for Permitted {
     fn param(&self) -> metrics::EndpointLabels {
         metrics::InboundEndpointLabels {
-            tls: self.tcp.tls.clone(),
+            tls: self.http.tcp.tls.clone(),
             authority: None,
-            target_addr: self.tcp.addr.into(),
-            policy: metrics::AuthzLabels {
-                server: self.param(),
-                authz: "default:all-unauthenticated".to_string(),
-            },
+            target_addr: self.http.tcp.addr.into(),
+            policy: self.permit.labels.clone(),
         }
         .into()
     }

linkerd/app/inbound/src/metrics/mod.rs

@@ -16,7 +16,7 @@ pub use linkerd_app_core::metrics::*;
 /// Holds outbound proxy metrics.
 #[derive(Clone, Debug)]
 pub struct Metrics {
-    pub(crate) http_authz: authz::HttpAuthzMetrics,
+    pub http_authz: authz::HttpAuthzMetrics,
     pub http_errors: error::HttpErrorMetrics,
 
     pub(crate) tcp_authz: authz::TcpAuthzMetrics,

linkerd/app/src/env.rs

@@ -470,6 +470,9 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
         allow_discovery: gateway_suffixes?.into_iter().flatten().collect(),
     };
 
+    let admin_listener_addr = admin_listener_addr?
+        .unwrap_or_else(|| parse_socket_addr(DEFAULT_ADMIN_LISTEN_ADDR).unwrap());
+
     let inbound = {
         let addr = ListenAddr(
             inbound_listener_addr?
@@ -552,6 +555,9 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
                         ports.insert(inbound_port);
                     }
 
+                    // Ensure that the admin server port is included in policy discovery.
+                    ports.insert(admin_listener_addr.port());
+
                     // The workload, which is opaque from the proxy's point-of-view, is sent to the
                     // policy controller to support policy discovery.
                     let workload = strings.get(ENV_POLICY_WORKLOAD)?.ok_or_else(|| {
@@ -723,10 +729,7 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
     let admin = super::admin::Config {
         metrics_retain_idle: metrics_retain_idle?.unwrap_or(DEFAULT_METRICS_RETAIN_IDLE),
         server: ServerConfig {
-            addr: ListenAddr(
-                admin_listener_addr?
-                    .unwrap_or_else(|| parse_socket_addr(DEFAULT_ADMIN_LISTEN_ADDR).unwrap()),
-            ),
+            addr: ListenAddr(admin_listener_addr),
             keepalive: inbound.proxy.server.keepalive,
             h2_settings,
         },

linkerd/app/src/lib.rs

@@ -151,16 +151,24 @@ impl Config {
         let inbound = Inbound::new(inbound, runtime.clone());
         let outbound = Outbound::new(outbound, runtime);
 
+        let inbound_policies = {
+            let dns = dns.resolver;
+            let metrics = metrics.control;
+            info_span!("policy").in_scope(|| inbound.build_policies(dns, metrics))
+        };
+
         let admin = {
             let identity = identity.local();
             let metrics = inbound.metrics();
+            let policy = inbound_policies.clone();
             let report = inbound
                 .metrics()
                 .and_then(outbound.metrics())
                 .and_then(report);
             info_span!("admin").in_scope(move || {
                 admin.build(
                     bind_admin,
+                    policy,
                     identity,
                     report,
                     metrics,
@@ -195,9 +203,7 @@ impl Config {
             let identity = identity.local();
             let inbound_addr = inbound_addr;
             let profiles = dst.profiles;
-            let dns = dns.resolver;
             let resolve = dst.resolve;
-            let control_metrics = metrics.control;
 
             Box::pin(async move {
                 Self::await_identity(identity)
@@ -210,9 +216,6 @@ impl Config {
                         .instrument(info_span!("outbound")),
                 );
 
-                let inbound_policies =
-                    info_span!("policy").in_scope(|| inbound.build_policies(dns, control_metrics));
-
                 tokio::spawn(
                     inbound
                         .serve(