Fix the issue of local probe bypassing flows on Windows

When proxyAll is enabled, kube-proxy can be replaced by AntreaProxy, then
Service traffic and non-Service traffic can be distinguished by ServiceCTMark
and NotServiceCTMark. Service traffic with ServiceCTMark should not bypass
Network Policies, and non-Service traffic generated by kubelet with
NotServiceCTMark should bypass Network Policies.

Signed-off-by: Hongliang Liu <[email protected]>

pkg/agent/openflow/client.go

@@ -688,7 +688,7 @@ func (c *client) InstallGatewayFlows() error {
 	}
 
 	// Add flow to ensure the liveness check packet could be forwarded correctly.
-	flows = append(flows, c.featurePodConnectivity.localProbeFlow(c.ovsDatapathType)...)
+	flows = append(flows, c.featurePodConnectivity.localProbeFlows(c.ovsDatapathType, c.proxyAll)...)
 	flows = append(flows, c.featurePodConnectivity.l3FwdFlowToGateway()...)
 
 	if err := c.ofEntryOperations.AddAll(flows); err != nil {

pkg/agent/openflow/fields.go

@@ -177,6 +177,9 @@ var (
 	// This CT mark is used in CtZone / CtZoneV6 and SNATCtZone / SNATCtZoneV6.
 	ServiceCTMark = binding.NewOneBitCTMark(4)
 
+	// CTMark[4]: Mark to indicate the connection is not for Service.
+	NotServiceCTMark = binding.NewOneBitZeroCTMark(4)
+
 	// CTMark[5]: Mark to indicate SNAT should be performed on the connection for Service.
 	// This CT mark is only used in CtZone / CtZoneV6.
 	ConnSNATCTMark = binding.NewOneBitCTMark(5)

pkg/agent/openflow/pipeline.go

@@ -2112,34 +2112,46 @@ func (f *featureNetworkPolicy) dnsPacketInFlow(conjunctionID uint32) binding.Flo
 		Done()
 }
 
-// localProbeFlow generates the flow to forward locally generated packets to ConntrackCommitTable, bypassing ingress
-// rules of Network Policies. The packets are sent by kubelet to probe the liveness/readiness of local Pods.
-// On Linux and when OVS kernel datapath is used, it identifies locally generated packets by matching the
-// HostLocalSourceMark, otherwise it matches the source IP. The difference is because:
+// localProbeFlows generates the flows to forward locally generated request packets to stageConntrack directly, bypassing
+// ingress rules of Network Policies. The packets are sent by kubelet to probe the liveness/readiness of local Pods.
+// On Linux and when OVS kernel datapath is used, the probe packets are identified by matching the HostLocalSourceMark.
+// On Windows or when OVS userspace (netdev) datapath is used, we need a different approach because:
 // 1. On Windows, kube-proxy userspace mode is used, and currently there is no way to distinguish kubelet generated
 //    traffic from kube-proxy proxied traffic.
 // 2. pkt_mark field is not properly supported for OVS userspace (netdev) datapath.
-// Note that there is a defect in the latter way that NodePort Service access by external clients will be masqueraded as
-// a local gateway IP to bypass Network Policies. See https://github.com/antrea-io/antrea/issues/280.
-// TODO: Fix it after replacing kube-proxy with AntreaProxy.
-func (f *featurePodConnectivity) localProbeFlow(ovsDatapathType ovsconfig.OVSDatapathType) []binding.Flow {
+// when proxyAll is disabled, the probe packets are identified by matching the source IP is the Antrea gateway IP;
+// otherwise, the packets are identified by matching both the Antrea gateway IP and NotServiceCTMark. Note that, when
+// proxyAll is disabled, currently there is no way to distinguish kubelet generated traffic from kube-proxy proxied traffic
+// only by matching the Antrea gateway IP. There is a defect that NodePort Service access by external clients will be
+// masqueraded as the Antrea gateway IP to bypass Network Policies. See https://github.com/antrea-io/antrea/issues/280.
+func (f *featurePodConnectivity) localProbeFlows(ovsDatapathType ovsconfig.OVSDatapathType, proxyAll bool) []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
 	if runtime.IsWindowsPlatform() || ovsDatapathType == ovsconfig.OVSDatapathNetdev {
 		for ipProtocol, gatewayIP := range f.gatewayIPs {
+			fb := IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
+				Cookie(cookieID).
+				MatchProtocol(ipProtocol).
+				MatchCTStateRpl(false).
+				MatchCTStateTrk(true).
+				MatchSrcIP(gatewayIP)
+			if proxyAll {
+				fb = fb.MatchCTMark(NotServiceCTMark)
+			}
+			flows = append(flows, fb.Action().GotoStage(stageConntrack).
+				Done())
+		}
+	} else {
+		for _, ipProtocol := range f.ipProtocols {
 			flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
 				Cookie(cookieID).
 				MatchProtocol(ipProtocol).
-				MatchSrcIP(gatewayIP).
+				MatchCTStateRpl(false).
+				MatchCTStateTrk(true).
+				MatchPktMark(types.HostLocalSourceMark, &types.HostLocalSourceMark).
 				Action().GotoStage(stageConntrack).
 				Done())
 		}
-	} else {
-		flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
-			Cookie(cookieID).
-			MatchPktMark(types.HostLocalSourceMark, &types.HostLocalSourceMark).
-			Action().GotoStage(stageConntrack).
-			Done())
 	}
 	return flows
 }