Expose policy labels on inbound transport metrics (#1215)

The inbound server only permits connections that are allowed by a
policy, but this decision is not observable at runtime.

This change modifies the server transport labels to include server and
authorization labels--prefixed by `srv_` and `saz_`, respectively
(matching the shortnames of the `Server` and `ServerAuthorization`
resources).

Furthermore, this change updates the `inbound_tcp_acept_errors_total`
metric to include an `unauthorized` error variant to expose a counter of
connections that were not permitted.

Author: olix0r

Cargo.lock

@@ -690,6 +690,7 @@ dependencies = [
  "linkerd-proxy-transport",
  "linkerd-reconnect",
  "linkerd-retry",
+ "linkerd-server-policy",
  "linkerd-service-profiles",
  "linkerd-stack",
  "linkerd-stack-metrics",

linkerd/app/admin/src/stack.rs

@@ -157,11 +157,13 @@ impl Config {
 
 impl Param<transport::labels::Key> for Tcp {
     fn param(&self) -> transport::labels::Key {
-        transport::labels::Key::Accept {
-            direction: transport::labels::Direction::In,
-            tls: self.tls.clone(),
-            target_addr: self.addr.into(),
-        }
+        transport::labels::Key::inbound_server(
+            self.tls.clone(),
+            self.addr.into(),
+            // TODO(ver) enforce policies on the proxy's admin port.
+            Default::default(),
+            Default::default(),
+        )
     }
 }
 

linkerd/app/core/Cargo.toml

@@ -50,6 +50,7 @@ linkerd-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd-proxy-transport = { path = "../../proxy/transport" }
 linkerd-reconnect = { path = "../../reconnect" }
 linkerd-retry = { path = "../../retry" }
+linkerd-server-policy = { path = "../../server-policy" }
 linkerd-timeout = { path = "../../timeout" }
 linkerd-tracing = { path = "../../tracing" }
 linkerd-transport-header = { path = "../../transport-header" }

linkerd/app/core/src/metrics/tcp_accept_errors.rs

@@ -1,7 +1,7 @@
 use crate::{
     metrics::{self, Counter, FmtMetrics},
     svc,
-    transport::{labels, OrigDstAddr},
+    transport::{labels, DeniedUnauthorized, DeniedUnknownPort, OrigDstAddr},
 };
 use linkerd_error::Error;
 use linkerd_error_metrics::{FmtLabels, LabelError, RecordError};
@@ -36,6 +36,7 @@ pub struct LabelAcceptErrors(());
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
 pub enum AcceptErrors {
     TlsDetectTimeout,
+    Unauthorized,
     Io,
     Other,
 }
@@ -94,6 +95,10 @@ impl LabelError<Error> for LabelAcceptErrors {
             } else if err.is::<std::io::Error>() {
                 // We ignore the error code because we want all labels to be consistent.
                 return AcceptErrors::Io;
+            } else if err.is::<DeniedUnknownPort>() || err.is::<DeniedUnauthorized>() {
+                // If the port is unknown, the default policy is `deny`; so handle it as
+                // unauthorized.
+                return AcceptErrors::Unauthorized;
             }
             curr = err.source();
         }
@@ -110,6 +115,7 @@ impl FmtLabels for AcceptErrors {
             Self::TlsDetectTimeout => fmt::Display::fmt("error=\"tls_detect_timeout\"", f),
             Self::Io => fmt::Display::fmt("error=\"io\"", f),
             Self::Other => fmt::Display::fmt("error=\"other\"", f),
+            Self::Unauthorized => fmt::Display::fmt("error=\"unauthorized\"", f),
         }
     }
 }

linkerd/app/core/src/transport/labels.rs

@@ -1,6 +1,7 @@
 pub use crate::metrics::{Direction, OutboundEndpointLabels};
 use linkerd_conditional::Conditional;
 use linkerd_metrics::FmtLabels;
+use linkerd_server_policy as policy;
 use linkerd_tls as tls;
 use std::{fmt, net::SocketAddr};
 
@@ -11,13 +12,17 @@ use std::{fmt, net::SocketAddr};
 /// Implements `FmtLabels`.
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
 pub enum Key {
-    Accept {
-        direction: Direction,
-        tls: tls::ConditionalServerTls,
-        target_addr: SocketAddr,
-    },
-    OutboundConnect(OutboundEndpointLabels),
-    InboundConnect,
+    Server(ServerLabels),
+    OutboundClient(OutboundEndpointLabels),
+    InboundClient,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+pub struct ServerLabels {
+    direction: Direction,
+    tls: tls::ConditionalServerTls,
+    target_addr: SocketAddr,
+    policy: Option<PolicyLabels>,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
@@ -29,40 +34,45 @@ pub(crate) struct TlsConnect<'t>(&'t tls::ConditionalClientTls);
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub(crate) struct TargetAddr(pub(crate) SocketAddr);
 
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+pub(crate) struct PolicyLabels {
+    server: policy::Labels,
+    authz: policy::Labels,
+}
+
 // === impl Key ===
 
 impl Key {
-    pub fn accept(
-        direction: Direction,
+    pub fn inbound_server(
         tls: tls::ConditionalServerTls,
         target_addr: SocketAddr,
+        server: policy::Labels,
+        authz: policy::Labels,
     ) -> Self {
-        Self::Accept {
-            direction,
+        Self::Server(ServerLabels::inbound(
             tls,
             target_addr,
-        }
+            PolicyLabels { server, authz },
+        ))
+    }
+
+    pub fn outbound_server(target_addr: SocketAddr) -> Self {
+        Self::Server(ServerLabels::outbound(target_addr))
     }
 }
 
 impl FmtLabels for Key {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::Accept {
-                direction,
-                tls,
-                target_addr,
-            } => {
-                direction.fmt_labels(f)?;
-                f.write_str(",peer=\"src\",")?;
-                (TargetAddr(*target_addr), TlsAccept::from(tls)).fmt_labels(f)
-            }
-            Self::OutboundConnect(endpoint) => {
+            Self::Server(l) => l.fmt_labels(f),
+
+            Self::OutboundClient(endpoint) => {
                 Direction::Out.fmt_labels(f)?;
                 write!(f, ",peer=\"dst\",")?;
                 endpoint.fmt_labels(f)
             }
-            Self::InboundConnect => {
+
+            Self::InboundClient => {
                 const NO_TLS: tls::client::ConditionalClientTls =
                     Conditional::None(tls::NoClientTls::Loopback);
 
@@ -74,6 +84,49 @@ impl FmtLabels for Key {
     }
 }
 
+impl ServerLabels {
+    fn inbound(
+        tls: tls::ConditionalServerTls,
+        target_addr: SocketAddr,
+        policy: PolicyLabels,
+    ) -> Self {
+        ServerLabels {
+            direction: Direction::In,
+            tls,
+            target_addr,
+            policy: Some(policy),
+        }
+    }
+
+    fn outbound(target_addr: SocketAddr) -> Self {
+        ServerLabels {
+            direction: Direction::Out,
+            tls: tls::ConditionalServerTls::None(tls::NoServerTls::Loopback),
+            target_addr,
+            policy: None,
+        }
+    }
+}
+
+impl FmtLabels for ServerLabels {
+    fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.direction.fmt_labels(f)?;
+        f.write_str(",peer=\"src\",")?;
+        (TargetAddr(self.target_addr), TlsAccept(&self.tls)).fmt_labels(f)?;
+
+        if let Some(policy) = self.policy.as_ref() {
+            for (k, v) in policy.server.iter() {
+                write!(f, ",srv_{}=\"{}\"", k, v)?;
+            }
+            for (k, v) in policy.authz.iter() {
+                write!(f, ",saz_{}=\"{}\"", k, v)?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
 // === impl TlsAccept ===
 
 impl<'t> From<&'t tls::ConditionalServerTls> for TlsAccept<'t> {
@@ -133,3 +186,38 @@ impl FmtLabels for TargetAddr {
         write!(f, "target_addr=\"{}\"", self.0)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    pub use super::*;
+
+    impl std::fmt::Display for ServerLabels {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            self.fmt_labels(f)
+        }
+    }
+
+    #[test]
+    fn server_labels() {
+        let labels = ServerLabels::inbound(
+            tls::ConditionalServerTls::Some(tls::ServerTls::Established {
+                client_id: Some("foo.id.example.com".parse().unwrap()),
+                negotiated_protocol: None,
+            }),
+            ([192, 0, 2, 4], 40000).into(),
+            PolicyLabels {
+                server: vec![("name".to_string(), "testserver".to_string())]
+                    .into_iter()
+                    .collect(),
+                authz: vec![("name".to_string(), "testauthz".to_string())]
+                    .into_iter()
+                    .collect(),
+            },
+        );
+        assert_eq!(
+            labels.to_string(),
+            "direction=\"inbound\",peer=\"src\",target_addr=\"192.0.2.4:40000\",tls=\"true\",\
+            client_id=\"foo.id.example.com\",srv_name=\"testserver\",saz_name=\"testauthz\""
+        );
+    }
+}

linkerd/app/core/src/transport/mod.rs

@@ -2,12 +2,25 @@ pub use linkerd_proxy_transport::*;
 use linkerd_stack::{ExtractParam, Param};
 pub use linkerd_transport_metrics as metrics;
 use std::sync::Arc;
+use thiserror::Error;
 
 pub mod labels;
 
 #[derive(Clone, Debug)]
 pub struct Metrics(metrics::Registry<labels::Key>);
 
+#[derive(Clone, Debug, Error)]
+#[error("connection denied on unknown port {0}")]
+pub struct DeniedUnknownPort(pub u16);
+
+#[derive(Debug, Error)]
+#[error("unauthorized connection from {client_addr} with identity {tls:?} to {dst_addr}")]
+pub struct DeniedUnauthorized {
+    pub client_addr: Remote<ClientAddr>,
+    pub dst_addr: OrigDstAddr,
+    pub tls: linkerd_tls::ConditionalServerTls,
+}
+
 impl From<metrics::Registry<labels::Key>> for Metrics {
     fn from(reg: metrics::Registry<labels::Key>) -> Self {
         Self(reg)

linkerd/app/gateway/src/lib.rs

@@ -21,7 +21,7 @@ use linkerd_app_core::{
     Error, Infallible, NameAddr, NameMatch,
 };
 use linkerd_app_inbound::{
-    direct::{ClientInfo, GatewayConnection, GatewayTransportHeader},
+    direct::{ClientInfo, GatewayConnection, GatewayTransportHeader, Legacy},
     Inbound,
 };
 use linkerd_app_outbound::{self as outbound, Outbound};
@@ -252,6 +252,7 @@ where
                  target,
                  protocol,
                  client,
+                 ..
              }| match protocol {
                 Some(proto) => Ok(svc::Either::A(HttpTransportHeader {
                     target,
@@ -307,14 +308,17 @@ impl Param<tls::ClientId> for HttpTransportHeader {
 
 // === impl HttpLegacy ===
 
-impl<E: Into<Error>> TryFrom<(Result<Option<http::Version>, E>, ClientInfo)> for HttpLegacy {
+impl<E: Into<Error>> TryFrom<(Result<Option<http::Version>, E>, Legacy)> for HttpLegacy {
     type Error = Error;
 
     fn try_from(
-        (version, client): (Result<Option<http::Version>, E>, ClientInfo),
+        (version, gateway): (Result<Option<http::Version>, E>, Legacy),
     ) -> Result<Self, Self::Error> {
         match version {
-            Ok(Some(version)) => Ok(Self { version, client }),
+            Ok(Some(version)) => Ok(Self {
+                version,
+                client: gateway.client,
+            }),
             Ok(None) => Err(RefusedNoTarget(()).into()),
             Err(e) => Err(e.into()),
         }