Fix the issue of local probe bypassing flows on Windows

pkg/agent/openflow/client.go

@@ -697,7 +697,8 @@ func (c *client) generatePipelines() {
 		c.networkConfig,
 		c.ovsDatapathType,
 		c.connectUplinkToBridge,
-		c.enableMulticast)
+		c.enableMulticast,
+		c.proxyAll)
 	c.activatedFeatures = append(c.activatedFeatures, c.featurePodConnectivity)
 	c.traceableFeatures = append(c.traceableFeatures, c.featurePodConnectivity)
 

pkg/agent/openflow/fields.go

@@ -173,11 +173,12 @@ var (
 	FromGatewayCTMark     = binding.NewCTMark(ConnSourceCTMarkField, gatewayVal)
 	FromBridgeCTMark      = binding.NewCTMark(ConnSourceCTMarkField, bridgeVal)
 
-	// CTMark[4]: Mark to indicate DNAT is performed on the connection for Service.
-	// This CT mark is used in CtZone / CtZoneV6 and SNATCtZone / SNATCtZoneV6.
-	ServiceCTMark = binding.NewOneBitCTMark(4)
+	// CTMark[4]: Marks to indicate whether DNAT is performed on the connection for Service.
+	// These CT marks are used in CtZone / CtZoneV6 and SNATCtZone / SNATCtZoneV6.
+	ServiceCTMark    = binding.NewOneBitCTMark(4)
+	NotServiceCTMark = binding.NewOneBitZeroCTMark(4)
 
-	// CTMark[5]: Mark to indicate SNAT should be performed on the connection for Service.
+	// CTMark[5]: Mark to indicate SNAT is performed on the connection for Service.
 	// This CT mark is only used in CtZone / CtZoneV6.
 	ConnSNATCTMark = binding.NewOneBitCTMark(5)
 

pkg/agent/openflow/pipeline.go

@@ -2094,34 +2094,48 @@ 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:
-// 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.
+// 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() []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 NetworkPolicies. See https://github.com/antrea-io/antrea/issues/280.
+func (f *featurePodConnectivity) localProbeFlows() []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
 	if runtime.IsWindowsPlatform() || f.ovsDatapathType == ovsconfig.OVSDatapathNetdev {
+		var ctMarksToMatch []*binding.CtMark
+		if f.proxyAll {
+			ctMarksToMatch = append(ctMarksToMatch, NotServiceCTMark)
+		}
 		for ipProtocol, gatewayIP := range f.gatewayIPs {
 			flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
 				Cookie(cookieID).
 				MatchProtocol(ipProtocol).
+				MatchCTStateRpl(false).
+				MatchCTStateTrk(true).
 				MatchSrcIP(gatewayIP).
+				MatchCTMark(ctMarksToMatch...).
 				Action().GotoStage(stageConntrack).
 				Done())
 		}
 	} else {
-		flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
-			Cookie(cookieID).
-			MatchPktMark(types.HostLocalSourceMark, &types.HostLocalSourceMark).
-			Action().GotoStage(stageConntrack).
-			Done())
+		for _, ipProtocol := range f.ipProtocols {
+			flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
+				Cookie(cookieID).
+				MatchProtocol(ipProtocol).
+				MatchCTStateRpl(false).
+				MatchCTStateTrk(true).
+				MatchPktMark(types.HostLocalSourceMark, &types.HostLocalSourceMark).
+				Action().GotoStage(stageConntrack).
+				Done())
+		}
 	}
 	return flows
 }

pkg/agent/openflow/pod_connectivity.go

@@ -44,6 +44,7 @@ type featurePodConnectivity struct {
 	ctZoneSrcField        *binding.RegField
 	ipCtZoneTypeRegMarks  map[binding.Protocol]*binding.RegMark
 	enableMulticast       bool
+	proxyAll              bool
 
 	category cookie.Category
 }
@@ -59,7 +60,8 @@ func newFeaturePodConnectivity(
 	networkConfig *config.NetworkConfig,
 	ovsDatapathType ovsconfig.OVSDatapathType,
 	connectUplinkToBridge bool,
-	enableMulticast bool) *featurePodConnectivity {
+	enableMulticast bool,
+	proxyAll bool) *featurePodConnectivity {
 	ctZones := make(map[binding.Protocol]int)
 	gatewayIPs := make(map[binding.Protocol]net.IP)
 	localCIDRs := make(map[binding.Protocol]net.IPNet)
@@ -101,6 +103,7 @@ func newFeaturePodConnectivity(
 		ipCtZoneTypeRegMarks:  ipCtZoneTypeRegMarks,
 		ctZoneSrcField:        getZoneSrcField(connectUplinkToBridge),
 		enableMulticast:       enableMulticast,
+		proxyAll:              proxyAll,
 		category:              cookie.PodConnectivity,
 	}
 }
@@ -140,7 +143,7 @@ func (f *featurePodConnectivity) initFlows() []binding.Flow {
 	flows = append(flows, f.gatewayIPSpoofGuardFlows()...)
 	flows = append(flows, f.l3FwdFlowToGateway()...)
 	// Add flow to ensure the liveliness check packet could be forwarded correctly.
-	flows = append(flows, f.localProbeFlow()...)
+	flows = append(flows, f.localProbeFlows()...)
 
 	if f.networkConfig.TrafficEncapMode.SupportsEncap() {
 		flows = append(flows, f.tunnelClassifierFlow(config.DefaultTunOFPort))

pkg/ovs/openflow/ofctrl_builder.go

@@ -244,6 +244,9 @@ func (b *ofFlowBuilder) MatchCTStateSNAT(set bool) FlowBuilder {
 }
 
 func (b *ofFlowBuilder) MatchCTMark(marks ...*CtMark) FlowBuilder {
+	if len(marks) == 0 {
+		return b
+	}
 	var value, mask uint32
 	for _, mark := range marks {
 		value |= mark.GetValue()

test/integration/agent/openflow_test.go

@@ -1222,7 +1222,7 @@ func prepareGatewayFlows(gwIPs []net.IP, gwMAC net.HardwareAddr, vMAC net.Hardwa
 				tableName: "IngressSecurityClassifier",
 				flows: []*ofTestUtils.ExpectFlow{
 					{
-						MatchStr: fmt.Sprintf("priority=210,%s,%s=%s", ipProtoStr, nwSrcStr, gwIP.String()),
+						MatchStr: fmt.Sprintf("priority=210,ct_state=-rpl+trk,%s,%s=%s", ipProtoStr, nwSrcStr, gwIP.String()),
 						ActStr:   "goto_table:ConntrackCommit",
 					},
 				},