Merge pull request #459 from viveksb007/main

fix #456: add IPv6 catchAll entry implicitly if no egress rule and no egress isolation

Author: Pavani-Panakanti

controllers/policyendpoints_controller.go

@@ -39,10 +39,6 @@ import (
 	networking "k8s.io/api/networking/v1"
 )
 
-const (
-	defaultLocalConntrackCacheCleanupPeriodInSeconds = 300
-)
-
 func log() logger.Logger {
 	return logger.Get()
 }
@@ -84,11 +80,12 @@ func prometheusRegister() {
 }
 
 // NewPolicyEndpointsReconciler constructs new PolicyEndpointReconciler
-func NewPolicyEndpointsReconciler(k8sClient client.Client, nodeIP string, ebpfClient ebpf.BpfClient) *PolicyEndpointsReconciler {
+func NewPolicyEndpointsReconciler(k8sClient client.Client, nodeIP string, ebpfClient ebpf.BpfClient, enableIPv6 bool) *PolicyEndpointsReconciler {
 	r := &PolicyEndpointsReconciler{
 		k8sClient:  k8sClient,
 		nodeIP:     nodeIP,
 		ebpfClient: ebpfClient,
+		enableIPv6: enableIPv6,
 	}
 
 	prometheusRegister()
@@ -111,6 +108,7 @@ type PolicyEndpointsReconciler struct {
 	networkPolicyToPodIdentifierMap sync.Map
 	//BPF Client instance
 	ebpfClient ebpf.BpfClient
+	enableIPv6 bool
 }
 
 //+kubebuilder:rbac:groups=networking.k8s.aws,resources=policyendpoints,verbs=get;list;watch
@@ -244,13 +242,13 @@ func (r *PolicyEndpointsReconciler) reconcilePolicyEndpoint(ctx context.Context,
 		if len(ingressRules) == 0 && !isIngressIsolated {
 			//Add allow-all entry to Ingress rule set
 			log().Info("No Ingress rules and no ingress isolation - Appending catch all entry")
-			r.addCatchAllEntry(ctx, &ingressRules)
+			r.addCatchAllEntry(&ingressRules)
 		}
 
 		if len(egressRules) == 0 && !isEgressIsolated {
 			//Add allow-all entry to Egress rule set
 			log().Info("No Egress rules and no egress isolation - Appending catch all entry")
-			r.addCatchAllEntry(ctx, &egressRules)
+			r.addCatchAllEntry(&egressRules)
 		}
 
 		// Setup/configure eBPF probes/maps for local pods
@@ -339,14 +337,14 @@ func (r *PolicyEndpointsReconciler) cleanupPod(ctx context.Context, targetPod ty
 				// No active ingress rules for this pod, but we only should land here
 				// if there are active egress rules. So, we need to add an allow-all entry to ingress rule set
 				log().Info("No Ingress rules and no ingress isolation - Appending catch all entry")
-				r.addCatchAllEntry(ctx, &ingressRules)
+				r.addCatchAllEntry(&ingressRules)
 			}
 
 			if noActiveEgressPolicies {
 				// No active egress rules for this pod but we only should land here
 				// if there are active ingress rules. So, we need to add an allow-all entry to egress rule set
 				log().Info("No Egress rules and no egress isolation - Appending catch all entry")
-				r.addCatchAllEntry(ctx, &egressRules)
+				r.addCatchAllEntry(&egressRules)
 			}
 
 			err = r.updateeBPFMaps(ctx, podIdentifier, ingressRules, egressRules)
@@ -647,18 +645,18 @@ func (r *PolicyEndpointsReconciler) deletePolicyEndpointFromPodIdentifierMap(ctx
 	}
 }
 
-func (r *PolicyEndpointsReconciler) addCatchAllEntry(ctx context.Context, firewallRules *[]fwrp.EbpfFirewallRules) {
+func (r *PolicyEndpointsReconciler) addCatchAllEntry(firewallRules *[]fwrp.EbpfFirewallRules) {
 	//Add allow-all entry to firewall rule set
-	catchAllRule := policyk8sawsv1.EndpointInfo{
-		CIDR: "0.0.0.0/0",
+	var catchAllCIDR string
+	if r.enableIPv6 {
+		catchAllCIDR = "::/0"
+	} else {
+		catchAllCIDR = "0.0.0.0/0"
 	}
 	*firewallRules = append(*firewallRules,
 		fwrp.EbpfFirewallRules{
-			IPCidr: catchAllRule.CIDR,
-			L4Info: catchAllRule.Ports,
+			IPCidr: policyk8sawsv1.NetworkAddress(catchAllCIDR),
 		})
-
-	return
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -706,14 +704,14 @@ func (r *PolicyEndpointsReconciler) DeriveFireWallRulesPerPodIdentifier(podIdent
 		// No active ingress rules for this pod, but we only should land here
 		// if there are active egress rules. So, we need to add an allow-all entry to ingress rule set
 		log().Info("No Ingress rules and no ingress isolation - Appending catch all entry")
-		r.addCatchAllEntry(context.Background(), &ingressRules)
+		r.addCatchAllEntry(&ingressRules)
 	}
 
 	if len(egressRules) == 0 && !isEgressIsolated {
 		// No active egress rules for this pod but we only should land here
 		// if there are active ingress rules. So, we need to add an allow-all entry to egress rule set
 		log().Info("No Egress rules and no egress isolation - Appending catch all entry")
-		r.addCatchAllEntry(context.Background(), &egressRules)
+		r.addCatchAllEntry(&egressRules)
 	}
 
 	return ingressRules, egressRules, nil

controllers/policyendpoints_controller_test.go

@@ -44,7 +44,7 @@ func TestPolicyEndpointReconcile(t *testing.T) {
 
 	t.Run("Reconcile call for Create PolicyEndpoint with PodEndpoint local to Node", func(t *testing.T) {
 		mockClient := mock_client.NewMockClient(ctrl)
-		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, nodeIp, &ebpf.MockBpfClient{})
+		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, nodeIp, &ebpf.MockBpfClient{}, false)
 
 		policyEndpoint := getPolicyEndpoint("allow-all-egress", "my-namespace", []policyendpoint.PodEndpoint{p1N1, p2N1})
 
@@ -90,7 +90,7 @@ func TestPolicyEndpointReconcile(t *testing.T) {
 
 	t.Run("Reconcile for Create and Delete PE", func(t *testing.T) {
 		mockClient := mock_client.NewMockClient(ctrl)
-		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, nodeIp, &ebpf.MockBpfClient{})
+		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, nodeIp, &ebpf.MockBpfClient{}, false)
 
 		policyEndpoint := getPolicyEndpoint("allow-all-egress", "my-namespace", []policyendpoint.PodEndpoint{p1N1, p2N1})
 
@@ -500,7 +500,7 @@ func TestDeriveIngressAndEgressFirewallRules(t *testing.T) {
 		defer ctrl.Finish()
 
 		mockClient := mock_client.NewMockClient(ctrl)
-		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil)
+		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil, false)
 		var policyEndpointsList []string
 		policyEndpointsList = append(policyEndpointsList, tt.policyEndpointName)
 		policyEndpointReconciler.podIdentifierToPolicyEndpointMap.Store(tt.podIdentifier, policyEndpointsList)
@@ -757,8 +757,7 @@ func TestAddCatchAllEntry(t *testing.T) {
 		}
 
 		t.Run(tt.name, func(t *testing.T) {
-			policyEndpointReconciler.addCatchAllEntry(context.Background(),
-				&tt.firewallRules)
+			policyEndpointReconciler.addCatchAllEntry(&tt.firewallRules)
 			assert.Equal(t, tt.want, sampleFirewallRulesWithCatchAllEntry)
 		})
 	}
@@ -915,7 +914,7 @@ func TestArePoliciesAvailableInLocalCache(t *testing.T) {
 		defer ctrl.Finish()
 
 		mockClient := mock_client.NewMockClient(ctrl)
-		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil)
+		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil, false)
 		var policyEndpointsList []string
 		policyEndpointsList = append(policyEndpointsList, tt.policyEndpointName...)
 		policyEndpointReconciler.podIdentifierToPolicyEndpointMap.Store(tt.podIdentifier, policyEndpointsList)
@@ -1160,7 +1159,7 @@ func TestDeriveFireWallRulesPerPodIdentifier(t *testing.T) {
 		defer ctrl.Finish()
 
 		mockClient := mock_client.NewMockClient(ctrl)
-		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil)
+		policyEndpointReconciler := NewPolicyEndpointsReconciler(mockClient, "", nil, false)
 		var policyEndpointsList []string
 		policyEndpointsList = append(policyEndpointsList, tt.policyEndpointName)
 		policyEndpointReconciler.podIdentifierToPolicyEndpointMap.Store(tt.podIdentifier, policyEndpointsList)

main.go

@@ -111,7 +111,7 @@ func main() {
 			ctrlConfig.EnableIPv6, ctrlConfig.ConntrackCacheCleanupPeriod, ctrlConfig.ConntrackCacheTableSize, npMode, isMultiNICEnabled))
 		ebpfClient.ReAttachEbpfProbes()
 
-		policyEndpointController = controllers.NewPolicyEndpointsReconciler(mgr.GetClient(), nodeIP, ebpfClient)
+		policyEndpointController = controllers.NewPolicyEndpointsReconciler(mgr.GetClient(), nodeIP, ebpfClient, ctrlConfig.EnableIPv6)
 
 		if err = policyEndpointController.SetupWithManager(ctx, mgr); err != nil {
 			log.Errorf("unable to create controller PolicyEndpoints %v", err)

pkg/fwruleprocessor/fw_rule_processor_test.go

@@ -11,13 +11,10 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-func TestFWRuleProcessor_ComputeMapEntriesFromEndpointRules(t *testing.T) {
+func TestFWRuleProcessor_ComputeMapEntriesFromEndpointRules_IPv4(t *testing.T) {
 	protocolTCP := corev1.ProtocolTCP
-	//protocolUDP := corev1.ProtocolUDP
-	//protocolSCTP := corev1.ProtocolSCTP
 
 	var testIP v1alpha1.NetworkAddress
-	var gotKeys []string
 
 	nodeIP := "10.1.1.1"
 	_, nodeIPCIDR, _ := net.ParseCIDR(nodeIP + "/32")
@@ -28,6 +25,8 @@ func TestFWRuleProcessor_ComputeMapEntriesFromEndpointRules(t *testing.T) {
 	testIP = "10.1.1.2/32"
 	_, testIPCIDR, _ := net.ParseCIDR(string(testIP))
 
+	_, catchAllCIDR, _ := net.ParseCIDR(string("0.0.0.0/0"))
+
 	testIPKey := utils.ComputeTrieKey(*testIPCIDR, false)
 	type args struct {
 		firewallRules []EbpfFirewallRules
@@ -56,11 +55,106 @@ func TestFWRuleProcessor_ComputeMapEntriesFromEndpointRules(t *testing.T) {
 			},
 			want: []string{string(nodeIPKey), string(testIPKey)},
 		},
+		{
+			name: "CatchAll CIDR",
+			args: args{
+				[]EbpfFirewallRules{
+					{
+						IPCidr: "0.0.0.0/0",
+					},
+				},
+			},
+			want: []string{string(nodeIPKey), string(utils.ComputeTrieKey(*catchAllCIDR, false))},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := NewFirewallRuleProcessor(nodeIP, "/32", false).ComputeMapEntriesFromEndpointRules(tt.args.firewallRules)
+			var gotKeys []string
+			if tt.wantErr != nil {
+				assert.EqualError(t, err, tt.wantErr.Error())
+			} else {
+				for key, _ := range got {
+					gotKeys = append(gotKeys, key)
+				}
+				sort.Strings(tt.want)
+				sort.Strings(gotKeys)
+				assert.Equal(t, tt.want, gotKeys)
+			}
+		})
+	}
+}
+
+func TestFWRuleProcessor_ComputeMapEntriesFromEndpointRules_IPv6(t *testing.T) {
+	protocolTCP := corev1.ProtocolTCP
+
+	nodeIP := "2001:db8:abcd:0012::1"
+	_, nodeIPCIDR, _ := net.ParseCIDR(nodeIP + "/128")
+	nodeIPKey := utils.ComputeTrieKey(*nodeIPCIDR, true)
+
+	var testPort int32
+	testPort = 80
+	_, testIPCIDR, _ := net.ParseCIDR("2001:db8:abcd:0012::10/128")
+
+	_, catchAllCIDR, _ := net.ParseCIDR("::/0")
+
+	testIPKey := utils.ComputeTrieKey(*testIPCIDR, true)
+	type args struct {
+		firewallRules []EbpfFirewallRules
+	}
+
+	tests := []struct {
+		name    string
+		args    args
+		want    []string
+		wantErr error
+	}{
+		{
+			name: "CIDR with Port and Protocol",
+			args: args{
+				[]EbpfFirewallRules{
+					{
+						IPCidr: "2001:db8:abcd:0012::10/128",
+						L4Info: []v1alpha1.Port{
+							{
+								Protocol: &protocolTCP,
+								Port:     &testPort,
+							},
+						},
+					},
+				},
+			},
+			want: []string{string(nodeIPKey), string(testIPKey)},
+		},
+		{
+			name: "CatchAll IPv4 CIDR",
+			args: args{
+				[]EbpfFirewallRules{
+					{
+						IPCidr: "::/0",
+					},
+				},
+			},
+			want: []string{string(nodeIPKey), string(utils.ComputeTrieKey(*catchAllCIDR, true))},
+		},
+		{
+			name: "CatchAll CIDR IPv4 ignored in rule computation",
+			args: args{
+				[]EbpfFirewallRules{
+					{
+						IPCidr: "0.0.0.0/0",
+					},
+				},
+			},
+			want: []string{string(nodeIPKey)},
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := NewFirewallRuleProcessor("10.1.1.1", "/32", false).ComputeMapEntriesFromEndpointRules(tt.args.firewallRules)
+			got, err := NewFirewallRuleProcessor(nodeIP, "/128", true).ComputeMapEntriesFromEndpointRules(tt.args.firewallRules)
+			var gotKeys []string
 			if tt.wantErr != nil {
 				assert.EqualError(t, err, tt.wantErr.Error())
 			} else {