Merge pull request #580 from seswarrajan/dev

K8s Net pol -- UDP handling

src/conf/local-file.yaml

@@ -2,10 +2,10 @@ application:
   name: knoxautopolicy
   network:
     operation-mode: 1                             # 1: cronjob | 2: one-time-job
-    operation-trigger: 5
+    operation-trigger: 100
     cron-job-time-interval: "0h0m10s"             # format: XhYmZs 
     network-log-limit: 10000
-    network-log-from: "kubearmor"                 # db|hubble|feed-consumer|kubearmor
+    network-log-from: "hubble"                    # db|hubble|feed-consumer|kubearmor
     #network-log-file: "/home/rahul/feeds.json"   # file path
     network-policy-to: "db"                       # db, file
     network-policy-dir: "./"

src/networkpolicy/deduplicator.go

@@ -974,13 +974,15 @@ func UpdateDuplicatedPolicy(existingPolicies []types.KnoxNetworkPolicy, discover
 
 	for _, policy := range existIngressPolicies {
 		if policy.Metadata["status"] == "updated" {
-			delete(policy.Metadata, "status")
+			policy.Metadata["status"] = "latest"
+			//delete(policy.Metadata, "status")
 			updatedPolicies = append(updatedPolicies, policy)
 		}
 	}
 	for _, policy := range existEgressPolicies {
 		if policy.Metadata["status"] == "updated" {
-			delete(policy.Metadata, "status")
+			policy.Metadata["status"] = "latest"
+			//delete(policy.Metadata, "status")
 			updatedPolicies = append(updatedPolicies, policy)
 		}
 	}

src/networkpolicy/helperFunctions.go

@@ -13,6 +13,7 @@ import (
 	"github.com/clarketm/json"
 
 	"github.com/accuknox/auto-policy-discovery/src/cluster"
+	"github.com/accuknox/auto-policy-discovery/src/config"
 	"github.com/accuknox/auto-policy-discovery/src/libs"
 	"github.com/accuknox/auto-policy-discovery/src/plugin"
 	wpb "github.com/accuknox/auto-policy-discovery/src/protobuf/v1/worker"
@@ -797,7 +798,8 @@ func GetNetPolicy(cluster, namespace, policyType string) *wpb.WorkerResponse {
 		}
 		response.K8SNetworkpolicy = nil
 	} else if strings.Contains(policyType, "generic") {
-		policies := plugin.ConvertKnoxNetPolicyToK8sNetworkPolicy(cluster, namespace)
+		knoxNetPolicies := libs.GetNetworkPolicies(config.CurrentCfg.ConfigDB, cluster, namespace, "latest", "", "")
+		policies := plugin.ConvertKnoxNetPolicyToK8sNetworkPolicy(cluster, namespace, knoxNetPolicies)
 
 		for i := range policies {
 			genericNetPol := wpb.Policy{}

src/networkpolicy/networkPolicy.go

@@ -1554,6 +1554,38 @@ func mergeNetworkPolicies(existPolicy types.KnoxNetworkPolicy, policies []types.
 	return mergeEgressPolicies(existPolicy, policies)
 }
 
+func checkIfIngressEgressPortExist(polType string, newToPort types.SpecPort, mergedPolicy types.KnoxNetworkPolicy) bool {
+	isExist := false
+
+	if polType == "EGRESS" {
+		for _, existEgress := range mergedPolicy.Spec.Egress {
+			if len(existEgress.ToPorts) == 0 {
+				continue
+			}
+			existToPort := existEgress.ToPorts[0]
+
+			if existToPort == newToPort {
+				isExist = true
+				break
+			}
+		}
+	} else if polType == "INGRESS" {
+		for _, existIngress := range mergedPolicy.Spec.Ingress {
+			if len(existIngress.ToPorts) == 0 {
+				continue
+			}
+			existToPort := existIngress.ToPorts[0]
+
+			if existToPort == newToPort {
+				isExist = true
+				break
+			}
+		}
+	}
+
+	return isExist
+}
+
 func mergeIngressPolicies(existPolicy types.KnoxNetworkPolicy, policies []types.KnoxNetworkPolicy) (types.KnoxNetworkPolicy, bool) {
 	mergedPolicy := existPolicy
 	updated := false
@@ -1598,13 +1630,18 @@ func mergeIngressPolicies(existPolicy types.KnoxNetworkPolicy, policies []types.
 						}
 					}
 				}
+			} else if len(newIngress.FromCIDRs) > 0 && len(newIngress.ToPorts) > 0 {
+				newToPort := newIngress.ToPorts[0]
+				ingressMatched = checkIfIngressEgressPortExist("INGRESS", newToPort, mergedPolicy)
 			}
+
 			if !ingressMatched {
 				mergedPolicy.Spec.Ingress = append(mergedPolicy.Spec.Ingress, newIngress)
 				updated = true
 			}
 		}
 	}
+
 	return mergedPolicy, updated
 }
 
@@ -1668,7 +1705,11 @@ func mergeEgressPolicies(existPolicy types.KnoxNetworkPolicy, policies []types.K
 						}
 					}
 				}
+			} else if len(newEgress.ToCIDRs) > 0 && len(newEgress.ToPorts) > 0 {
+				newToPort := newEgress.ToPorts[0]
+				egressMatched = checkIfIngressEgressPortExist("EGRESS", newToPort, mergedPolicy)
 			}
+
 			if !egressMatched {
 				mergedPolicy.Spec.Egress = append(mergedPolicy.Spec.Egress, newEgress)
 				updated = true
@@ -1726,6 +1767,44 @@ func mergeHttpRules(existRule types.L47Rule, newRule types.L47Rule) (bool, bool,
 	return false, false, nil
 }
 
+func populateIngressEgressPolicyFromKnoxNetLog(log *types.KnoxNetworkLog, pods []types.Pod) types.KnoxNetworkPolicy {
+	iePolicy := types.KnoxNetworkPolicy{}
+	var cidrs []string
+	cidrs = append(cidrs, "0.0.0.0/32")
+
+	specVal := types.SpecCIDR{
+		CIDRs: cidrs,
+	}
+
+	portVal := types.SpecPort{
+		Port:     strconv.Itoa(log.DstPort),
+		Protocol: libs.GetProtocol(log.Protocol),
+	}
+
+	if log.Direction == "EGRESS" {
+		iePolicy = buildNewKnoxEgressPolicy()
+		egress := types.Egress{}
+
+		egress.ToPorts = append(egress.ToPorts, portVal)
+		egress.ToCIDRs = append(egress.ToCIDRs, specVal)
+
+		iePolicy.Spec.Egress = append(iePolicy.Spec.Egress, egress)
+
+	} else if log.Direction == "INGRESS" {
+		iePolicy = buildNewKnoxIngressPolicy()
+		ingress := types.Ingress{}
+
+		ingress.ToPorts = append(ingress.ToPorts, portVal)
+		ingress.FromCIDRs = append(ingress.FromCIDRs, specVal)
+
+		iePolicy.Spec.Ingress = append(iePolicy.Spec.Ingress, ingress)
+	}
+
+	iePolicy.Spec.Selector.MatchLabels = getEndpointMatchLabels(log.SrcPodName, pods)
+
+	return iePolicy
+}
+
 func convertKnoxNetworkLogToKnoxNetworkPolicy(log *types.KnoxNetworkLog, pods []types.Pod) (_, _ *types.KnoxNetworkPolicy) {
 	var ingressPolicy, egressPolicy *types.KnoxNetworkPolicy = nil, nil
 
@@ -1858,6 +1937,13 @@ func convertKnoxNetworkLogToKnoxNetworkPolicy(log *types.KnoxNetworkLog, pods []
 			ePolicy.Metadata["namespace"] = log.SrcNamespace
 			egressPolicy = &ePolicy
 		}
+	} else if log.DstPodName == "" && len(log.DstReservedLabels) == 0 {
+		iePolicy := populateIngressEgressPolicyFromKnoxNetLog(log, pods)
+		if log.Direction == "EGRESS" {
+			egressPolicy = &iePolicy
+		} else if log.Direction == "INGRESS" {
+			ingressPolicy = &iePolicy
+		}
 	}
 
 	if !isValidPolicy(ingressPolicy) {
@@ -2134,47 +2220,79 @@ func PopulateNetworkPoliciesFromNetworkLogs(networkLogs []types.KnoxNetworkLog)
 
 func writeNetworkPoliciesYamlToDB(policies []types.KnoxNetworkPolicy) {
 	clusters := []string{}
+	res := []types.PolicyYaml{}
 
 	for _, pol := range policies {
 		clusters = append(clusters, pol.Metadata["cluster_name"])
 	}
 
-	// convert knoxPolicy to CiliumPolicy
-	ciliumPolicies := plugin.ConvertKnoxPoliciesToCiliumPolicies(policies)
+	if cfg.CurrentCfg.ConfigNetPolicy.NetworkLogFrom == "kubearmor" {
+		k8sNetPolicies := plugin.ConvertKnoxNetPolicyToK8sNetworkPolicy("", "", policies)
 
-	res := []types.PolicyYaml{}
+		for i, np := range k8sNetPolicies {
+			np.ClusterName = ""
+			jsonBytes, err := json.Marshal(np)
+			if err != nil {
+				log.Error().Msg(err.Error())
+				continue
+			}
+			yamlBytes, err := yaml.JSONToYAML(jsonBytes)
+			if err != nil {
+				log.Error().Msg(err.Error())
+				continue
+			}
 
-	for i, ciliumPolicy := range ciliumPolicies {
-		jsonBytes, err := json.Marshal(ciliumPolicy)
-		if err != nil {
-			log.Error().Msg(err.Error())
-			continue
-		}
-		yamlBytes, err := yaml.JSONToYAML(jsonBytes)
-		if err != nil {
-			log.Error().Msg(err.Error())
-			continue
-		}
+			policyYaml := types.PolicyYaml{
+				Type:      types.PolicyTypeNetwork,
+				Kind:      np.Kind,
+				Name:      np.Name,
+				Namespace: np.Namespace,
+				Cluster:   clusters[i],
+				Labels:    np.Labels,
+				Yaml:      yamlBytes,
+			}
+			res = append(res, policyYaml)
 
-		var labels types.LabelMap
-		if ciliumPolicy.Kind == cu.ResourceTypeCiliumNetworkPolicy {
-			labels = ciliumPolicy.Spec.EndpointSelector.MatchLabels
-		} else {
-			labels = ciliumPolicy.Spec.NodeSelector.MatchLabels
+			PolicyStore.Publish(&policyYaml)
 		}
 
-		policyYaml := types.PolicyYaml{
-			Type:      types.PolicyTypeNetwork,
-			Kind:      ciliumPolicy.Kind,
-			Name:      ciliumPolicy.Metadata["name"],
-			Namespace: ciliumPolicy.Metadata["namespace"],
-			Cluster:   clusters[i],
-			Labels:    labels,
-			Yaml:      yamlBytes,
-		}
-		res = append(res, policyYaml)
+	} else {
 
-		PolicyStore.Publish(&policyYaml)
+		// convert knoxPolicy to CiliumPolicy
+		ciliumPolicies := plugin.ConvertKnoxPoliciesToCiliumPolicies(policies)
+
+		for i, ciliumPolicy := range ciliumPolicies {
+			jsonBytes, err := json.Marshal(ciliumPolicy)
+			if err != nil {
+				log.Error().Msg(err.Error())
+				continue
+			}
+			yamlBytes, err := yaml.JSONToYAML(jsonBytes)
+			if err != nil {
+				log.Error().Msg(err.Error())
+				continue
+			}
+
+			var labels types.LabelMap
+			if ciliumPolicy.Kind == cu.ResourceTypeCiliumNetworkPolicy {
+				labels = ciliumPolicy.Spec.EndpointSelector.MatchLabels
+			} else {
+				labels = ciliumPolicy.Spec.NodeSelector.MatchLabels
+			}
+
+			policyYaml := types.PolicyYaml{
+				Type:      types.PolicyTypeNetwork,
+				Kind:      ciliumPolicy.Kind,
+				Name:      ciliumPolicy.Metadata["name"],
+				Namespace: ciliumPolicy.Metadata["namespace"],
+				Cluster:   clusters[i],
+				Labels:    labels,
+				Yaml:      yamlBytes,
+			}
+			res = append(res, policyYaml)
+
+			PolicyStore.Publish(&policyYaml)
+		}
 	}
 
 	if err := libs.UpdateOrInsertPolicyYamls(CfgDB, res); err != nil {