Fix NP not working on Hairpin

Fix #5681

Network policy didn't work when using a server Pod to establish a
connection to the service provided by itself. This hairpin service
connection initiated through a local Pod will be SNATed to the
gateway IP, which will prevent it from being correctly categorized by
the network policy during the Ingress rule enforcement.

This commit added a bypass flow to always allow the hairpin service
connection to address this issue. Given we don't consider self-access
blocking to be a valid case.

Signed-off-by: graysonwu <wgrayson@vmware.com>

pkg/agent/openflow/network_policy_test.go

@@ -1387,6 +1387,7 @@ func networkPolicyInitFlows(ovsMeterSupported, externalNodeEnabled, l7NetworkPol
 		"cookie=0x1020000000000, table=IngressSecurityClassifier, priority=200,reg0=0x20/0xf0 actions=goto_table:IngressMetric",
 		"cookie=0x1020000000000, table=IngressSecurityClassifier, priority=200,reg0=0x10/0xf0 actions=goto_table:IngressMetric",
 		"cookie=0x1020000000000, table=IngressSecurityClassifier, priority=200,reg0=0x40/0xf0 actions=goto_table:IngressMetric",
+		"cookie=0x1020000000000, table=IngressSecurityClassifier, priority=200,ct_mark=0x40/0x40 actions=goto_table:ConntrackCommit",
 		"cookie=0x1020000000000, table=AntreaPolicyEgressRule, priority=64990,ct_state=-new+est,ip actions=goto_table:EgressMetric",
 		"cookie=0x1020000000000, table=AntreaPolicyEgressRule, priority=64990,ct_state=-new+rel,ip actions=goto_table:EgressMetric",
 		"cookie=0x1020000000000, table=AntreaPolicyIngressRule, priority=64990,ct_state=-new+est,ip actions=goto_table:IngressMetric",

pkg/agent/openflow/pipeline.go

@@ -2170,6 +2170,12 @@ func (f *featureNetworkPolicy) ingressClassifierFlows() []binding.Flow {
 			MatchRegMark(ToUplinkRegMark).
 			Action().GotoTable(IngressMetricTable.GetID()).
 			Done(),
+		// This generates the flow to match the hairpin service packets and forward them to stageConntrack.
+		IngressSecurityClassifierTable.ofTable.BuildFlow(priorityNormal).
+			Cookie(cookieID).
+			MatchCTMark(HairpinCTMark).
+			Action().GotoStage(stageConntrack).
+			Done(),
 	}
 	if f.enableAntreaPolicy && f.proxyAll {
 		// This generates the flow to match the NodePort Service packets and forward them to AntreaPolicyIngressRuleTable.

test/e2e/antreapolicy_test.go

@@ -3445,7 +3445,7 @@ func testToServices(t *testing.T) {
 		services = append(services, ipv4Svc)
 	}
 	if clusterInfo.podV6NetworkCIDR != "" {
-		ipv6Svc := k8sUtils.BuildService("ipv6-svc", namespaces["x"], 80, 80, map[string]string{"pod": "b"}, nil)
+		ipv6Svc := k8sUtils.BuildService("ipv6-svc", namespaces["x"], 80, 80, map[string]string{"pod": "a"}, nil)
 		ipv6Svc.Spec.IPFamilies = []v1.IPFamily{v1.IPv6Protocol}
 		services = append(services, ipv6Svc)
 	}
@@ -3466,7 +3466,8 @@ func testToServices(t *testing.T) {
 	builder = builder.SetName("test-acnp-to-services").
 		SetTier("application").
 		SetPriority(1.0)
-	builder.AddToServicesRule(svcRefs, "svc", []ACNPAppliedToSpec{{NSSelector: map[string]string{"ns": namespaces["y"]}}}, crdv1alpha1.RuleActionDrop)
+	builder.AddToServicesRule(svcRefs, "x-to-svc", []ACNPAppliedToSpec{{NSSelector: map[string]string{"ns": namespaces["x"]}}}, crdv1alpha1.RuleActionDrop)
+	builder.AddToServicesRule(svcRefs, "y-to-svc", []ACNPAppliedToSpec{{NSSelector: map[string]string{"ns": namespaces["y"]}}}, crdv1alpha1.RuleActionDrop)
 	time.Sleep(networkPolicyDelay)
 
 	acnp := builder.Get()
@@ -3476,6 +3477,12 @@ func testToServices(t *testing.T) {
 	var testcases []podToAddrTestStep
 	for _, service := range builtSvcs {
 		eachServiceCases := []podToAddrTestStep{
+			{
+				Pod(namespaces["x"] + "/a"),
+				service.Spec.ClusterIP,
+				service.Spec.Ports[0].Port,
+				Dropped,
+			},
 			{
 				Pod(namespaces["y"] + "/b"),
 				service.Spec.ClusterIP,

test/e2e/networkpolicy_test.go

@@ -81,6 +81,10 @@ func TestNetworkPolicy(t *testing.T) {
 	t.Run("testIngressPolicyWithEndPort", func(t *testing.T) {
 		testIngressPolicyWithEndPort(t, data)
 	})
+	t.Run("testAllowHairpinService", func(t *testing.T) {
+		t.Cleanup(exportLogsForSubtest(t, data))
+		testAllowHairpinService(t, data)
+	})
 }
 
 func testNetworkPolicyStats(t *testing.T, data *TestData) {
@@ -919,6 +923,49 @@ func testIngressPolicyWithEndPort(t *testing.T, data *TestData) {
 	}
 }
 
+func testAllowHairpinService(t *testing.T, data *TestData) {
+	serverNode := workerNodeName(1)
+	serverPort := int32(80)
+	serverName, _, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", serverNode, data.testNamespace, false)
+	defer cleanupFunc()
+
+	service, err := data.CreateService("nginx", data.testNamespace, serverPort, serverPort, map[string]string{"app": "nginx"}, false, false, corev1.ServiceTypeClusterIP, nil)
+	if err != nil {
+		t.Fatalf("Error when creating nginx service: %v", err)
+	}
+	defer data.deleteService(service.Namespace, service.Name)
+
+	clientName, _, cleanupFunc := createAndWaitForPod(t, data, data.createBusyboxPodOnNode, "test-client-", serverNode, data.testNamespace, false)
+	defer cleanupFunc()
+
+	spec := &networkingv1.NetworkPolicySpec{
+		PodSelector: metav1.LabelSelector{},
+		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
+	}
+	np, err := data.createNetworkPolicy("test-networkpolicy-ns-iso", spec)
+	if err != nil {
+		t.Fatalf("Error when creating network policy: %v", err)
+	}
+	defer func() {
+		if err = data.deleteNetworkpolicy(np); err != nil {
+			t.Fatalf("Error when deleting network policy: %v", err)
+		}
+	}()
+
+	npCheck := func(clientName, serverIP, containerName string, serverPort int32, wantErr bool) {
+		if err = data.runNetcatCommandFromTestPodWithProtocol(clientName, data.testNamespace, containerName, serverIP, serverPort, "tcp"); wantErr && err == nil {
+			t.Fatalf("Pod %s should not be able to connect %s, but was able to connect", clientName, net.JoinHostPort(serverIP, fmt.Sprint(serverPort)))
+		} else if !wantErr && err != nil {
+			t.Fatalf("Pod %s should be able to connect %s, but was not able to connect", clientName, net.JoinHostPort(serverIP, fmt.Sprint(serverPort)))
+		}
+	}
+
+	for _, clusterIP := range service.Spec.ClusterIPs {
+		npCheck(serverName, clusterIP, nginxContainerName, serverPort, false)
+		npCheck(clientName, clusterIP, busyboxContainerName, serverPort, true)
+	}
+}
+
 func createAndWaitForPod(t *testing.T, data *TestData, createFunc func(name string, ns string, nodeName string, hostNetwork bool) error, namePrefix string, nodeName string, ns string, hostNetwork bool) (string, *PodIPs, func()) {
 	name := randName(namePrefix)
 	return createAndWaitForPodWithExactName(t, data, createFunc, name, nodeName, ns, hostNetwork)