Audit access policy implementation (#12846)

Followup to #12845

This expands the policy controller index in the following ways:

- Adds the new Audit variant to the DefaultPolicy enum
- Expands the function that synthesizes the authorizations for a given default policy (DefaultPolicy::default_authzs) so that it also creates an Unauthenticated client auth and a allow-all NetworkMatch for the new Audit default policy.
- Now that a Server can have a default policy different than Deny, when generating InboundServer authorizations (PolicyIndex::client_authzs) make sure to append the default authorizations when DefaultPolicy is Allow or Audit

Also, the admission controller ensures the new accessPolicy field contains a valid value.

## Tests

New integration tests added:

- e2e_audit.rs exercising first the audit policy in Server, and then at the namespace level
- in admit_server.rs a new test checks invalid accessPolicy values are rejected.
- in inbound_api.rs server_with_audit_policy verifies the synthesized audit authorization is returned for a Server with accessPolicy=audit

> [!NOTE]
> Please check linkerd/website#1805 for how this is supposed to work from the user's perspective.

policy-controller/k8s/api/src/policy/server.rs

@@ -18,6 +18,7 @@ pub struct ServerSpec {
     pub selector: Selector,
     pub port: Port,
     pub proxy_protocol: Option<ProxyProtocol>,
+    pub access_policy: Option<String>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]

policy-controller/k8s/index/src/defaults.rs

@@ -21,6 +21,9 @@ pub enum DefaultPolicy {
 
     /// Indicates that all traffic is denied unless explicitly permitted by an authorization policy.
     Deny,
+
+    /// Indicates that all traffic is let through, but gets audited
+    Audit,
 }
 
 // === impl DefaultPolicy ===
@@ -47,6 +50,7 @@ impl std::str::FromStr for DefaultPolicy {
                 cluster_only: true,
             }),
             "deny" => Ok(Self::Deny),
+            "audit" => Ok(Self::Audit),
             s => Err(anyhow!("invalid mode: {:?}", s)),
         }
     }
@@ -72,6 +76,7 @@ impl DefaultPolicy {
                 cluster_only: true,
             } => "cluster-unauthenticated",
             Self::Deny => "deny",
+            Self::Audit => "audit",
         }
     }
 
@@ -80,34 +85,48 @@ impl DefaultPolicy {
         config: &ClusterInfo,
     ) -> HashMap<AuthorizationRef, ClientAuthorization> {
         let mut authzs = HashMap::default();
+        let auth_ref = AuthorizationRef::Default(self.as_str());
+
         if let DefaultPolicy::Allow {
             authenticated_only,
             cluster_only,
         } = self
         {
-            let authentication = if authenticated_only {
-                ClientAuthentication::TlsAuthenticated(vec![IdentityMatch::Suffix(vec![])])
-            } else {
-                ClientAuthentication::Unauthenticated
-            };
-            let networks = if cluster_only {
-                config.networks.iter().copied().map(Into::into).collect()
-            } else {
-                vec![
-                    "0.0.0.0/0".parse::<IpNet>().unwrap().into(),
-                    "::/0".parse::<IpNet>().unwrap().into(),
-                ]
-            };
             authzs.insert(
-                AuthorizationRef::Default(self.as_str()),
-                ClientAuthorization {
-                    authentication,
-                    networks,
-                },
+                auth_ref,
+                Self::default_client_authz(config, authenticated_only, cluster_only),
             );
-        };
+        } else if let DefaultPolicy::Audit = self {
+            authzs.insert(auth_ref, Self::default_client_authz(config, false, false));
+        }
+
         authzs
     }
+
+    fn default_client_authz(
+        config: &ClusterInfo,
+        authenticated_only: bool,
+        cluster_only: bool,
+    ) -> ClientAuthorization {
+        let authentication = if authenticated_only {
+            ClientAuthentication::TlsAuthenticated(vec![IdentityMatch::Suffix(vec![])])
+        } else {
+            ClientAuthentication::Unauthenticated
+        };
+        let networks = if cluster_only {
+            config.networks.iter().copied().map(Into::into).collect()
+        } else {
+            vec![
+                "0.0.0.0/0".parse::<IpNet>().unwrap().into(),
+                "::/0".parse::<IpNet>().unwrap().into(),
+            ]
+        };
+
+        ClientAuthorization {
+            authentication,
+            networks,
+        }
+    }
 }
 
 impl std::fmt::Display for DefaultPolicy {
@@ -140,6 +159,7 @@ mod test {
                 authenticated_only: false,
                 cluster_only: true,
             },
+            DefaultPolicy::Audit,
         ] {
             assert_eq!(
                 default.to_string().parse::<DefaultPolicy>().unwrap(),

policy-controller/k8s/index/src/inbound/index.rs

@@ -1753,6 +1753,10 @@ impl PolicyIndex {
             authzs.insert(reference, authz);
         }
 
+        if let Some(p) = server.access_policy {
+            authzs.extend(p.default_authzs(&self.cluster_info));
+        }
+
         authzs
     }
 

policy-controller/k8s/index/src/inbound/server.rs

@@ -1,4 +1,4 @@
-use crate::ClusterInfo;
+use crate::{ClusterInfo, DefaultPolicy};
 use linkerd_policy_controller_core::inbound::ProxyProtocol;
 use linkerd_policy_controller_k8s_api::{
     self as k8s, policy::server::Port, policy::server::Selector,
@@ -11,6 +11,7 @@ pub(crate) struct Server {
     pub selector: Selector,
     pub port_ref: Port,
     pub protocol: ProxyProtocol,
+    pub access_policy: Option<DefaultPolicy>,
 }
 
 impl Server {
@@ -20,6 +21,7 @@ impl Server {
             selector: srv.spec.selector,
             port_ref: srv.spec.port,
             protocol: proxy_protocol(srv.spec.proxy_protocol, cluster),
+            access_policy: srv.spec.access_policy.and_then(|p| p.parse().ok()),
         }
     }
 }

policy-controller/k8s/index/src/inbound/tests.rs

@@ -45,7 +45,7 @@ struct TestConfig {
     _tracing: tracing::subscriber::DefaultGuard,
 }
 
-const DEFAULTS: [DefaultPolicy; 5] = [
+const DEFAULTS: [DefaultPolicy; 6] = [
     DefaultPolicy::Deny,
     DefaultPolicy::Allow {
         authenticated_only: true,
@@ -63,6 +63,7 @@ const DEFAULTS: [DefaultPolicy; 5] = [
         authenticated_only: false,
         cluster_only: true,
     },
+    DefaultPolicy::Audit,
 ];
 
 pub fn mk_pod_with_containers(
@@ -121,6 +122,7 @@ fn mk_server(
             port,
             selector: k8s::policy::server::Selector::Pod(pod_labels.into_iter().collect()),
             proxy_protocol,
+            access_policy: None,
         },
     }
 }
@@ -177,6 +179,13 @@ fn mk_default_policy(
                 networks: cluster_nets,
             },
         )),
+        DefaultPolicy::Audit => Some((
+            AuthorizationRef::Default("audit"),
+            ClientAuthorization {
+                authentication: ClientAuthentication::Unauthenticated,
+                networks: all_nets,
+            },
+        )),
     }
     .into_iter()
     .collect()

policy-controller/k8s/index/src/inbound/tests/annotation.rs

@@ -109,6 +109,7 @@ fn authenticated_annotated() {
                     authenticated_only: true,
                 },
                 DefaultPolicy::Deny => DefaultPolicy::Deny,
+                DefaultPolicy::Audit => DefaultPolicy::Audit,
             };
             InboundServer {
                 reference: ServerRef::Default(policy.as_str()),

policy-controller/k8s/status/src/tests/routes.rs

@@ -47,6 +47,7 @@ fn make_server(
             port,
             selector: linkerd_k8s_api::server::Selector::Pod(pod_labels.into_iter().collect()),
             proxy_protocol,
+            access_policy: None,
         },
     }
 }

policy-controller/src/admission.rs

@@ -336,7 +336,8 @@ impl Validate<MeshTLSAuthenticationSpec> for Admission {
 
 #[async_trait::async_trait]
 impl Validate<ServerSpec> for Admission {
-    /// Checks that `spec` doesn't select the same pod/ports as other existing Servers
+    /// Checks that `spec` doesn't select the same pod/ports as other existing Servers, and that
+    /// `accessPolicy` contains a valid value
     //
     // TODO(ver) this isn't rigorous about detecting servers that select the same port if one port
     // specifies a numeric port and the other specifies the port's name.
@@ -369,6 +370,12 @@ impl Validate<ServerSpec> for Admission {
             }
         }
 
+        if let Some(policy) = spec.access_policy {
+            policy
+                .parse::<index::DefaultPolicy>()
+                .map_err(|err| anyhow!("Invalid 'accessPolicy' field: {err}"))?;
+        }
+
         Ok(())
     }
 }

policy-test/src/web.rs

@@ -33,7 +33,7 @@ pub fn pod(ns: &str) -> k8s::Pod {
     }
 }
 
-pub fn server(ns: &str) -> k8s::policy::Server {
+pub fn server(ns: &str, access_policy: Option<String>) -> k8s::policy::Server {
     k8s::policy::Server {
         metadata: k8s::ObjectMeta {
             namespace: Some(ns.to_string()),
@@ -46,6 +46,7 @@ pub fn server(ns: &str) -> k8s::policy::Server {
             )))),
             port: k8s::policy::server::Port::Name("http".to_string()),
             proxy_protocol: Some(k8s::policy::server::ProxyProtocol::Http1),
+            access_policy,
         },
     }
 }

policy-test/tests/admit_server.rs

@@ -16,6 +16,7 @@ async fn accepts_valid() {
             selector: Selector::Pod(api::labels::Selector::default()),
             port: Port::Number(80.try_into().unwrap()),
             proxy_protocol: None,
+            access_policy: None,
         },
     })
     .await;
@@ -34,6 +35,7 @@ async fn accepts_server_updates() {
                 selector: Selector::Pod(api::labels::Selector::from_iter(Some(("app", "test")))),
                 port: Port::Number(80.try_into().unwrap()),
                 proxy_protocol: None,
+                access_policy: None,
             },
         };
 
@@ -60,6 +62,7 @@ async fn rejects_identitical_pod_selector() {
             selector: Selector::Pod(api::labels::Selector::from_iter(Some(("app", "test")))),
             port: Port::Number(80.try_into().unwrap()),
             proxy_protocol: None,
+            access_policy: None,
         };
 
         let api = kube::Api::namespaced(client, &ns);
@@ -106,6 +109,7 @@ async fn rejects_all_pods_selected() {
                 selector: Selector::Pod(api::labels::Selector::from_iter(Some(("app", "test")))),
                 port: Port::Number(80.try_into().unwrap()),
                 proxy_protocol: Some(ProxyProtocol::Http2),
+                access_policy: None,
             },
         };
         api.create(&kube::api::PostParams::default(), &test0)
@@ -123,6 +127,7 @@ async fn rejects_all_pods_selected() {
                 port: Port::Number(80.try_into().unwrap()),
                 // proxy protocol doesn't factor into the selection
                 proxy_protocol: Some(ProxyProtocol::Http1),
+                access_policy: None,
             },
         };
         api.create(&kube::api::PostParams::default(), &test1)
@@ -179,3 +184,21 @@ async fn rejects_invalid_proxy_protocol() {
     })
     .await;
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn rejects_invalid_access_policy() {
+    admission::rejects(|ns| Server {
+        metadata: api::ObjectMeta {
+            namespace: Some(ns),
+            name: Some("test".to_string()),
+            ..Default::default()
+        },
+        spec: ServerSpec {
+            selector: Selector::Pod(api::labels::Selector::default()),
+            port: Port::Number(80.try_into().unwrap()),
+            proxy_protocol: None,
+            access_policy: Some("foobar".to_string()),
+        },
+    })
+    .await;
+}