WIP: E2E test case for Kmesh L4 authorization

bpf/kmesh/workload/xdp.c

@@ -106,9 +106,11 @@ SEC("xdp_auth")
 int xdp_authz(struct xdp_md *ctx)
 {
     if (!is_authz_offload_enabled()) {
+        BPF_LOG(INFO, XDP, "offload authorization is DISABLED");
         bpf_tail_call(ctx, &map_of_xdp_tailcall, TAIL_CALL_AUTH_IN_USER_SPACE);
         return XDP_PASS;
     }
+    BPF_LOG(INFO, XDP, "offload authorization is ENABLED");
 
     struct match_context match_ctx = {0};
     struct bpf_sock_tuple tuple_key = {0};

pkg/status/status_server.go

@@ -318,11 +318,31 @@
 }
 
 func (s *Server) authzHandler(w http.ResponseWriter, r *http.Request) {
-	if r.Method != http.MethodPost {
+	if r.Method != http.MethodGet && r.Method != http.MethodPost {
 		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
 		return
 	}
 
+	configMap, err := bpf.GetKmeshConfigMap(s.kmeshConfigMap)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("update authz in KmeshConfigMap failed: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	if r.Method == http.MethodGet {
+		var result string
+		switch configMap.AuthzOffload {
+		case constants.ENABLED:
+			result = "enabled"
+		case constants.DISABLED:
+			result = "disabled"
+		}
+
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(result))
+		return
+	}
+
 	authzInfo := r.URL.Query().Get("enable")
 	enabled, err := strconv.ParseBool(authzInfo)
 	if err != nil {
@@ -331,11 +351,6 @@
 		return
 	}
 
-	configMap, err := bpf.GetKmeshConfigMap(s.kmeshConfigMap)
-	if err != nil {
-		http.Error(w, fmt.Sprintf("update authz in KmeshConfigMap failed: %v", err), http.StatusBadRequest)
-		return
-	}
 	if enabled {
 		configMap.AuthzOffload = constants.ENABLED
 	} else {
@@ -347,6 +362,7 @@
 	}
 
 	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("OK"))
 }
 
 func (s *Server) getLoggerNames(w http.ResponseWriter) {

test/e2e/baseline_test.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/netip"
+	"os/exec"
 	"sort"
 	"strings"
 	"testing"
@@ -689,6 +690,141 @@ func TestBookinfo(t *testing.T) {
 	})
 }
 
+func TestAuthorizationL4(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("L4 Authorization").Run(func(t framework.TestContext) {
+			// Enable authorizaiton offload to xdp.
+
+			if len(apps.ServiceWithWaypointAtServiceGranularity) == 0 {
+				t.Fatal(fmt.Errorf("need at least 1 instance of apps.ServiceWithWaypointAtServiceGranularity"))
+			}
+			src := apps.ServiceWithWaypointAtServiceGranularity[0]
+
+			clients := src.WorkloadsOrFail(t)
+			dst := apps.EnrolledToKmesh
+
+			addresses := clients.Addresses()
+			if len(addresses) < 2 {
+				t.Fatal(fmt.Errorf("need at least 2 clients"))
+			}
+			selectedAddress := addresses[0]
+
+			authzCases := []struct {
+				name string
+				spec string
+			}{
+				{
+					name: "allow",
+					spec: `
+  action: ALLOW
+`,
+				},
+				{
+					name: "deny",
+					spec: `
+  action: DENY
+`,
+				},
+			}
+
+			chooseChecker := func(action string, ip string) echo.Checker {
+				switch action {
+				case "allow":
+					if ip != selectedAddress {
+						return check.NotOK()
+					} else {
+						return check.OK()
+					}
+				case "deny":
+					if ip != selectedAddress {
+						return check.OK()
+					} else {
+						return check.NotOK()
+					}
+				default:
+					t.Fatal("invalid action")
+				}
+
+				return check.OK()
+			}
+
+			count := 0
+			workloads := dst.WorkloadsOrFail(t)
+			for _, client := range workloads {
+				if count == len(workloads) {
+					break
+				}
+				podName := client.PodName()
+				namespace := apps.Namespace.Name()
+				timeout := time.After(5 * time.Second)
+				ticker := time.NewTicker(500 * time.Millisecond)
+				defer ticker.Stop()
+			InnerLoop:
+				for {
+					select {
+					case <-timeout:
+						t.Fatalf("Timeout: XDP eBPF program not found on pod %s", podName)
+					case <-ticker.C:
+						cmd := exec.Command("kubectl", "exec", "-n", namespace, podName, "--", "sh", "-c", "ip a | grep xdp")
+						output, err := cmd.CombinedOutput()
+						if err == nil && len(output) > 0 {
+							t.Logf("XDP program is loaded on pod %s", podName)
+							count++
+							break InnerLoop
+						}
+						t.Logf("Waiting for XDP program to load on pod %s: %v", podName, err)
+					}
+				}
+			}
+
+			for _, tc := range authzCases {
+				t.ConfigIstio().Eval(apps.Namespace.Name(), map[string]string{
+					"Destination": dst.Config().Service,
+					"Ip":          selectedAddress,
+				}, `apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: policy
+spec:
+  selector:
+    matchLabels:
+      app: "{{.Destination}}"
+`+tc.spec+`
+  rules:
+  - from:
+    - source:
+        ipBlocks:
+        - "{{.Ip}}"
+`).ApplyOrFail(t)
+
+				for _, client := range clients {
+					opt := echo.CallOptions{
+						To:                      dst,
+						Port:                    echo.Port{Name: "tcp"},
+						Scheme:                  scheme.TCP,
+						NewConnectionPerRequest: true,
+						// Due to the mechanism of Kmesh L4 authorization, we need to set the timeout slightly longer.
+						Timeout: time.Minute * 2,
+					}
+
+					var name string
+					if client.Address() != selectedAddress {
+						name = tc.name + ", not selected address"
+					} else {
+						name = tc.name + ", selected address"
+					}
+
+					opt.Check = chooseChecker(tc.name, client.Address())
+
+					t.NewSubTestf("%v", name).Run(func(t framework.TestContext) {
+						src.WithWorkloads(client).CallOrFail(t, opt)
+					})
+				}
+			}
+		})
+	})
+}
+
 func runTest(t *testing.T, f func(t framework.TestContext, src echo.Instance, dst echo.Instance, opt echo.CallOptions)) {
 	framework.NewTest(t).Run(func(t framework.TestContext) {
 		runTestContext(t, f)

test/e2e/main_test.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"runtime"
 	"testing"
@@ -65,6 +66,8 @@ var (
 
 	apps = &EchoDeployments{}
 
+	kmeshctl = NewKmeshctl()
+
 	// used to validate telemetry in-cluster
 	prom prometheus.Instance
 )
@@ -177,10 +180,18 @@ func SetupApps(t resource.Context, i istio.Instance, apps *EchoDeployments) erro
 				{
 					Replicas: 1,
 					Version:  "v1",
+					Labels: map[string]string{
+						"app":     EnrolledToKmesh,
+						"version": "v1",
+					},
 				},
 				{
 					Replicas: 1,
 					Version:  "v2",
+					Labels: map[string]string{
+						"app":     EnrolledToKmesh,
+						"version": "v2",
+					},
 				},
 			},
 		})
@@ -372,3 +383,27 @@ func deleteWaypointProxy(ctx resource.Context, ns namespace.Instance, name strin
 		return nil
 	}, retry.Timeout(time.Minute*10), retry.BackoffDelay(time.Millisecond*200))
 }
+
+type kmeshctlWrapper struct {
+}
+
+func NewKmeshctl() *kmeshctlWrapper {
+	return &kmeshctlWrapper{}
+}
+
+// Invoke will invokes an kmeshctl command and returns the output and exception.
+func (k *kmeshctlWrapper) Authz(subcmd string) (string, error) {
+	cmd := exec.Command("kmeshctl", "authz", subcmd)
+	output, err := cmd.Output()
+	if err != nil {
+		return "", err
+	}
+
+	return string(output), nil
+}
+
+func (k *kmeshctlWrapper) AuthzOrFatal(t test.Failer, subcmd string) {
+	if _, err := k.Authz(subcmd); err != nil {
+		t.Fatal("failed to set authz to %d using kmeshctl: %v", subcmd, err)
+	}
+}

test/e2e/run_test.sh

@@ -137,6 +137,16 @@ function setup_kmesh() {
         echo "Waiting for pods of Kmesh daemon to enter Running state..."
         sleep 1
     done
+
+    # Set log of each Kmesh pods.
+    PODS=$(kubectl get pods -n kmesh-system -l app=kmesh -o jsonpath='{.items[*].metadata.name}')
+
+    sleep 10
+
+    for POD in $PODS; do
+        echo $POD
+        kmeshctl log $POD --set bpf:debug
+    done
 }
 
 export KIND_REGISTRY_NAME="kind-registry"
@@ -161,6 +171,11 @@ function build_and_push_images() {
     HUB="${KIND_REGISTRY}" TAG="latest" make docker.push
 }
 
+function install_kmeshctl() {
+    # Install kmeshctl
+    cp kmeshctl $TMPBIN
+}
+
 function install_dependencies() {
     # 1. Install kind.
     if ! which kind &> /dev/null
@@ -274,6 +289,7 @@ fi
 if [[ -z "${SKIP_BUILD:-}" ]]; then
     setup_kind_registry
     build_and_push_images
+    install_kmeshctl
 fi
 
 kubectl config use-context "kind-$NAME"
@@ -286,7 +302,7 @@ if [[ -z "${SKIP_SETUP:-}" ]]; then
     setup_kmesh
 fi
 
-cmd="go test -v -tags=integ $ROOT_DIR/test/e2e/... -istio.test.kube.loadbalancer=false ${PARAMS[*]}"
+cmd="go test -v -tags=integ $ROOT_DIR/test/e2e/... -istio.test.kube.loadbalancer=false -istio.test.echo.callTimeout 300s ${PARAMS[*]}"
 
 bash -c "$cmd"