Add e2e test

Signed-off-by: Dyanngg <dingyang@vmware.com>

pkg/agent/openflow/network_policy_test.go

@@ -65,6 +65,7 @@ var (
 	actionAllow  = crdv1alpha1.RuleActionAllow
 	actionDrop   = crdv1alpha1.RuleActionDrop
 	port8080     = intstr.FromInt(8080)
+	port32800    = int32(32800)
 	protocolICMP = v1beta2.ProtocolICMP
 	priority100  = uint16(100)
 	priority200  = uint16(200)
@@ -469,6 +470,22 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						UID:       "id4",
 					},
 				},
+				{
+					Direction: v1beta2.DirectionIn,
+					From:      parseAddresses([]string{"192.168.1.51"}),
+					Action:    &actionDrop,
+					Priority:  &priority100,
+					To:        []types.Address{NewOFPortAddress(2)},
+					Service:   []v1beta2.Service{{Protocol: &protocolTCP, SrcPort: &port32800}},
+					FlowID:    uint32(14),
+					TableID:   AntreaPolicyIngressRuleTable.GetID(),
+					PolicyRef: &v1beta2.NetworkPolicyReference{
+						Type:      v1beta2.AntreaNetworkPolicy,
+						Namespace: "ns1",
+						Name:      "np5",
+						UID:       "id5",
+					},
+				},
 			},
 			expectedFlowsFn: func(c *client) []binding.Flow {
 				cookiePolicy := c.cookieAllocator.Request(cookie.NetworkPolicy).Raw()
@@ -492,6 +509,11 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						Action().LoadToRegField(CNPConjIDField, 13).
 						Action().LoadRegMark(CnpDenyRegMark).
 						Action().GotoTable(IngressMetricTable.GetID()).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
+						MatchConjID(14).
+						Action().LoadToRegField(CNPConjIDField, 14).
+						Action().LoadRegMark(CnpDenyRegMark).
+						Action().GotoTable(IngressMetricTable.GetID()).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolIP).MatchSrcIP(net.ParseIP("192.168.1.40")).
 						Action().Conjunction(10, 1, 2).
@@ -504,7 +526,8 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						Action().Conjunction(10, 1, 2).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolIP).MatchSrcIP(net.ParseIP("192.168.1.51")).
-						Action().Conjunction(11, 1, 3).Done(),
+						Action().Conjunction(11, 1, 3).
+						Action().Conjunction(14, 1, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority201).Cookie(cookiePolicy).
 						MatchTunnelID(1).
 						Action().Conjunction(13, 1, 3).Done(),
@@ -520,7 +543,8 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						Action().Conjunction(12, 2, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchRegFieldWithValue(TargetOFPortField, uint32(2)).
-						Action().Conjunction(10, 2, 2).Done(),
+						Action().Conjunction(10, 2, 2).
+						Action().Conjunction(14, 2, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchRegFieldWithValue(TargetOFPortField, uint32(3)).
 						Action().Conjunction(11, 2, 3).Done(),
@@ -533,6 +557,9 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolTCP).MatchDstPort(8080, nil).
 						Action().Conjunction(11, 3, 3).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
+						MatchProtocol(binding.ProtocolTCP).MatchSrcPort(32800, nil).
+						Action().Conjunction(14, 3, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority200).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolTCP).MatchDstPort(8080, nil).
 						Action().Conjunction(12, 3, 3).Done(),
@@ -554,6 +581,9 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 					IngressMetricTable.ofTable.BuildFlow(priorityNormal).Cookie(cookiePolicy).
 						MatchRegMark(CnpDenyRegMark).MatchRegFieldWithValue(CNPConjIDField, 13).
 						Action().Drop().Done(),
+					IngressMetricTable.ofTable.BuildFlow(priorityNormal).Cookie(cookiePolicy).
+						MatchRegMark(CnpDenyRegMark).MatchRegFieldWithValue(CNPConjIDField, 14).
+						Action().Drop().Done(),
 					IngressDefaultTable.ofTable.BuildFlow(priority200).Cookie(cookiePolicy).
 						MatchTunnelID(uint64(UnknownLabelIdentity)).
 						MatchRegFieldWithValue(TargetOFPortField, uint32(1)).

test/e2e/antreapolicy_test.go

@@ -538,12 +538,6 @@ func testInvalidACNPPodSelectorNsSelectorMatchExpressions(t *testing.T, data *Te
 // testACNPAllowXBtoA tests traffic from X/B to pods with label A, after applying the default deny
 // k8s NetworkPolicies in all namespaces and ACNP to allow X/B to A.
 func testACNPAllowXBtoA(t *testing.T) {
-	portStart, portEnd, err := k8sUtils.getTCPv4SourcePortRangeFromPod(namespaces["x"], "a")
-	if err != nil {
-		log.Infof("ERROR in retrieving port range: %v", err)
-	} else {
-		log.Infof("SUCCESS in retrieving port range: %v - %v", portStart, portEnd)
-	}
 	builder := &ClusterNetworkPolicySpecBuilder{}
 	builder = builder.SetName("acnp-allow-xb-to-a").
 		SetPriority(1.0).
@@ -574,6 +568,42 @@ func testACNPAllowXBtoA(t *testing.T) {
 	executeTests(t, testCase)
 }
 
+// testACNPSourcePort tests ACNP source port filtering. The agnhost image used in E2E tests uses
+// ephemeral ports to initiate TCP connections, which should be 32768–60999 by default
+// (https://en.wikipedia.org/wiki/Ephemeral_port). This test retrieves the port range from
+// the client Pod and uses it in sourcePort and sourceEndPort of an ACNP rule to verify that
+// packets can be matched by source port.
+func testACNPSourcePort(t *testing.T) {
+	portStart, portEnd, err := k8sUtils.getTCPv4SourcePortRangeFromPod(namespaces["x"], "a")
+	failOnError(err, t)
+	builder := &ClusterNetworkPolicySpecBuilder{}
+	builder = builder.SetName("acnp-source-port").
+		SetPriority(1.0).
+		SetAppliedToGroup([]ACNPAppliedToSpec{{PodSelector: map[string]string{"pod": "a"}}})
+	builder.AddIngressForSrcPort(ProtocolTCP, &portStart, &portEnd, nil, nil, nil, nil, nil, map[string]string{"pod": "b"}, map[string]string{"ns": namespaces["x"]},
+		nil, nil, false, nil, crdv1alpha1.RuleActionDrop, "", "", nil)
+
+	reachability := NewReachability(allPods, Connected)
+	reachability.Expect(Pod(namespaces["x"]+"/b"), Pod(namespaces["x"]+"/a"), Dropped)
+	reachability.Expect(Pod(namespaces["x"]+"/b"), Pod(namespaces["y"]+"/a"), Dropped)
+	reachability.Expect(Pod(namespaces["x"]+"/b"), Pod(namespaces["z"]+"/a"), Dropped)
+	testStep := []*TestStep{
+		{
+			"Port 80",
+			reachability,
+			[]metav1.Object{builder.Get()},
+			[]int32{80},
+			ProtocolTCP,
+			0,
+			nil,
+		},
+	}
+	testCase := []*TestCase{
+		{"ACNP Drop X/B to A based on source port", testStep},
+	}
+	executeTests(t, testCase)
+}
+
 // testACNPAllowXBtoYA tests traffic from X/B to Y/A on named port 81, after applying the default deny
 // k8s NetworkPolicies in all namespaces and ACNP to allow X/B to Y/A.
 func testACNPAllowXBtoYA(t *testing.T) {
@@ -4306,6 +4336,7 @@ func TestAntreaPolicy(t *testing.T) {
 		t.Run("Case=ACNPPriorityConflictingRule", func(t *testing.T) { testACNPPriorityConflictingRule(t) })
 		t.Run("Case=ACNPRulePriority", func(t *testing.T) { testACNPRulePriority(t) })
 		t.Run("Case=ANPPortRange", func(t *testing.T) { testANPPortRange(t) })
+		t.Run("Case=ACNPSourcePort", func(t *testing.T) { testACNPSourcePort(t) })
 		t.Run("Case=ANPBasic", func(t *testing.T) { testANPBasic(t) })
 		t.Run("Case=testANPMultipleAppliedToSingleRule", func(t *testing.T) { testANPMultipleAppliedTo(t, data, true) })
 		t.Run("Case=testANPMultipleAppliedToMultipleRules", func(t *testing.T) { testANPMultipleAppliedTo(t, data, false) })

test/e2e/k8s_util.go

@@ -133,7 +133,7 @@ func (k *KubernetesUtils) LabelPod(ns, name, key, value string) (*v1.Pod, error)
 	return k.clientset.CoreV1().Pods(ns).Update(context.TODO(), pod, metav1.UpdateOptions{})
 }
 
-func (k *KubernetesUtils) getTCPv4SourcePortRangeFromPod(podNamespace, podNameLabel string) (int, int, error) {
+func (k *KubernetesUtils) getTCPv4SourcePortRangeFromPod(podNamespace, podNameLabel string) (int32, int32, error) {
 	pod, err := k.GetPodByLabel(podNamespace, podNameLabel)
 	if err != nil {
 		return 0, 0, err
@@ -153,9 +153,9 @@ func (k *KubernetesUtils) getTCPv4SourcePortRangeFromPod(podNamespace, podNameLa
 		log.Errorf("Failed to retrieve TCP source port range for Pod %s/%s", podNamespace, podNameLabel)
 		return 0, 0, err
 	}
-	startPort, _ := strconv.Atoi(ports[0])
-	endPort, _ := strconv.Atoi(ports[1])
-	return startPort, endPort, nil
+	startPort, _ := strconv.ParseInt(ports[0], 0, 32)
+	endPort, _ := strconv.ParseInt(strings.TrimSuffix(ports[1], "\n"), 0, 32)
+	return int32(startPort), int32(endPort), nil
 }
 
 func (k *KubernetesUtils) probe(

test/e2e/utils/anp_spec_builder.go

@@ -151,7 +151,7 @@ func (b *AntreaNetworkPolicySpecBuilder) AddIngress(protoc AntreaPolicyProtocol,
 			Group:                  ruleGroup,
 		}}
 	}
-	ports, protocols := GenPortsOrProtocols(protoc, port, portName, endPort, icmpType, icmpCode, igmpType, groupAddress)
+	ports, protocols := GenPortsOrProtocols(protoc, port, portName, endPort, nil, nil, icmpType, icmpCode, igmpType, groupAddress)
 	newRule := crdv1alpha1.Rule{
 		From:        policyPeer,
 		Ports:       ports,

test/e2e/utils/cnp_spec_builder.go

@@ -162,7 +162,76 @@ func (b *ClusterNetworkPolicySpecBuilder) AddIngress(protoc AntreaPolicyProtocol
 			ServiceAccount:    serviceAccount,
 		}}
 	}
-	ports, protocols := GenPortsOrProtocols(protoc, port, portName, endPort, icmpType, icmpCode, igmpType, groupAddress)
+	ports, protocols := GenPortsOrProtocols(protoc, port, portName, endPort, nil, nil, icmpType, icmpCode, igmpType, groupAddress)
+	newRule := crdv1alpha1.Rule{
+		From:      policyPeer,
+		Ports:     ports,
+		Protocols: protocols,
+		Action:    &action,
+		Name:      name,
+		AppliedTo: appliedTos,
+	}
+	b.Spec.Ingress = append(b.Spec.Ingress, newRule)
+	return b
+}
+
+// TODO: added new function to avoid merge conflicts. Unify this function with 'addIngress' when
+//   all conflicting PRs are merged.
+func (b *ClusterNetworkPolicySpecBuilder) AddIngressForSrcPort(protoc AntreaPolicyProtocol,
+	srcPort *int32, endPort, icmpType, icmpCode, igmpType *int32,
+	groupAddress, cidr *string, podSelector map[string]string, nsSelector map[string]string,
+	podSelectorMatchExp []metav1.LabelSelectorRequirement, nsSelectorMatchExp []metav1.LabelSelectorRequirement, selfNS bool,
+	ruleAppliedToSpecs []ACNPAppliedToSpec, action crdv1alpha1.RuleAction, ruleClusterGroup, name string, serviceAccount *crdv1alpha1.NamespacedName) *ClusterNetworkPolicySpecBuilder {
+
+	var pSel *metav1.LabelSelector
+	var nSel *metav1.LabelSelector
+	var ns *crdv1alpha1.PeerNamespaces
+	var appliedTos []crdv1alpha1.AppliedTo
+	matchSelf := crdv1alpha1.NamespaceMatchSelf
+
+	if b.Spec.Ingress == nil {
+		b.Spec.Ingress = []crdv1alpha1.Rule{}
+	}
+
+	if podSelector != nil || podSelectorMatchExp != nil {
+		pSel = &metav1.LabelSelector{
+			MatchLabels:      podSelector,
+			MatchExpressions: podSelectorMatchExp,
+		}
+	}
+	if nsSelector != nil || nsSelectorMatchExp != nil {
+		nSel = &metav1.LabelSelector{
+			MatchLabels:      nsSelector,
+			MatchExpressions: nsSelectorMatchExp,
+		}
+	}
+	if selfNS == true {
+		ns = &crdv1alpha1.PeerNamespaces{
+			Match: matchSelf,
+		}
+	}
+	var ipBlock *crdv1alpha1.IPBlock
+	if cidr != nil {
+		ipBlock = &crdv1alpha1.IPBlock{
+			CIDR: *cidr,
+		}
+	}
+	for _, at := range ruleAppliedToSpecs {
+		appliedTos = append(appliedTos, b.GetAppliedToPeer(at.PodSelector, at.NSSelector, at.PodSelectorMatchExp, at.NSSelectorMatchExp, at.Group, at.Service))
+	}
+	// An empty From/To in ACNP rules evaluates to match all addresses.
+	policyPeer := make([]crdv1alpha1.NetworkPolicyPeer, 0)
+	if pSel != nil || nSel != nil || ns != nil || ipBlock != nil || ruleClusterGroup != "" || serviceAccount != nil {
+		policyPeer = []crdv1alpha1.NetworkPolicyPeer{{
+			PodSelector:       pSel,
+			NamespaceSelector: nSel,
+			Namespaces:        ns,
+			IPBlock:           ipBlock,
+			Group:             ruleClusterGroup,
+			ServiceAccount:    serviceAccount,
+		}}
+	}
+	ports, protocols := GenPortsOrProtocols(protoc, nil, nil, nil, srcPort, endPort, icmpType, icmpCode, igmpType, groupAddress)
 	newRule := crdv1alpha1.Rule{
 		From:      policyPeer,
 		Ports:     ports,
@@ -232,7 +301,7 @@ func (b *ClusterNetworkPolicySpecBuilder) AddFQDNRule(fqdn string,
 		appliedTos = append(appliedTos, b.GetAppliedToPeer(at.PodSelector, at.NSSelector, at.PodSelectorMatchExp, at.NSSelectorMatchExp, at.Group, at.Service))
 	}
 	policyPeer := []crdv1alpha1.NetworkPolicyPeer{{FQDN: fqdn}}
-	ports, _ := GenPortsOrProtocols(protoc, port, portName, endPort, nil, nil, nil, nil)
+	ports, _ := GenPortsOrProtocols(protoc, port, portName, endPort, nil, nil, nil, nil, nil, nil)
 	newRule := crdv1alpha1.Rule{
 		To:        policyPeer,
 		Ports:     ports,

test/e2e/utils/helper.go

@@ -46,7 +46,7 @@ func AntreaPolicyProtocolToK8sProtocol(antreaProtocol AntreaPolicyProtocol) (v1.
 	}
 }
 
-func GenPortsOrProtocols(protoc AntreaPolicyProtocol, port *int32, portName *string, endPort, icmpType, icmpCode, igmpType *int32, groupAddress *string) ([]crdv1alpha1.NetworkPolicyPort, []crdv1alpha1.NetworkPolicyProtocol) {
+func GenPortsOrProtocols(protoc AntreaPolicyProtocol, port *int32, portName *string, endPort, srcPort, srcEndPort, icmpType, icmpCode, igmpType *int32, groupAddress *string) ([]crdv1alpha1.NetworkPolicyPort, []crdv1alpha1.NetworkPolicyProtocol) {
 	if protoc == ProtocolICMP {
 		return nil, []crdv1alpha1.NetworkPolicyProtocol{
 			{
@@ -80,16 +80,18 @@ func GenPortsOrProtocols(protoc AntreaPolicyProtocol, port *int32, portName *str
 			},
 		}
 	}
-	if port != nil || endPort != nil {
+	if port != nil || endPort != nil || srcPort != nil || srcEndPort != nil {
 		var pVal *intstr.IntOrString
 		if port != nil {
 			pVal = &intstr.IntOrString{IntVal: *port}
 		}
 		ports = []crdv1alpha1.NetworkPolicyPort{
 			{
-				Port:     pVal,
-				EndPort:  endPort,
-				Protocol: &k8sProtocol,
+				Port:          pVal,
+				EndPort:       endPort,
+				SourcePort:    srcPort,
+				SourceEndPort: srcEndPort,
+				Protocol:      &k8sProtocol,
 			},
 		}
 	}

test/e2e/vmagent_test.go

@@ -487,7 +487,7 @@ func createANPWithFQDN(t *testing.T, data *TestData, name string, namespace stri
 	for fqdn, action := range fqdnSettings {
 		ruleName := fmt.Sprintf("name-%d", i)
 		policyPeer := []crdv1alpha1.NetworkPolicyPeer{{FQDN: fqdn}}
-		ports, _ := GenPortsOrProtocols(ProtocolTCP, nil, nil, nil, nil, nil, nil, nil)
+		ports, _ := GenPortsOrProtocols(ProtocolTCP, nil, nil, nil, nil, nil, nil, nil, nil, nil)
 		newRule := crdv1alpha1.Rule{
 			To:     policyPeer,
 			Ports:  ports,