Audit access policy implementation

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.

Required test changes are addressed in #12847. Also note you'll need the proxy changes at linkerd/linkerd2-proxy#3068 to make this work.

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/src/admission.rs

@@ -319,7 +319,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.
@@ -346,6 +347,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(())
     }
 }