Repository: kmesh-net/kmesh

Issue:
[OSPP 2025] E2E of IPsce and offload Authorization
In version 1.0, Kmesh introduced the IPsec feature and dropped some of the authorization capabilities into the OS kernel. However, there was a lack of testing to look after this part of the functionality. This led to hidden risks when the code went live and new versions were released. Therefore, there is a need to write e2e tests for these two features and to document their use.
1、IPsec
2、Offload Authorization
IPsec is a feature that ensures secure communication between nodes, requiring cross-node testing with a self-designed testing plan. Offload Authorization is unique to Kmesh, offloading IP and Port authentication capabilities to the Linux kernel's XDP. Therefore, it is necessary to test both the user-space programs for Authorization and the XDP programs in the kernel. And different tests are performed for complementary authentication scenarios

Discussion:
@LiZhenCheng9527: ## OUTPUT:

1. code of IPSec e2e Test
2. E2E Test Code of Offload Authorization 
2.1. Testing for IP authentication
2.2. Testing for Port authentication
2.3. Testing for Http Header authentication
2.4. Additional: Tests for namespace, hosts authentication
3.  Test and Implementation Documentation for Ipsec and Offload Authorization@LiZhenCheng9527: #1401 @xiaojiangao123: I want to work on this project. Please assign it to me.@hzxuzhonghu: /assign @xiaojiangao123 @xiaojiangao123: #1489 

Solution patch:
diff --git a/deploy/charts/kmesh-helm/templates/kmesh-rbac.yaml b/deploy/charts/kmesh-helm/templates/kmesh-rbac.yaml
index dbc57537..1ebee642 100644
--- a/deploy/charts/kmesh-helm/templates/kmesh-rbac.yaml
+++ b/deploy/charts/kmesh-helm/templates/kmesh-rbac.yaml
@@ -26,6 +26,9 @@ rules:
   - daemonsets
   verbs:
   - get
+- apiGroups: ["kmesh.net"]
+  resources: ["kmeshnodeinfos"]
+  verbs: ["get", "create", "update", "delete", "list", "watch"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
diff --git a/deploy/charts/kmesh-helm/templates/kmesh.net_kmeshnodeinfos.yaml b/deploy/charts/kmesh-helm/templates/kmesh.net_kmeshnodeinfos.yaml
new file mode 100644
index 00000000..a52b07dc
--- /dev/null
+++ b/deploy/charts/kmesh-helm/templates/kmesh.net_kmeshnodeinfos.yaml
@@ -0,0 +1,78 @@
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.16.4
+  name: kmeshnodeinfos.kmesh.net
+spec:
+  group: kmesh.net
+  names:
+    kind: KmeshNodeInfo
+    listKind: KmeshNodeInfoList
+    plural: kmeshnodeinfos
+    singular: kmeshnodeinfo
+  scope: Namespaced
+  versions:
+  - name: v1alpha1
+    schema:
+      openAPIV3Schema:
+        description: KmeshNode is the Schema for the kmeshnodes API
+        properties:
+          apiVersion:
+            description: |-
+              APIVersion defines the versioned schema of this representation of an object.
+              Servers should convert recognized schemas to the latest internal value, and
+              may reject unrecognized values.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
+            type: string
+          kind:
+            description: |-
+              Kind is a string value representing the REST resource this object represents.
+              Servers may infer this from the endpoint the client submits requests to.
+              Cannot be updated.
+              In CamelCase.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
+            type: string
+          metadata:
+            type: object
+          spec:
+            properties:
+              addresses:
+                description: |-
+                  Addresses is used to store the internal ip address information on the
+                  host. The IP address information is used to generate the IPsec state
+                  information. IPsec uses this information to determine which network
+                  adapter is used to encrypt and send data.
+                items:
+                  type: string
+                type: array
+              bootID:
+                description: |-
+                  bootid is used to generate the ipsec key. After the node is restarted,
+                  the key needs to be updated.
+                type: string
+              podCIDRS:
+                description: |-
+                  PodCIDRs used in IPsec checks the destination of the data to
+                  determine which IPsec state is used for encryption.
+                items:
+                  type: string
+                type: array
+              spi:
+                description: |-
+                  The SPI is used to identify the version number of the current key.
+                  The communication can be normal only when both communication parties
+                  have spis and the spi keys are the same.
+                type: integer
+            required:
+            - addresses
+            - bootID
+            - podCIDRS
+            - spi
+            type: object
+          status:
+            type: object
+        type: object
+    served: true
+    storage: true
diff --git a/docs/proposal/ipsec_authz_e2e_test.md b/docs/proposal/ipsec_authz_e2e_test.md
new file mode 100644
index 00000000..2e27149b
--- /dev/null
+++ b/docs/proposal/ipsec_authz_e2e_test.md
@@ -0,0 +1,302 @@
+---
+title: IPSec & Offload Authorization E2E Test
+authors:
+- "@xiaojiangao123" # Authors' GitHub accounts here.
+reviewers:
+- "@robot"
+- TBD
+approvers:
+- "@robot"
+- TBD
+
+creation-date: 2025-5-12
+
+---
+
+## IPSec & Offload Authorization E2E Test Proposal
+
+### Overview
+
+This proposal aims to design end-to-end (E2E) test cases for the IPSec feature and Offload Authorization of Kmesh. The IPSec E2E tests focus on basic functionality, security, and reliability, verifying proper execution of IPSec tunnel setup, encryption/decryption, key management, and fault recovery. The Offload Authorization E2E tests include IP authorization, Port authorization, Header authorization, namespace authorization, and hosts authorization.
+
+### Motivation
+
+Kmesh’s IPSec functionality is a key component for secure communication within the service mesh. Its stability and reliability directly affect the security of the entire mesh. The Offload Authorization feature is a unique Kmesh capability that offloads IP and Port authorization to the Linux kernel via XDP. Without comprehensive E2E testing, version upgrades and releases may pose potential risks.
+
+#### Goals
+
+1. Develop complete E2E test cases  
+2. Cover all functional scenarios of Kmesh IPSec  
+3. Cover all types of Offload Authorization mechanisms
+
+### Proposal
+
+For the IPSec feature, we designed IPSec E2E test cases covering three core scenarios: basic connectivity, key rotation, and fault recovery.
+
+1. The basic connectivity test verifies IPSec tunnel establishment and the correctness of encrypted communication using tcpdump to inspect ESP packets.
+
+2. The key rotation test ensures the reliability of the PSK update mechanism and validates service continuity during key changes.
+
+For the Offload Authorization feature, E2E test cases are designed for IP, Port, Header, Namespace, and Hosts authorization.
+
+The test environment requires at least a 2-nodes Kubernetes cluster using httpbin and sleep or fortio as test applications.
+
+### Design Details
+
+#### 1. Test Environment Preparation
+
+##### Requirements
+
+- At least a 2-nodes Kubernetes cluster with Kmesh installed  
+- Tools: tcpdump  
+- Applications: httpbin, sleep, fortio
+
+#### 2. IPSec Test Scenarios
+
+##### 2.1 Basic Connectivity E2E Test
+
+###### Test Steps
+
+- Deploy httpbin and sleep applications on different nodes  
+- Verify connectivity between applications  
+- Check IPSec state and policy rules:
+
+   ```bash
+   ip xfrm state show
+   ```
+
+   **Expected Output:**
+
+   ```plaintext
+   src {{SRC_IP}} dst {{DST_IP}}
+       proto esp spi 0x{{SPI}} reqid 1 mode tunnel
+       replay-window 0 
+       output-mark 0xd0/0xffffffff
+       aead rfc4106(gcm(aes)) {{KEY}} 128
+       anti-replay context: seq 0x0, oseq 0x0, bitmap 0x00000000
+       sel src ::/0 dst ::/0 
+   ```
+
+   ```bash
+   ip xfrm policy show
+   ```
+
+   **Expected Output:**
+
+   ```plaintext
+   src ::/0 dst {{DST_SUBNET}} 
+       dir out priority 0 
+       mark 0xe0/0xffffffff 
+       tmpl src {{SRC_IP}} dst {{DST_IP}}
+           proto esp spi 0x{{SPI}} reqid 1 mode tunnel
+   ```
+
+- Verify encryption using tcpdump; analyze ESP headers and confirm payload is encrypted:
+
+   ```bash
+   tcpdump -i any esp
+   ```
+
+   **Expected Output:** ESP packets should be visible during communication.
+
+##### 2.2 Key Rotation E2E Test
+
+###### Test Steps
+
+- Deploy httpbin and sleep applications on different nodes
+- Record initial SPI and pre-shared key:
+
+   ```bash
+   ip xfrm state show
+   ```
+
+   **Expected Output:**
+
+   ```plaintext
+   src {{SRC_IP}} dst {{DST_IP}}
+       proto esp spi 0x{{INITIAL_SPI}} reqid 1 mode tunnel
+       aead rfc4106(gcm(aes)) {{INITIAL_KEY}} 128
+   ```
+
+- Send continuous traffic between applications
+- Update the pre-shared key
+
+   ```bash
+   kubectl create secret
+   ```
+
+- Check if SPI and key are updated in xfrm rules:
+
+   ```bash
+   ip xfrm state show
+   ```
+
+   **Expected Output:**
+
+   ```plaintext
+   src {{SRC_IP}} dst {{DST_IP}}
+       proto esp spi 0x{{NEW_SPI}} reqid 1 mode tunnel
+       aead rfc4106(gcm(aes)) {{NEW_KEY}} 128
+   ```
+
+- Verify communication continuity, encryption status
+
+#### 3. Offload Authorization Test Scenarios
+
+Apply the corresponding security policies (ALLOW/DENY) and routing strategies, and verify connectivity, including traffic that matches the rules and will be allowed and traffic that will be denied by the rules.
+
+Here are some AuthorizationPolicy samples.
+
+##### 3.1  IP Authorization Test
+
+- Example security policy YAML:
+
+```yaml
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: ip-allow-policy
+  namespace: test-ns1
+spec:
+  action: ALLOW
+  rules:
+  - from:
+    - source:
+        ipBlocks:
+        - "{{.Ip}}"
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: ip-deny-policy
+  namespace: test-ns1
+spec:
+  action: DENY
+  rules:
+  - from:
+    - source:
+        ipBlocks:
+        - "{{.Ip}}"
+```
+
+##### 3.2 Port Authorization Test
+
+- Example security policy YAML:
+
+```yaml
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: port-allow-policy
+  namespace: test-ns1
+spec:
+  action: ALLOW
+  rules:
+  - to:
+    - operation:
+        ports: {{.Ports}}
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: port-deny-policy
+  namespace: test-ns1
+spec:
+  action: DENY
+  rules:
+  - to:
+    - operation:
+        ports: {{.Ports}}
+```
+
+##### 3.3 Namespace Authorization Test
+
+- Example security policy YAML:
+
+```yaml
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: namespace-allow-policy
+  namespace: test-ns1
+spec:
+  action: ALLOW
+  rules:
+  - from:
+    - source:
+        namespaces:
+        - "{{.SourceNamespace}}"
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: namespace-deny-policy
+  namespace: test-ns1
+spec:
+  action: DENY
+  rules:
+  - from:
+    - source:
+        namespaces:
+        - "{{.SourceNamespace}}"
+```
+
+##### 3.4 Header Authorization Test
+
+- Example security policy YAML:
+
+```yaml
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: header-allow-policy
+  namespace: test-ns1
+spec:
+  action: ALLOW
+  rules:
+  - when:
+    - key: request.headers[{{.HeaderName}}]
+      values: ["{{.HeaderValue}}"]
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: header-deny-policy
+  namespace: test-ns1
+spec:
+  action: DENY
+  rules:
+  - when:
+    - key: request.headers[{{.HeaderName}}]
+      values: ["{{.HeaderValue}}"]
+```
+
+##### 3.5 Hosts Authorization Test
+
+- Example security policy YAML:
+
+```yaml
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: host-allow-policy
+  namespace: test-ns1
+spec:
+  action: ALLOW
+  rules:
+  - to:
+    - operation:
+        hosts: ["{{.TargetHost}}"]
+---
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: host-deny-policy
+  namespace: test-ns1
+spec:
+  action: DENY
+  rules:
+  - to:
+    - operation:
+        hosts: ["{{.TargetHost}}"]
+```



Test patch:
diff --git a/test/e2e/authorization_test.go b/test/e2e/authorization_test.go
new file mode 100644
index 00000000..e03eb640
--- /dev/null
+++ b/test/e2e/authorization_test.go
@@ -0,0 +1,647 @@
+//go:build integ
+// +build integ
+
+/*
+ * Copyright The Kmesh Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// NOTE: THE CODE IN THIS FILE IS MAINLY REFERENCED FROM ISTIO INTEGRATION
+// FRAMEWORK(https://github.com/istio/istio/tree/master/tests/integration)
+// AND ADAPTED FOR KMESH.
+
+package kmesh
+
+import (
+	"fmt"
+	"istio.io/istio/pkg/test/echo/common/scheme"
+	"istio.io/istio/pkg/test/framework"
+	"os/exec"
+	"testing"
+	"time"
+
+	"istio.io/istio/pkg/test/framework/components/echo"
+	"istio.io/istio/pkg/test/framework/components/echo/check"
+)
+
+// Wait for all the XDP programs of dst to load
+func waitForXDPOnDstWorkloads(t framework.TestContext, dst echo.Instances) {
+	count := 0
+	workloads := dst.WorkloadsOrFail(t)
+	namespace := apps.Namespace.Name()
+	for _, client := range workloads {
+		if count == len(workloads) {
+			break
+		}
+		podName := client.PodName()
+		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)
+			}
+		}
+	}
+}
+
+func TestIPAuthorization(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("IP Authorization").Run(func(t framework.TestContext) {
+
+			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()
+			}
+
+			waitForXDPOnDstWorkloads(t, dst)
+
+			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: ip-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 TestPortAuthorization(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("Port Authorization").Run(func(t framework.TestContext) {
+
+			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)
+			client := clients[0]
+			dst := apps.EnrolledToKmesh
+
+			// Define the test port
+			selectedPodPort := 19090
+			selectedServicePort := 9090
+			notSelectedPodPort := 16060
+			notSelectedServicePort := 9091
+
+			// Echo Pod healthy port
+			readyPort := 8080
+			livenessPort := 3333
+
+			authzCases := []struct {
+				name  string
+				spec  string
+				ports string
+			}{
+				{
+					name:  "allow",
+					spec:  `  action: ALLOW`,
+					ports: fmt.Sprintf(`["%d", "%d", "%d"]`, selectedPodPort, readyPort, livenessPort),
+				},
+				{
+					name:  "deny",
+					spec:  `  action: DENY`,
+					ports: fmt.Sprintf(`["%d"]`, selectedPodPort),
+				},
+			}
+
+			chooseChecker := func(action string, servicePort int) echo.Checker {
+				switch action {
+				case "allow":
+					if servicePort == selectedServicePort {
+						return check.OK()
+					} else {
+						return check.NotOK()
+					}
+				case "deny":
+					if servicePort == selectedServicePort {
+						return check.NotOK()
+					} else {
+						return check.OK()
+					}
+				default:
+					t.Fatal("invalid action")
+				}
+
+				return check.OK()
+			}
+
+			portTestCases := []struct {
+				servicePort int
+				podPort     int
+				description string
+			}{
+				{
+					servicePort: selectedServicePort,
+					podPort:     selectedPodPort,
+					description: fmt.Sprintf("service port %d (pod port %d)", selectedServicePort, selectedPodPort),
+				},
+				{
+					servicePort: notSelectedServicePort,
+					podPort:     notSelectedPodPort,
+					description: fmt.Sprintf("service port %d (pod port %d)", notSelectedServicePort, notSelectedPodPort),
+				},
+			}
+
+			waitForXDPOnDstWorkloads(t, dst)
+
+			for _, tc := range authzCases {
+				t.ConfigIstio().Eval(apps.Namespace.Name(), map[string]string{
+					"Destination": dst.Config().Service,
+					"Ports":       tc.ports,
+				}, `apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: port-policy
+spec:
+  selector:
+    matchLabels:
+      app: "{{.Destination}}"
+`+tc.spec+`
+  rules:
+  - to:
+    - operation:
+        ports: {{.Ports}}
+`).ApplyOrFail(t)
+
+				for _, portTest := range portTestCases {
+					opt := echo.CallOptions{
+						To:                      dst,
+						Port:                    echo.Port{Name: "tcp", ServicePort: portTest.servicePort},
+						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
+					name = tc.name + ", " + portTest.description
+
+					opt.Check = chooseChecker(tc.name, portTest.servicePort)
+
+					t.NewSubTestf("%v", name).Run(func(t framework.TestContext) {
+						src.WithWorkloads(client).CallOrFail(t, opt)
+					})
+				}
+			}
+		})
+	})
+}
+
+func TestNamespaceAuthorization(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("Namespace Authorization").Run(func(t framework.TestContext) {
+
+			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)
+			client := clients[0]
+			dst := apps.EnrolledToKmesh
+
+			// get the namespace of the apps
+			selectedNamespace := apps.Namespace.Name()
+
+			authzCases := []struct {
+				name string
+				spec string
+			}{
+				{
+					name: "allow",
+					spec: `
+  action: ALLOW
+`,
+				},
+				{
+					name: "deny",
+					spec: `
+  action: DENY
+`,
+				},
+			}
+
+			chooseChecker := func(action string) echo.Checker {
+				switch action {
+				case "allow":
+					return check.OK()
+				case "deny":
+					return check.NotOK()
+				default:
+					t.Fatal("invalid action")
+				}
+
+				return check.OK()
+			}
+
+			waitForXDPOnDstWorkloads(t, dst)
+
+			for _, tc := range authzCases {
+				t.ConfigIstio().Eval(apps.Namespace.Name(), map[string]string{
+					"Destination":     dst.Config().Service,
+					"SourceNamespace": selectedNamespace,
+				}, `apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: namespace-policy
+spec:
+  selector:
+    matchLabels:
+      app: "{{.Destination}}"
+`+tc.spec+`
+  rules:
+  - from:
+    - source:
+        namespaces:
+        - "{{.SourceNamespace}}"
+`).ApplyOrFail(t)
+
+				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
+				name = tc.name + ", namespace " + selectedNamespace
+
+				opt.Check = chooseChecker(tc.name)
+
+				t.NewSubTestf("%v", name).Run(func(t framework.TestContext) {
+					src.WithWorkloads(client).CallOrFail(t, opt)
+				})
+
+			}
+		})
+	})
+}
+
+func TestHeaderAuthorization(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("Header Authorization").Run(func(t framework.TestContext) {
+
+			if len(apps.ServiceWithWaypointAtServiceGranularity) == 0 {
+				t.Fatal(fmt.Errorf("need at least 1 instance of apps.ServiceWithWaypointAtServiceGranularity"))
+			}
+			src := apps.EnrolledToKmesh[0]
+
+			clients := src.WorkloadsOrFail(t)
+			client := clients[0]
+			dst := apps.ServiceWithWaypointAtServiceGranularity
+
+			// Define the test header and ports - use pod port
+			selectedHeaderName := "x-api-key"
+			selectedHeaderValue := "secret-token"
+			notSelectedHeaderValue := "wrong-token"
+			targetHttpServicePort := 80 // Target HTTP Service port
+
+			authzCases := []struct {
+				name string
+				spec string
+			}{
+				{
+					name: "allow",
+					spec: `
+  action: ALLOW
+`,
+				},
+				{
+					name: "deny",
+					spec: `
+  action: DENY
+`,
+				},
+			}
+
+			chooseChecker := func(action string, headerMatches bool) echo.Checker {
+				switch action {
+				case "allow":
+					if !headerMatches {
+						return check.NotOK()
+					} else {
+						return check.OK()
+					}
+				case "deny":
+					if !headerMatches {
+						return check.OK()
+					} else {
+						return check.NotOK()
+					}
+				default:
+					t.Fatal("invalid action")
+				}
+
+				return check.OK()
+			}
+
+			headerTestCases := []struct {
+				headerValue string
+				matches     bool
+				description string
+			}{
+				{
+					headerValue: selectedHeaderValue,
+					matches:     true,
+					description: "matching header",
+				},
+				{
+					headerValue: notSelectedHeaderValue,
+					matches:     false,
+					description: "non-matching header",
+				},
+			}
+
+			waitForXDPOnDstWorkloads(t, dst)
+
+			for _, tc := range authzCases {
+
+				t.ConfigIstio().Eval(apps.Namespace.Name(), map[string]string{
+					"Destination": dst.Config().Service,
+					"HeaderName":  selectedHeaderName,
+					"HeaderValue": selectedHeaderValue,
+				}, `apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: header-policy
+spec:
+  targetRefs:
+  - kind: Service
+    group: ""
+    name: "{{.Destination}}"
+`+tc.spec+`
+  rules:
+  - when:
+    - key: request.headers[{{.HeaderName}}]
+      values: ["{{.HeaderValue}}"]
+`).ApplyOrFail(t)
+
+				for _, headerTest := range headerTestCases {
+					opt := echo.CallOptions{
+						To:     dst,
+						Port:   echo.Port{Name: "http", ServicePort: targetHttpServicePort},
+						Scheme: scheme.HTTP,
+						HTTP: echo.HTTP{
+							Path: "/api/test",
+							Headers: map[string][]string{
+								selectedHeaderName: {headerTest.headerValue},
+							},
+						},
+						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
+					name = fmt.Sprintf("%s, %s on service port %d", tc.name, headerTest.description, targetHttpServicePort)
+
+					opt.Check = chooseChecker(tc.name, headerTest.matches)
+
+					t.NewSubTestf("%v", name).Run(func(t framework.TestContext) {
+						src.WithWorkloads(client).CallOrFail(t, opt)
+					})
+				}
+			}
+		})
+	})
+}
+
+func TestHostAuthorization(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("Host Authorization").Run(func(t framework.TestContext) {
+
+			if len(apps.ServiceWithWaypointAtServiceGranularity) == 0 {
+				t.Fatal(fmt.Errorf("need at least 1 instance of apps.ServiceWithWaypointAtServiceGranularity"))
+			}
+			src := apps.EnrolledToKmesh[0]
+
+			clients := src.WorkloadsOrFail(t)
+			client := clients[0]
+			dst := apps.ServiceWithWaypointAtServiceGranularity
+
+			// Define variables
+			selectedHost := "example.com"
+			notSelectedHost := "wrong.example.com"
+			targetHttpServicePort := 80
+
+			authzCases := []struct {
+				name string
+				spec string
+			}{
+				{
+					name: "allow",
+					spec: `
+  action: ALLOW
+`,
+				},
+				{
+					name: "deny",
+					spec: `
+  action: DENY
+`,
+				},
+			}
+
+			hostTestCases := []struct {
+				hostValue   string
+				matches     bool
+				description string
+			}{
+				{
+					hostValue:   selectedHost,
+					matches:     true,
+					description: "matching host",
+				},
+				{
+					hostValue:   notSelectedHost,
+					matches:     false,
+					description: "non-matching host",
+				},
+			}
+
+			chooseChecker := func(action string, hostMatches bool) echo.Checker {
+				switch action {
+				case "allow":
+					if !hostMatches {
+						return check.NotOK()
+					} else {
+						return check.OK()
+					}
+				case "deny":
+					if !hostMatches {
+						return check.OK()
+					} else {
+						return check.NotOK()
+					}
+				default:
+					t.Fatal("invalid action")
+				}
+
+				return check.OK()
+			}
+
+			waitForXDPOnDstWorkloads(t, dst)
+
+			for _, tc := range authzCases {
+
+				t.ConfigIstio().Eval(apps.Namespace.Name(), map[string]string{
+					"Destination": dst.Config().Service,
+					"TargetHost":  selectedHost,
+				}, `apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: host-policy
+spec:
+  targetRefs:
+  - kind: Service
+    group: ""
+    name: "{{.Destination}}"
+`+tc.spec+`
+  rules:
+  - to:
+    - operation:
+        hosts: ["{{.TargetHost}}"]
+`).ApplyOrFail(t)
+
+				for _, hostTest := range hostTestCases {
+					opt := echo.CallOptions{
+						To:     dst,
+						Port:   echo.Port{Name: "http", ServicePort: targetHttpServicePort},
+						Scheme: scheme.HTTP,
+						HTTP: echo.HTTP{
+							Path: "/api/test",
+							Headers: map[string][]string{
+								"Host": {hostTest.hostValue},
+							},
+						},
+						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
+					name = tc.name + ", " + hostTest.description
+
+					opt.Check = chooseChecker(tc.name, hostTest.matches)
+
+					t.NewSubTestf("%v", name).Run(func(t framework.TestContext) {
+						src.WithWorkloads(client).CallOrFail(t, opt)
+					})
+				}
+
+			}
+		})
+	})
+}
diff --git a/test/e2e/baseline_test.go b/test/e2e/baseline_test.go
index ad5580b1..fd2e09f2 100644
--- a/test/e2e/baseline_test.go
+++ b/test/e2e/baseline_test.go
@@ -29,7 +29,6 @@ import (
 	"net"
 	"net/http"
 	"net/netip"
-	"os/exec"
 	"sort"
 	"strings"
 	"testing"
@@ -751,141 +750,6 @@ func TestMixNsAndServiceWaypoint(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 TestBookinfo(t *testing.T) {
 	framework.NewTest(t).Run(func(t framework.TestContext) {
 		namespace := apps.Namespace.Name()
diff --git a/test/e2e/ipsec_test.go b/test/e2e/ipsec_test.go
new file mode 100644
index 00000000..2121e43e
--- /dev/null
+++ b/test/e2e/ipsec_test.go
@@ -0,0 +1,549 @@
+//go:build integ
+// +build integ
+
+/*
+ * Copyright The Kmesh Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// NOTE: THE CODE IN THIS FILE IS MAINLY REFERENCED FROM ISTIO INTEGRATION
+// FRAMEWORK(https://github.com/istio/istio/tree/master/tests/integration)
+// AND ADAPTED FOR KMESH.
+package kmesh
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"istio.io/istio/pkg/test/framework"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"os/exec"
+	"strings"
+	"testing"
+	"text/template"
+	"time"
+)
+
+type DeployParams struct {
+	Name          string
+	AppLabel      string
+	Image         string
+	NodeName      string
+	SvcPort       int
+	TargetPort    int
+	ContainerPort int
+}
+
+var sleepTmpl = `
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{.Name}}
+  labels:
+    app: {{.AppLabel}}
+    service: {{.AppLabel}}
+spec:
+  ports:
+  - port: 80
+    name: http
+  selector:
+    app: {{.AppLabel}}
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{.Name}}
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{.AppLabel}}
+  template:
+    metadata:
+      labels:
+        app: {{.AppLabel}}
+    spec:
+      nodeName: {{.NodeName}}
+      containers:
+      - name: {{.Name}}
+        image: {{.Image}}
+        command: ["/bin/sleep", "infinity"]
+        imagePullPolicy: IfNotPresent
+`
+
+var echoTmpl = `
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{.Name}}
+  namespace: default
+spec:
+  ports:
+  - name: http
+    port: {{.SvcPort}}
+    targetPort: {{.TargetPort}}
+  selector:
+    app: {{.AppLabel}}
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{.Name}}
+  namespace: default
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{.AppLabel}}
+  template:
+    metadata:
+      labels:
+        app: {{.AppLabel}}
+    spec:
+      containers:
+      - name: {{.Name}}
+        image: {{.Image}}
+        imagePullPolicy: IfNotPresent
+        args:
+        - --port={{.ContainerPort}}
+        ports:
+        - containerPort: {{.ContainerPort}}
+`
+
+func labelNamespace(t framework.TestContext, namespace, key, value string) error {
+	var cmd *exec.Cmd
+	if value == "" {
+		cmd = exec.Command("kubectl", "label", "namespace", namespace, key+"-")
+	} else {
+		cmd = exec.Command("kubectl", "label", "namespace", namespace, key+"="+value, "--overwrite")
+	}
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("kubectl label namespace failed: %v\n%s", err, string(out))
+		return err
+	}
+	t.Logf("namespace %s label set: %s=%s", namespace, key, value)
+	return nil
+}
+
+func deployService(t framework.TestContext, tmplText string, params DeployParams) error {
+	tmpl, err := template.New("yaml").Parse(tmplText)
+	if err != nil {
+		t.Fatalf("failed to parse template: %v", err)
+		return err
+	}
+	var buf bytes.Buffer
+	if err := tmpl.Execute(&buf, params); err != nil {
+		t.Fatalf("failed to execute template: %v", err)
+		return err
+	}
+	cmd := exec.Command("kubectl", "apply", "-f", "-")
+	cmd.Stdin = &buf
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("kubectl apply failed: %v\n%s", err, string(out))
+		return err
+	}
+	t.Logf("applied resources for %s:\n%s", params.Name, string(out))
+	return nil
+}
+
+func deleteService(t framework.TestContext, name, namespace string) error {
+	t.Logf("Deleting deployment and service %s in namespace %s...", name, namespace)
+	cmd := exec.Command("kubectl", "delete", "deployment,service", name, "-n", namespace)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Logf("Failed to delete %s: %v\n%s", name, err, string(out))
+		return err
+	}
+	t.Logf("Deleted %s: %s", name, string(out))
+	return nil
+}
+
+func getPodNameAndIP(t framework.TestContext, namespace string, appLabel string) (string, string, error) {
+	client := t.Clusters().Default().Kube().CoreV1().Pods(namespace)
+	selector := "app=" + appLabel
+
+	for {
+		podList, err := client.List(context.TODO(), metav1.ListOptions{LabelSelector: selector})
+		if err != nil {
+			t.Logf("list pods failed: %v, retrying...", err)
+			time.Sleep(3 * time.Second)
+			continue
+		}
+		if len(podList.Items) == 0 {
+			t.Logf("no pods found for app=%s, retrying...", appLabel)
+			time.Sleep(3 * time.Second)
+			continue
+		}
+		for _, p := range podList.Items {
+			if p.Status.Phase == corev1.PodRunning && p.Status.PodIP != "" && p.Status.PodIP != "<none>" {
+				return p.Name, p.Status.PodIP, nil
+			}
+			t.Logf("pod %s status=%s ip=%s, waiting...", p.Name, p.Status.Phase, p.Status.PodIP)
+		}
+		time.Sleep(3 * time.Second)
+	}
+}
+
+func downloadTcpdump(t framework.TestContext, containerName string) error {
+	checkCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "which tcpdump")
+	out, err := checkCmd.CombinedOutput()
+	if err == nil && strings.TrimSpace(string(out)) != "" {
+		t.Logf("tcpdump is already installed in container: %s", containerName)
+		return nil
+	}
+	t.Logf("tcpdump not found, installing in container: %s", containerName)
+
+	updateCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "apt update")
+	out, err = updateCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("apt update failed: %v\n%s", err, string(out))
+		return err
+	}
+	t.Logf("apt update success")
+
+	installCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "apt install tcpdump -y")
+	out, err = installCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("apt install tcpdump failed: %v\n%s", err, string(out))
+		return err
+	}
+	t.Logf("apt install tcpdump success")
+	return nil
+}
+
+func checkIPSecRules(t framework.TestContext, containerName string) error {
+	stateCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "ip xfrm state")
+	stateOut, err := stateCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("failed to run ip xfrm state: %v\n%s", err, string(stateOut))
+		return err
+	}
+	t.Logf("ip xfrm state output:\n%s", string(stateOut))
+
+	policyCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "ip xfrm policy")
+	policyOut, err := policyCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("failed to run ip xfrm policy: %v\n%s", err, string(policyOut))
+		return err
+	}
+	t.Logf("ip xfrm policy output:\n%s", string(policyOut))
+
+	stateStr := string(stateOut)
+	policyStr := string(policyOut)
+
+	if !strings.Contains(stateStr, "proto esp spi") || !strings.Contains(stateStr, "mode tunnel") {
+		t.Fatalf("ip xfrm state output does not contain expected ESP tunnel rules")
+		return fmt.Errorf("ip xfrm state output does not contain expected ESP tunnel rules")
+	}
+	if !strings.Contains(stateStr, "aead rfc4106(gcm(aes))") {
+		t.Fatalf("ip xfrm state output does not contain expected AEAD algorithm")
+		return fmt.Errorf("ip xfrm state output does not contain expected AEAD algorithm")
+	}
+
+	if !strings.Contains(policyStr, "tmpl src") || !strings.Contains(policyStr, "proto esp") || !strings.Contains(policyStr, "mode tunnel") {
+		t.Fatalf("ip xfrm policy output does not contain expected tunnel template")
+		return fmt.Errorf("ip xfrm policy output does not contain expected tunnel template")
+	}
+	if !strings.Contains(policyStr, "dir out") || !strings.Contains(policyStr, "dir in") || !strings.Contains(policyStr, "dir fwd") {
+		t.Fatalf("ip xfrm policy output does not contain expected directions (out/in/fwd)")
+		return fmt.Errorf("ip xfrm policy output does not contain expected directions (out/in/fwd)")
+	}
+	if !strings.Contains(policyStr, "mark") {
+		t.Fatalf("ip xfrm policy output does not contain expected mark field")
+		return fmt.Errorf("ip xfrm policy output does not contain expected mark field")
+	}
+
+	t.Logf("IPSec xfrm state and policy outputs match expected format.")
+	return nil
+}
+
+func runTcpdumpESP(t framework.TestContext, containerName string, duration int) (string, error) {
+	cmd := exec.Command("docker", "exec", containerName, "bash", "-c",
+		fmt.Sprintf("timeout %d tcpdump -i any | grep ESP", duration))
+	out, err := cmd.CombinedOutput()
+	if err != nil && !strings.Contains(err.Error(), "exit status 124") {
+		t.Fatalf("tcpdump failed: %v\n%s", err, string(out))
+		return "", err
+	}
+	return string(out), nil
+}
+
+func curlFromSleepToEcho(t framework.TestContext, sleepPod, echoIP string, echoPort int) (string, error) {
+	t.Logf("Curling from sleep pod %s to echo pod IP %s...", sleepPod, echoIP)
+	var url string
+	if strings.Contains(echoIP, ":") {
+		// IPv6 address
+		url = fmt.Sprintf("http://[%s]:%d", echoIP, echoPort)
+	} else {
+		// IPv4 address
+		url = fmt.Sprintf("http://%s:%d", echoIP, echoPort)
+	}
+	cmd := exec.Command("kubectl", "exec", sleepPod, "--", "curl", url)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("curl failed: %v\n%s", err, string(out))
+		return "", err
+	}
+	return string(out), nil
+}
+
+func genRandomKeyHex() string {
+	b := make([]byte, 36)
+	_, err := rand.Read(b)
+	if err != nil {
+		panic("failed to generate random key")
+	}
+	return hex.EncodeToString(b)
+}
+
+func rotateIPSecKeyAndWait(t framework.TestContext, containerName string, intervalSec int) error {
+	t.Logf("Rotating IPSec key with kmeshctl secret ...")
+	keyHex := genRandomKeyHex()
+	rotateCmd := exec.Command("kmeshctl", "secret", "create", "--key="+keyHex)
+	out, err := rotateCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("Failed to rotate IPSec key: %v\n%s", err, string(out))
+		return err
+	}
+	t.Logf("kmeshctl secret update success")
+
+	t.Logf("Waiting for SPI 0x00000002 to appear in both state and policy (interval %ds)...", intervalSec)
+	for {
+		time.Sleep(time.Duration(intervalSec) * time.Second)
+		stateCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "ip xfrm state")
+		stateOut, err := stateCmd.CombinedOutput()
+		if err != nil {
+			t.Logf("ip xfrm state failed: %v\n%s", err, string(stateOut))
+			continue
+		}
+		policyCmd := exec.Command("docker", "exec", containerName, "bash", "-c", "ip xfrm policy")
+		policyOut, err := policyCmd.CombinedOutput()
+		if err != nil {
+			t.Logf("ip xfrm policy failed: %v\n%s", err, string(policyOut))
+			continue
+		}
+		stateHasSPI := strings.Contains(string(stateOut), "spi 0x00000002")
+		policyHasSPI := strings.Contains(string(policyOut), "spi 0x00000002")
+		if stateHasSPI && policyHasSPI {
+			t.Logf("Found new SPI 0x00000002 in both ip xfrm state and policy!")
+			return nil
+		}
+		t.Logf("SPI 0x00000002 not found in both state and policy, waiting...")
+	}
+}
+
+func TestIPsec(t *testing.T) {
+	secretCmd := exec.Command("kmeshctl", "secret", "create")
+	out, err := secretCmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("Failed to rotate IPSec key: %v\n%s", err, string(out))
+		return
+	}
+
+	// wait until both ip xfrm state and policy are non-empty before continuing
+	for {
+		stateCmd := exec.Command("docker", "exec", "kmesh-testing-control-plane", "bash", "-c", "ip xfrm state")
+		stateOut, _ := stateCmd.CombinedOutput()
+
+		policyCmd := exec.Command("docker", "exec", "kmesh-testing-control-plane", "bash", "-c", "ip xfrm policy")
+		policyOut, _ := policyCmd.CombinedOutput()
+
+		if strings.TrimSpace(string(stateOut)) != "" && strings.TrimSpace(string(policyOut)) != "" {
+			t.Logf("ip xfrm state and policy populated")
+			break
+		}
+
+		t.Logf("waiting for ip xfrm state/policy to be populated (stateLen=%d policyLen=%d)...", len(stateOut), len(policyOut))
+		time.Sleep(3 * time.Second)
+	}
+
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("IPsec Connectivity").Run(func(t framework.TestContext) {
+
+			// prepare params
+			if err := labelNamespace(t, "default", "istio.io/dataplane-mode", "Kmesh"); err != nil {
+				return
+			}
+			sleepParams := DeployParams{
+				Name:     "sleep-1",
+				AppLabel: "sleep-1",
+				Image:    "curlimages/curl",
+				NodeName: "kmesh-testing-control-plane",
+			}
+			echoParams := DeployParams{
+				Name:          "echo-1",
+				AppLabel:      "echo-1",
+				Image:         "gcr.io/istio-testing/app:latest",
+				SvcPort:       80,
+				TargetPort:    8080,
+				ContainerPort: 8080,
+			}
+
+			// deploy
+			if err := deployService(t, sleepTmpl, sleepParams); err != nil {
+				return
+			}
+			if err := deployService(t, echoTmpl, echoParams); err != nil {
+				_ = deleteService(t, sleepParams.Name, "default")
+				return
+			}
+
+			// cleanup
+			t.Cleanup(func() {
+				_ = deleteService(t, sleepParams.Name, "default")
+				_ = deleteService(t, echoParams.Name, "default")
+				_ = labelNamespace(t, "default", "istio.io/dataplane-mode", "")
+			})
+
+			t.Logf("Resources applied. Waiting for pods...")
+
+			sleepPodName, sleepPodIP, err := getPodNameAndIP(t, "default", sleepParams.AppLabel)
+			if err != nil {
+				return
+			}
+			t.Logf("sleep pod name: %s, ip: %s", sleepPodName, sleepPodIP)
+
+			echoPodName, echoPodIP, err := getPodNameAndIP(t, "default", echoParams.AppLabel)
+			if err != nil {
+				return
+			}
+			t.Logf("echo pod name: %s, ip: %s", echoPodName, echoPodIP)
+
+			if err = downloadTcpdump(t, "kmesh-testing-control-plane"); err != nil {
+				return
+			}
+			if err = checkIPSecRules(t, "kmesh-testing-control-plane"); err != nil {
+				return
+			}
+
+			t.Logf("Starting tcpdump in kmesh-testing-control-plane container...")
+			tcpdumpCh := make(chan string)
+			go func() {
+				out, _ := runTcpdumpESP(t, "kmesh-testing-control-plane", 10)
+				tcpdumpCh <- out
+			}()
+			time.Sleep(2 * time.Second)
+
+			t.Logf("Curling from sleep pod to echo pod...")
+			_, err = curlFromSleepToEcho(t, sleepPodName, echoPodIP, echoParams.ContainerPort)
+			if err != nil {
+				<-tcpdumpCh
+				return
+			}
+			t.Logf("curl success")
+			tcpdumpOut := <-tcpdumpCh
+
+			if strings.Contains(tcpdumpOut, "ESP") {
+				t.Logf("Connectivity test success: ESP packets detected during curl")
+			} else {
+				t.Fatalf("Connectivity test failed: No ESP packets detected.")
+			}
+		})
+	})
+
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		t.NewSubTest("IPsec Key Rotation").Run(func(t framework.TestContext) {
+
+			// prepare params
+			if err := labelNamespace(t, "default", "istio.io/dataplane-mode", "Kmesh"); err != nil {
+				return
+			}
+			sleepParams := DeployParams{
+				Name:     "sleep-2",
+				AppLabel: "sleep-2",
+				Image:    "curlimages/curl",
+				NodeName: "kmesh-testing-control-plane",
+			}
+			echoParams := DeployParams{
+				Name:          "echo-2",
+				AppLabel:      "echo-2",
+				Image:         "gcr.io/istio-testing/app:latest",
+				SvcPort:       80,
+				TargetPort:    8080,
+				ContainerPort: 8080,
+			}
+
+			// deploy
+			if err := deployService(t, sleepTmpl, sleepParams); err != nil {
+				return
+			}
+			if err := deployService(t, echoTmpl, echoParams); err != nil {
+				_ = deleteService(t, sleepParams.Name, "default")
+				return
+			}
+
+			// cleanup
+			t.Cleanup(func() {
+				_ = deleteService(t, sleepParams.Name, "default")
+				_ = deleteService(t, echoParams.Name, "default")
+				_ = labelNamespace(t, "default", "istio.io/dataplane-mode", "")
+			})
+
+			t.Logf("Resources applied. Waiting for pods...")
+
+			sleepPodName, sleepPodIP, err := getPodNameAndIP(t, "default", sleepParams.AppLabel)
+			if err != nil {
+				return
+			}
+			t.Logf("sleep pod name: %s, ip: %s", sleepPodName, sleepPodIP)
+
+			echoPodName, echoPodIP, err := getPodNameAndIP(t, "default", echoParams.AppLabel)
+			if err != nil {
+				return
+			}
+			t.Logf("echo pod name: %s, ip: %s", echoPodName, echoPodIP)
+
+			if err = downloadTcpdump(t, "kmesh-testing-control-plane"); err != nil {
+				return
+			}
+			if err = checkIPSecRules(t, "kmesh-testing-control-plane"); err != nil {
+				return
+			}
+
+			// rotate key and wait for new SPI in both state and policy
+			if err := rotateIPSecKeyAndWait(t, "kmesh-testing-control-plane", 5); err != nil {
+				return
+			}
+
+			t.Logf("Starting tcpdump in kmesh-testing-control-plane container after key rotation...")
+			tcpdumpCh := make(chan string)
+			go func() {
+				out, _ := runTcpdumpESP(t, "kmesh-testing-control-plane", 10)
+				tcpdumpCh <- out
+			}()
+			time.Sleep(2 * time.Second)
+
+			t.Logf("Curling from sleep pod to echo pod after key rotation...")
+			_, err = curlFromSleepToEcho(t, sleepPodName, echoPodIP, echoParams.ContainerPort)
+			if err != nil {
+				<-tcpdumpCh
+				return
+			}
+			t.Logf("curl after key rotation success")
+			tcpdumpOut := <-tcpdumpCh
+			t.Logf("tcpdump output after key rotation:\n%s", tcpdumpOut)
+
+			if strings.Contains(tcpdumpOut, "ESP") && strings.Contains(tcpdumpOut, "spi=0x00000002") {
+				t.Logf("Key rotation test success: ESP packets with new SPI detected after key rotation")
+			} else {
+				t.Fatalf("Key rotation test failed: No ESP packets with new SPI detected after key rotation.")
+			}
+		})
+	})
+}
diff --git a/test/e2e/run_test.sh b/test/e2e/run_test.sh
index 0fabb50e..0a670b7d 100755
--- a/test/e2e/run_test.sh
+++ b/test/e2e/run_test.sh
@@ -117,7 +117,7 @@ function setup_kmesh() {
 	# skip dns proxy for ipv6
 	[[ -n ${IPV6:-} ]] && extra_args="--set features.dnsProxy.enabled=false"
 	helm install kmesh $ROOT_DIR/deploy/charts/kmesh-helm -n kmesh-system --create-namespace --set deploy.kmesh.image.repository=localhost:5000/kmesh \
-		--set deploy.kmesh.containers.kmeshDaemonArgs="--mode=dual-engine --enable-bypass=false --monitoring=true" \
+		--set deploy.kmesh.containers.kmeshDaemonArgs="--mode=dual-engine --enable-bypass=false --monitoring=true --enable-ipsec=true" \
 		$extra_args
 
 	# Wait for all Kmesh pods to be ready.