Merge pull request #394 from kubescape/feature/processtree

Feature/processtree

main.go

@@ -32,6 +32,8 @@ import (
 	"github.com/kubescape/node-agent/pkg/objectcache/k8scache"
 	"github.com/kubescape/node-agent/pkg/objectcache/networkneighborhoodcache"
 	objectcachev1 "github.com/kubescape/node-agent/pkg/objectcache/v1"
+	"github.com/kubescape/node-agent/pkg/processmanager"
+	processmanagerv1 "github.com/kubescape/node-agent/pkg/processmanager/v1"
 	"github.com/kubescape/node-agent/pkg/relevancymanager"
 	relevancymanagerv1 "github.com/kubescape/node-agent/pkg/relevancymanager/v1"
 	rulebinding "github.com/kubescape/node-agent/pkg/rulebindingmanager"
@@ -193,26 +195,27 @@ func main() {
 	var networkManagerClient networkmanager.NetworkManagerClient
 	var dnsManagerClient dnsmanager.DNSManagerClient
 	var dnsResolver dnsmanager.DNSResolver
-	if cfg.EnableNetworkTracing {
+	if cfg.EnableNetworkTracing || cfg.EnableRuntimeDetection {
 		dnsManager := dnsmanager.CreateDNSManager()
 		dnsManagerClient = dnsManager
 		// NOTE: dnsResolver is set for threat detection.
 		dnsResolver = dnsManager
 		networkManagerClient = networkmanagerv2.CreateNetworkManager(ctx, cfg, clusterData.ClusterName, k8sClient, storageClient, dnsManager, preRunningContainersIDs, k8sObjectCache)
 	} else {
-		if cfg.EnableRuntimeDetection {
-			logger.L().Ctx(ctx).Fatal("Network tracing is disabled, but runtime detection is enabled. Network tracing is required for runtime detection.")
-		}
 		dnsManagerClient = dnsmanager.CreateDNSManagerMock()
 		dnsResolver = dnsmanager.CreateDNSManagerMock()
 		networkManagerClient = networkmanager.CreateNetworkManagerMock()
 	}
 
 	var ruleManager rulemanager.RuleManagerClient
+	var processManager processmanager.ProcessManagerClient
 	var objCache objectcache.ObjectCache
 	var ruleBindingNotify chan rulebinding.RuleBindingNotify
 
 	if cfg.EnableRuntimeDetection {
+		// create the process manager
+		processManager = processmanagerv1.CreateProcessManager(ctx)
+
 		// create ruleBinding cache
 		ruleBindingCache := rulebindingcachev1.NewCache(nodeName, k8sClient)
 		dWatcher.AddAdaptor(ruleBindingCache)
@@ -235,7 +238,7 @@ func main() {
 		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, nodeName)
 
 		// create runtimeDetection managers
-		ruleManager, err = rulemanagerv1.CreateRuleManager(ctx, cfg, k8sClient, ruleBindingCache, objCache, exporter, prometheusExporter, nodeName, clusterData.ClusterName)
+		ruleManager, err = rulemanagerv1.CreateRuleManager(ctx, cfg, k8sClient, ruleBindingCache, objCache, exporter, prometheusExporter, nodeName, clusterData.ClusterName, processManager)
 		if err != nil {
 			logger.L().Ctx(ctx).Fatal("error creating RuleManager", helpers.Error(err))
 		}
@@ -244,6 +247,7 @@ func main() {
 		ruleManager = rulemanager.CreateRuleManagerMock()
 		objCache = objectcache.NewObjectCacheMock()
 		ruleBindingNotify = make(chan rulebinding.RuleBindingNotify, 1)
+		processManager = processmanager.CreateProcessManagerMock()
 	}
 
 	// Create the node profile manager
@@ -269,7 +273,7 @@ func main() {
 	}
 
 	// Create the container handler
-	mainHandler, err := containerwatcher.CreateIGContainerWatcher(cfg, applicationProfileManager, k8sClient, relevancyManager, networkManagerClient, dnsManagerClient, prometheusExporter, ruleManager, malwareManager, preRunningContainersIDs, &ruleBindingNotify, containerRuntime, nil)
+	mainHandler, err := containerwatcher.CreateIGContainerWatcher(cfg, applicationProfileManager, k8sClient, relevancyManager, networkManagerClient, dnsManagerClient, prometheusExporter, ruleManager, malwareManager, preRunningContainersIDs, &ruleBindingNotify, containerRuntime, nil, processManager)
 	if err != nil {
 		logger.L().Ctx(ctx).Fatal("error creating the container watcher", helpers.Error(err))
 	}

pkg/containerwatcher/v1/container_watcher.go

@@ -42,6 +42,7 @@ import (
 	tracersshtype "github.com/kubescape/node-agent/pkg/ebpf/gadgets/ssh/types"
 	tracersymlink "github.com/kubescape/node-agent/pkg/ebpf/gadgets/symlink/tracer"
 	tracersymlinktype "github.com/kubescape/node-agent/pkg/ebpf/gadgets/symlink/types"
+	"github.com/kubescape/node-agent/pkg/processmanager"
 
 	"github.com/kubescape/node-agent/pkg/malwaremanager"
 	"github.com/kubescape/node-agent/pkg/metricsmanager"
@@ -153,11 +154,13 @@ type IGContainerWatcher struct {
 	ruleBindingPodNotify *chan rulebinding.RuleBindingNotify
 	// container runtime
 	runtime *containerutilsTypes.RuntimeConfig
+	// process manager
+	processManager processmanager.ProcessManagerClient
 }
 
 var _ containerwatcher.ContainerWatcher = (*IGContainerWatcher)(nil)
 
-func CreateIGContainerWatcher(cfg config.Config, applicationProfileManager applicationprofilemanager.ApplicationProfileManagerClient, k8sClient *k8sinterface.KubernetesApi, relevancyManager relevancymanager.RelevancyManagerClient, networkManagerClient networkmanager.NetworkManagerClient, dnsManagerClient dnsmanager.DNSManagerClient, metrics metricsmanager.MetricsManager, ruleManager rulemanager.RuleManagerClient, malwareManager malwaremanager.MalwareManagerClient, preRunningContainers mapset.Set[string], ruleBindingPodNotify *chan rulebinding.RuleBindingNotify, runtime *containerutilsTypes.RuntimeConfig, thirdPartyEventReceivers *maps.SafeMap[utils.EventType, mapset.Set[containerwatcher.EventReceiver]]) (*IGContainerWatcher, error) {
+func CreateIGContainerWatcher(cfg config.Config, applicationProfileManager applicationprofilemanager.ApplicationProfileManagerClient, k8sClient *k8sinterface.KubernetesApi, relevancyManager relevancymanager.RelevancyManagerClient, networkManagerClient networkmanager.NetworkManagerClient, dnsManagerClient dnsmanager.DNSManagerClient, metrics metricsmanager.MetricsManager, ruleManager rulemanager.RuleManagerClient, malwareManager malwaremanager.MalwareManagerClient, preRunningContainers mapset.Set[string], ruleBindingPodNotify *chan rulebinding.RuleBindingNotify, runtime *containerutilsTypes.RuntimeConfig, thirdPartyEventReceivers *maps.SafeMap[utils.EventType, mapset.Set[containerwatcher.EventReceiver]], processManager processmanager.ProcessManagerClient) (*IGContainerWatcher, error) {
 	// Use container collection to get notified for new containers
 	containerCollection := &containercollection.ContainerCollection{}
 	// Create a tracer collection instance
@@ -203,6 +206,7 @@ func CreateIGContainerWatcher(cfg config.Config, applicationProfileManager appli
 			path = event.Args[0]
 		}
 		metrics.ReportEvent(utils.ExecveEventType)
+		processManager.ReportEvent(utils.ExecveEventType, &event)
 		applicationProfileManager.ReportFileExec(k8sContainerID, path, event.Args)
 		relevancyManager.ReportFileExec(event.Runtime.ContainerID, k8sContainerID, path)
 		ruleManager.ReportEvent(utils.ExecveEventType, &event)
@@ -449,6 +453,7 @@ func CreateIGContainerWatcher(cfg config.Config, applicationProfileManager appli
 		runtime:                      runtime,
 		thirdPartyTracers:            mapset.NewSet[containerwatcher.CustomTracer](),
 		thirdPartyContainerReceivers: mapset.NewSet[containerwatcher.ContainerReceiver](),
+		processManager:               processManager,
 	}, nil
 }
 
@@ -494,11 +499,16 @@ func (ch *IGContainerWatcher) UnregisterContainerReceiver(receiver containerwatc
 
 func (ch *IGContainerWatcher) Start(ctx context.Context) error {
 	if !ch.running {
-
 		if err := ch.startContainerCollection(ctx); err != nil {
 			return fmt.Errorf("setting up container collection: %w", err)
 		}
 
+		// We want to populate the initial processes before starting the tracers but after retrieving the shims.
+		if err := ch.processManager.PopulateInitialProcesses(); err != nil {
+			ch.stopContainerCollection()
+			return fmt.Errorf("populating initial processes: %w", err)
+		}
+
 		if err := ch.startTracers(); err != nil {
 			ch.stopContainerCollection()
 			return fmt.Errorf("starting app behavior tracing: %w", err)

pkg/containerwatcher/v1/container_watcher_private.go

@@ -86,6 +86,7 @@ func (ch *IGContainerWatcher) startContainerCollection(ctx context.Context) erro
 		ch.networkManager.ContainerCallback,
 		ch.malwareManager.ContainerCallback,
 		ch.ruleManager.ContainerCallback,
+		ch.processManager.ContainerCallback,
 	}
 
 	for receiver := range ch.thirdPartyContainerReceivers.Iter() {

pkg/containerwatcher/v1/open_test.go

@@ -23,7 +23,7 @@ func BenchmarkIGContainerWatcher_openEventCallback(b *testing.B) {
 	assert.NoError(b, err)
 	mockExporter := metricsmanager.NewMetricsMock()
 
-	mainHandler, err := CreateIGContainerWatcher(cfg, nil, nil, relevancyManager, nil, nil, mockExporter, nil, nil, nil, nil, nil, nil)
+	mainHandler, err := CreateIGContainerWatcher(cfg, nil, nil, relevancyManager, nil, nil, mockExporter, nil, nil, nil, nil, nil, nil, nil)
 	assert.NoError(b, err)
 	event := &traceropentype.Event{
 		Event: types.Event{

pkg/processmanager/process_manager_interface.go

@@ -0,0 +1,20 @@
+package processmanager
+
+import (
+	apitypes "github.com/armosec/armoapi-go/armotypes"
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/node-agent/pkg/utils"
+)
+
+// ProcessManagerClient is the interface for the process manager client.
+// It provides methods to get process tree for a container or a PID.
+// The manager is responsible for maintaining the process tree for all containers.
+type ProcessManagerClient interface {
+	GetProcessTreeForPID(containerID string, pid int) (apitypes.Process, error)
+	// PopulateInitialProcesses is called to populate the initial process tree (parsed from /proc) for all containers.
+	PopulateInitialProcesses() error
+
+	// ReportEvent will be called to report new exec events to the process manager.
+	ReportEvent(eventType utils.EventType, event utils.K8sEvent)
+	ContainerCallback(notif containercollection.PubSubEvent)
+}

pkg/processmanager/process_manager_mock.go

@@ -0,0 +1,32 @@
+package processmanager
+
+import (
+	apitypes "github.com/armosec/armoapi-go/armotypes"
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/node-agent/pkg/utils"
+)
+
+type ProcessManagerMock struct {
+}
+
+var _ ProcessManagerClient = (*ProcessManagerMock)(nil)
+
+func CreateProcessManagerMock() *ProcessManagerMock {
+	return &ProcessManagerMock{}
+}
+
+func (p *ProcessManagerMock) GetProcessTreeForPID(containerID string, pid int) (apitypes.Process, error) {
+	return apitypes.Process{}, nil
+}
+
+func (p *ProcessManagerMock) PopulateInitialProcesses() error {
+	return nil
+}
+
+func (p *ProcessManagerMock) ReportEvent(eventType utils.EventType, event utils.K8sEvent) {
+	// no-op
+}
+
+func (p *ProcessManagerMock) ContainerCallback(notif containercollection.PubSubEvent) {
+	// no-op
+}