Expand tests to account for audit access policy

Followup to #12846, branched off alpeb/policy-audit-impl

This fixes the policy controller unit and integration tests by accounting for the new Audit default policy and the new accessPolicy field in Server.

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

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-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;
+}

policy-test/tests/e2e_audit.rs

@@ -0,0 +1,123 @@
+use kube::{Client, ResourceExt};
+use linkerd_policy_controller_k8s_api as k8s;
+use linkerd_policy_test::{create, create_ready_pod, curl, web, with_temp_ns, LinkerdInject};
+
+#[tokio::test(flavor = "current_thread")]
+async fn server_audit() {
+    with_temp_ns(|client, ns| async move {
+        // Create a server with no policy that should block traffic to the associated pod
+        let srv = create(&client, web::server(&ns, None)).await;
+
+        // Create the web pod and wait for it to be ready.
+        tokio::join!(
+            create(&client, web::service(&ns)),
+            create_ready_pod(&client, web::pod(&ns))
+        );
+
+        // All requests should fail
+        let curl = curl::Runner::init(&client, &ns).await;
+        let (injected, uninjected) = tokio::join!(
+            curl.run("curl-injected", "http://web", LinkerdInject::Enabled),
+            curl.run("curl-uninjected", "http://web", LinkerdInject::Disabled),
+        );
+        let (injected_status, uninjected_status) =
+            tokio::join!(injected.exit_code(), uninjected.exit_code());
+        assert_ne!(injected_status, 0, "injected curl must fail");
+        assert_ne!(uninjected_status, 0, "uninjected curl must fail");
+
+        // Patch the server with accessPolicy audit
+        let patch = serde_json::json!({
+            "spec": {
+                "accessPolicy": "audit",
+            }
+        });
+        let patch = k8s::Patch::Merge(patch);
+        let api = k8s::Api::<k8s::policy::Server>::namespaced(client.clone(), &ns);
+        api.patch(&srv.name_unchecked(), &k8s::PatchParams::default(), &patch)
+            .await
+            .expect("failed to patch server");
+
+        // All requests should succeed
+        let (injected, uninjected) = tokio::join!(
+            curl.run("curl-audit-injected", "http://web", LinkerdInject::Enabled),
+            curl.run(
+                "curl-audit-uninjected",
+                "http://web",
+                LinkerdInject::Disabled
+            ),
+        );
+        let (injected_status, uninjected_status) =
+            tokio::join!(injected.exit_code(), uninjected.exit_code());
+        assert_eq!(injected_status, 0, "injected curl must contact web");
+        assert_eq!(uninjected_status, 0, "uninjected curl must contact web");
+    })
+    .await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn ns_audit() {
+    with_temp_ns(|client, ns| async move {
+        change_access_policy(client.clone(), &ns, "cluster-authenticated").await;
+
+        // Create the web pod and wait for it to be ready.
+        tokio::join!(
+            create(&client, web::service(&ns)),
+            create_ready_pod(&client, web::pod(&ns))
+        );
+
+        // Unmeshed requests should fail
+        let curl = curl::Runner::init(&client, &ns).await;
+        let (injected, uninjected) = tokio::join!(
+            curl.run("curl-injected", "http://web", LinkerdInject::Enabled),
+            curl.run("curl-uninjected", "http://web", LinkerdInject::Disabled),
+        );
+        let (injected_status, uninjected_status) =
+            tokio::join!(injected.exit_code(), uninjected.exit_code());
+        assert_eq!(injected_status, 0, "injected curl must contact web");
+        assert_ne!(uninjected_status, 0, "uninjected curl must fail");
+
+        change_access_policy(client.clone(), &ns, "audit").await;
+
+        // Recreate pod for it to pick the new default policy
+        let api = kube::Api::<k8s::api::core::v1::Pod>::namespaced(client.clone(), &ns);
+        kube::runtime::wait::delete::delete_and_finalize(
+            api,
+            "web",
+            &kube::api::DeleteParams::foreground(),
+        )
+        .await
+        .expect("web pod must be deleted");
+
+        create_ready_pod(&client, web::pod(&ns)).await;
+
+        // All requests should work
+        let (injected, uninjected) = tokio::join!(
+            curl.run("curl-audit-injected", "http://web", LinkerdInject::Enabled),
+            curl.run(
+                "curl-audit-uninjected",
+                "http://web",
+                LinkerdInject::Disabled
+            ),
+        );
+        let (injected_status, uninjected_status) =
+            tokio::join!(injected.exit_code(), uninjected.exit_code());
+        assert_eq!(injected_status, 0, "injected curl must contact web");
+        assert_eq!(uninjected_status, 0, "uninject curl must contact web");
+    })
+    .await;
+}
+
+async fn change_access_policy(client: Client, ns: &str, policy: &str) {
+    let api = k8s::Api::<k8s::Namespace>::all(client.clone());
+    let patch = serde_json::json!({
+        "metadata": {
+            "annotations": {
+                "config.linkerd.io/default-inbound-policy": policy,
+            }
+        }
+    });
+    let patch = k8s::Patch::Merge(patch);
+    api.patch(ns, &k8s::PatchParams::default(), &patch)
+        .await
+        .expect("failed to patch namespace");
+}

policy-test/tests/e2e_authorization_policy.rs

@@ -16,7 +16,7 @@ async fn meshtls() {
         //
         // The policy requires that all connections are authenticated with MeshTLS.
         let (srv, all_mtls) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, all_authenticated(&ns))
         );
         create(
@@ -60,7 +60,7 @@ async fn targets_route() {
         //
         // The policy requires that all connections are authenticated with MeshTLS.
         let (srv, all_mtls) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, all_authenticated(&ns)),
         );
         // Create a route which matches the /allowed path.
@@ -171,7 +171,7 @@ async fn targets_namespace() {
         //
         // The policy requires that all connections are authenticated with MeshTLS.
         let (_srv, all_mtls) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, all_authenticated(&ns))
         );
         create(
@@ -220,7 +220,7 @@ async fn meshtls_namespace() {
         // The policy requires that all connections are authenticated with MeshTLS
         // and come from service accounts in the given namespace.
         let (srv, mtls_ns) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, ns_authenticated(&ns))
         );
         create(
@@ -277,7 +277,7 @@ async fn network() {
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
         let (srv, allow_ips) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, allow_ips(&ns, Some(blessed_ip)))
         );
         create(
@@ -351,7 +351,7 @@ async fn both() {
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
         let (srv, allow_ips, all_mtls) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(
                 &client,
                 allow_ips(&ns, vec![blessed_injected_ip, blessed_uninjected_ip]),
@@ -451,7 +451,7 @@ async fn either() {
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
         let (srv, allow_ips, all_mtls) = tokio::join!(
-            create(&client, web::server(&ns)),
+            create(&client, web::server(&ns, None)),
             create(&client, allow_ips(&ns, vec![blessed_uninjected_ip])),
             create(&client, all_authenticated(&ns))
         );
@@ -528,7 +528,7 @@ async fn either() {
 async fn empty_authentications() {
     with_temp_ns(|client, ns| async move {
         // Create a policy that does not require any authentications.
-        let srv = create(&client, web::server(&ns)).await;
+        let srv = create(&client, web::server(&ns, None)).await;
         create(
             &client,
             authz_policy(&ns, "web", LocalTargetRef::from_resource(&srv), None),

policy-test/tests/e2e_server_authorization.rs

@@ -9,7 +9,7 @@ use linkerd_policy_test::{
 #[tokio::test(flavor = "current_thread")]
 async fn meshtls() {
     with_temp_ns(|client, ns| async move {
-        let srv = create(&client, web::server(&ns)).await;
+        let srv = create(&client, web::server(&ns, None)).await;
 
         create(
             &client,
@@ -66,7 +66,7 @@ async fn network() {
 
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
-        let srv = create(&client, web::server(&ns)).await;
+        let srv = create(&client, web::server(&ns, None)).await;
         create(
             &client,
             server_authz(
@@ -143,7 +143,7 @@ async fn both() {
 
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
-        let srv = create(&client, web::server(&ns)).await;
+        let srv = create(&client, web::server(&ns, None)).await;
         create(
             &client,
             server_authz(
@@ -243,7 +243,7 @@ async fn either() {
 
         // Once we know the IP of the (blocked) pod, create an web
         // authorization policy that permits connections from this pod.
-        let srv = create(&client, web::server(&ns)).await;
+        let srv = create(&client, web::server(&ns, None)).await;
         tokio::join!(
             create(
                 &client,