APIs for providing logs and collecting policies

1. New protobuf definition
2. GRPC server changes for new rpcs

src/analyzer/analyzerTest.go

@@ -0,0 +1,89 @@
+package analyzer
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	apb "github.com/accuknox/knoxAutoPolicy/src/protobuf/v1/analyzer"
+	"google.golang.org/grpc"
+)
+
+var (
+	client apb.AnalyzerClient
+)
+
+func analyzerSystemTest() {
+	time.Sleep(10 * time.Second)
+	fmt.Println("START ANALYZER SYSTEM POLICY TEST")
+	systemLogs := apb.SystemLogs{}
+	systemTestLogData1 := apb.KnoxSystemLog{}
+	systemTestLogData2 := apb.KnoxSystemLog{}
+
+	// Test data
+	systemTestLogData1.LogID = 100
+	systemTestLogData1.ClusterName = "Test Cluster 001"
+	systemTestLogData1.HostName = "host-001"
+	systemTestLogData1.Namespace = "linux-001"
+	systemTestLogData1.PodName = "MyPod-001"
+	systemTestLogData1.SourceOrigin = "origin-001"
+	systemTestLogData1.Source = "source-001"
+	systemTestLogData1.Operation = "operation-001"
+	systemTestLogData1.ResourceOrigin = "resource-origin-001"
+	systemTestLogData1.Resource = "resource-001"
+	systemTestLogData1.Data = "data-001"
+	systemTestLogData1.ReadOnly = true
+	systemTestLogData1.Result = "result-001"
+	systemLogs.SysLog = append(systemLogs.SysLog, &systemTestLogData1)
+
+	systemTestLogData2.LogID = 200
+	systemTestLogData2.ClusterName = "Test Cluster 002"
+	systemTestLogData2.HostName = "host-002"
+	systemTestLogData2.Namespace = "linux-002"
+	systemTestLogData2.PodName = "MyPod-002"
+	systemTestLogData2.SourceOrigin = "origin-002"
+	systemTestLogData2.Source = "source-002"
+	systemTestLogData2.Operation = "operation-002"
+	systemTestLogData2.ResourceOrigin = "resource-origin-002"
+	systemTestLogData2.Resource = "resource-002"
+	systemTestLogData2.Data = "data-002"
+	systemTestLogData2.ReadOnly = false
+	systemTestLogData2.Result = "result-002"
+	systemLogs.SysLog = append(systemLogs.SysLog, &systemTestLogData2)
+
+	response, err := client.GetSystemPolicies(context.Background(), &systemLogs)
+	if err != nil {
+		log.Fatal("Error")
+	} else {
+		log.Printf("Response : %v\n", response)
+	}
+}
+
+func analyzerNetworkTest() {
+	time.Sleep(20 * time.Second)
+	fmt.Println("START ANALYZER NETWORK POLICY TEST")
+}
+
+func StartAnalyzerTest() {
+	time.Sleep(10 * time.Second)
+	fmt.Println("START ANALYZER TEST")
+
+	ctx, _ := context.WithTimeout(context.Background(), 30*time.Second)
+	grpcClientConn, err := grpc.DialContext(ctx, ":9089", grpc.WithInsecure(), grpc.WithBlock())
+	if err != nil {
+		log.Println("gRPC Dial failed")
+		log.Fatal(err)
+	} else {
+		log.Println("CONNECTION OK")
+	}
+
+	client = apb.NewAnalyzerClient(grpcClientConn)
+	if client == nil {
+		log.Fatal("invalid client handle")
+	} else {
+		log.Println("CLIENT NOT NIL")
+	}
+	go analyzerSystemTest()
+	go analyzerNetworkTest()
+}

src/analyzer/networkPolicyAnalyzer.go

@@ -0,0 +1,71 @@
+package analyzer
+
+import (
+	"encoding/json"
+	"log"
+
+	netpolicy "github.com/accuknox/knoxAutoPolicy/src/networkpolicy"
+	apb "github.com/accuknox/knoxAutoPolicy/src/protobuf/v1/analyzer"
+	types "github.com/accuknox/knoxAutoPolicy/src/types"
+)
+
+func extractNetworkPoliciesFromNetworkLogs(networkLogs []types.KnoxNetworkLog) []*apb.KnoxNetworkPolicy {
+
+	pbNetPolicies := []*apb.KnoxNetworkPolicy{}
+	netPolicies := netpolicy.PopulateNetworkPoliciesFromNetworkLogs(networkLogs)
+
+	for _, netPolicy := range netPolicies {
+		pbNetPolicy := apb.KnoxNetworkPolicy{}
+		pbNetPolicyBytes, err := json.Marshal(netPolicy)
+		if err != nil {
+			log.Printf("Failed to marshall : %v\n", err)
+		} else {
+			pbNetPolicy.NetworkPolicy = pbNetPolicyBytes
+			pbNetPolicies = append(pbNetPolicies, &pbNetPolicy)
+		}
+	}
+
+	return pbNetPolicies
+}
+
+func populateNetworkLogs(pbNetworkLog []*apb.KnoxNetworkLog) []types.KnoxNetworkLog {
+	networkLogs := []types.KnoxNetworkLog{}
+
+	// Populate KnoxNetworkLog from Protobuf's NetworkLog
+	for _, pbNetLog := range pbNetworkLog {
+		netLog := types.KnoxNetworkLog{}
+		netLog.FlowID = int(pbNetLog.FlowID)
+		netLog.ClusterName = pbNetLog.ClusterName
+		netLog.SrcNamespace = pbNetLog.SrcNamespace
+		netLog.SrcPodName = pbNetLog.SrcPodName
+		netLog.DstNamespace = pbNetLog.DstNamespace
+		netLog.DstPodName = pbNetLog.DstPodName
+		netLog.EtherType = int(pbNetLog.EtherType)
+		netLog.Protocol = int(pbNetLog.Protocol)
+		netLog.SrcIP = pbNetLog.SrcIP
+		netLog.DstIP = pbNetLog.DstIP
+		netLog.SrcPort = int(pbNetLog.SrcPort)
+		netLog.DstPort = int(pbNetLog.DstPort)
+		netLog.SynFlag = pbNetLog.SynFlag
+		netLog.IsReply = pbNetLog.IsReply
+		netLog.DNSQuery = pbNetLog.DNSQuery
+		netLog.DNSRes = pbNetLog.DNSRes
+		netLog.DNSResIPs = append(netLog.DNSResIPs, pbNetLog.DNSResIPs...)
+		netLog.HTTPMethod = pbNetLog.HTTPMethod
+		netLog.HTTPPath = pbNetLog.HTTPPath
+		netLog.Direction = pbNetLog.Direction
+		netLog.Action = pbNetLog.Action
+
+		networkLogs = append(networkLogs, netLog)
+	}
+
+	return networkLogs
+}
+
+func GetNetworkPolicies(pbNetworkLog []*apb.KnoxNetworkLog) []*apb.KnoxNetworkPolicy {
+
+	networkLogs := populateNetworkLogs(pbNetworkLog)
+	networkPolicies := extractNetworkPoliciesFromNetworkLogs(networkLogs)
+
+	return networkPolicies
+}

src/analyzer/systemPolicyAnalyzer.go

@@ -0,0 +1,64 @@
+package analyzer
+
+import (
+	"encoding/json"
+
+	apb "github.com/accuknox/knoxAutoPolicy/src/protobuf/v1/analyzer"
+	syspolicy "github.com/accuknox/knoxAutoPolicy/src/systempolicy"
+	types "github.com/accuknox/knoxAutoPolicy/src/types"
+	"github.com/rs/zerolog/log"
+)
+
+func extractSystemPoliciesFromSystemLogs(systemLogs []types.KnoxSystemLog) []*apb.KnoxSystemPolicy {
+
+	pbSystemPolicies := []*apb.KnoxSystemPolicy{}
+	systemPolicies := syspolicy.PopulateSystemPoliciesFromSystemLogs(systemLogs)
+
+	for _, sysPolicy := range systemPolicies {
+		pbSysPolicy := apb.KnoxSystemPolicy{}
+		pbSysPolicyBytes, err := json.Marshal(sysPolicy)
+		if err != nil {
+			log.Printf("Failed to marshall : %v\n", err)
+		} else {
+			pbSysPolicy.SystemPolicy = pbSysPolicyBytes
+			pbSystemPolicies = append(pbSystemPolicies, &pbSysPolicy)
+		}
+	}
+
+	return pbSystemPolicies
+}
+
+func populateSystemLogs(pbSysLogs []*apb.KnoxSystemLog) []types.KnoxSystemLog {
+	sysLogs := []types.KnoxSystemLog{}
+
+	// Populate KnoxSystemLog from Protobuf's SystemLog
+	for _, pbSysLog := range pbSysLogs {
+		sysLog := types.KnoxSystemLog{}
+		sysLog.LogID = int(pbSysLog.LogID)
+		sysLog.ClusterName = pbSysLog.ClusterName
+		sysLog.HostName = pbSysLog.HostName
+		sysLog.Namespace = pbSysLog.Namespace
+		sysLog.PodName = pbSysLog.PodName
+		sysLog.SourceOrigin = pbSysLog.SourceOrigin
+		sysLog.Source = pbSysLog.Source
+		sysLog.Operation = pbSysLog.Operation
+		sysLog.ResourceOrigin = pbSysLog.ResourceOrigin
+		sysLog.Resource = pbSysLog.Resource
+		sysLog.Data = pbSysLog.Data
+		sysLog.ReadOnly = pbSysLog.ReadOnly
+		sysLog.Result = pbSysLog.Result
+
+		sysLogs = append(sysLogs, sysLog)
+	}
+
+	log.Printf("\nsysLogs : %v\n", sysLogs)
+	return sysLogs
+}
+
+func GetSystemPolicies(pbSystemLogs []*apb.KnoxSystemLog) []*apb.KnoxSystemPolicy {
+
+	systemLogs := populateSystemLogs(pbSystemLogs)
+	systemPolicies := extractSystemPoliciesFromSystemLogs(systemLogs)
+
+	return systemPolicies
+}

src/networkpolicy/networkPolicy.go

@@ -1515,28 +1515,12 @@ func DiscoverNetworkPolicy(namespace string,
 	return networkPolicies
 }
 
-func DiscoverNetworkPolicyMain() {
-	if NetworkWorkerStatus == STATUS_RUNNING {
-		return
-	} else {
-		NetworkWorkerStatus = STATUS_RUNNING
-	}
+func PopulateNetworkPoliciesFromNetworkLogs(sysLogs []types.KnoxNetworkLog) []types.KnoxNetworkPolicy {
 
-	defer func() {
-		NetworkWorkerStatus = STATUS_IDLE
-	}()
-
-	// init the configuration related to the network policy
-	InitNetPolicyDiscoveryConfiguration()
-
-	// get network logs
-	allNetworkLogs := getNetworkLogs()
-	if allNetworkLogs == nil || len(allNetworkLogs) < OperationTrigger {
-		return
-	}
+	discoveredNetworkPolicies := []types.KnoxNetworkPolicy{}
 
 	// get cluster names, iterate each cluster
-	clusteredLogs := clusteringNetworkLogs(allNetworkLogs)
+	clusteredLogs := clusteringNetworkLogs(sysLogs)
 	for clusterName, networkLogs := range clusteredLogs {
 		log.Info().Msgf("Network policy discovery started for cluster [%s]", clusterName)
 
@@ -1572,6 +1556,7 @@ func DiscoverNetworkPolicyMain() {
 
 			// discover network policies based on the network logs
 			discoveredNetPolicies := DiscoverNetworkPolicy(namespace, logsPerNamespace, services, endpoints, pods)
+			discoveredNetworkPolicies = append(discoveredNetworkPolicies, discoveredNetPolicies...)
 
 			// get existing network policies in db
 			existingNetPolicies := libs.GetNetworkPolicies(CfgDB, clusterName, namespace, "latest")
@@ -1597,6 +1582,30 @@ func DiscoverNetworkPolicyMain() {
 		// update cluster global variables
 		updateMultiClusterVariables(clusterName)
 	}
+
+	return discoveredNetworkPolicies
+}
+
+func DiscoverNetworkPolicyMain() {
+	if NetworkWorkerStatus == STATUS_RUNNING {
+		return
+	} else {
+		NetworkWorkerStatus = STATUS_RUNNING
+	}
+
+	defer func() {
+		NetworkWorkerStatus = STATUS_IDLE
+	}()
+
+	// init the configuration related to the network policy
+	InitNetPolicyDiscoveryConfiguration()
+
+	// get network logs
+	allNetworkLogs := getNetworkLogs()
+	if allNetworkLogs == nil || len(allNetworkLogs) < OperationTrigger {
+		return
+	}
+
 }
 
 // ===================================== //

src/protobuf/Makefile

@@ -5,4 +5,5 @@ build:
 	protoc -I=. --go_out . --go_opt paths=source_relative --go-grpc_out . --go-grpc_opt paths=source_relative v1/config/config.proto
 	protoc -I=. --go_out . --go_opt paths=source_relative --go-grpc_out . --go-grpc_opt paths=source_relative v1/worker/worker.proto
 	protoc -I=. --go_out . --go_opt paths=source_relative --go-grpc_out . --go-grpc_opt paths=source_relative v1/consumer/consumer.proto
+	protoc -I=. --go_out . --go_opt paths=source_relative --go-grpc_out . --go-grpc_opt paths=source_relative v1/analyzer/analyzer.proto