Replace host sensor with node agent sensing

cmd/main.go

@@ -33,6 +33,7 @@ import (
 	"github.com/kubescape/node-agent/pkg/exporters"
 	"github.com/kubescape/node-agent/pkg/fimmanager"
 	"github.com/kubescape/node-agent/pkg/healthmanager"
+	hostsensormanager "github.com/kubescape/node-agent/pkg/hostsensormanager"
 	"github.com/kubescape/node-agent/pkg/malwaremanager"
 	malwaremanagerv1 "github.com/kubescape/node-agent/pkg/malwaremanager/v1"
 	"github.com/kubescape/node-agent/pkg/metricsmanager"
@@ -181,6 +182,17 @@ func main() {
 	}
 	dWatcher.AddAdaptor(k8sObjectCache)
 
+	// Create the host sensor manager
+	hostSensorConfig := hostsensormanager.Config{
+		Enabled:  cfg.EnableHostSensor,
+		Interval: cfg.HostSensorInterval,
+		NodeName: cfg.NodeName,
+	}
+	hostSensorManager, err := hostsensormanager.NewHostSensorManager(hostSensorConfig)
+	if err != nil {
+		logger.L().Ctx(ctx).Fatal("error creating HostSensorManager", helpers.Error(err))
+	}
+
 	// Create the seccomp manager
 	var seccompManager seccompmanager.SeccompManagerClient
 	if cfg.EnableSeccomp {
@@ -394,6 +406,12 @@ func main() {
 	// Start the prometheusExporter
 	prometheusExporter.Start()
 
+	// Start the host sensor manager
+	if err = hostSensorManager.Start(ctx); err != nil {
+		logger.L().Ctx(ctx).Fatal("error starting host sensor manager", helpers.Error(err))
+	}
+	defer hostSensorManager.Stop()
+
 	// Start the FIM manager
 	if fimManager != nil {
 		err = fimManager.Start(ctx)

configuration/config.json

@@ -95,5 +95,7 @@
         "exporters": {
           "stdoutExporter": true
         }
-      }
+      },
+      "hostSensorEnabled": true,
+      "hostSensorInterval": "1m"
     }

go.mod

@@ -50,6 +50,7 @@ require (
 	github.com/spf13/afero v1.15.0
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
+	github.com/weaveworks/procspy v0.0.0-20150706124340-cb970aa190c3
 	go.uber.org/multierr v1.11.0
 	golang.org/x/net v0.47.0
 	golang.org/x/sys v0.38.0

go.sum

@@ -1936,6 +1936,8 @@ github.com/wagoodman/go-partybus v0.0.0-20230516145632-8ccac152c651 h1:jIVmlAFIq
 github.com/wagoodman/go-partybus v0.0.0-20230516145632-8ccac152c651/go.mod h1:b26F2tHLqaoRQf8DywqzVaV1MQ9yvjb0OMcNl7Nxu20=
 github.com/wagoodman/go-progress v0.0.0-20230925121702-07e42b3cdba0 h1:0KGbf+0SMg+UFy4e1A/CPVvXn21f1qtWdeJwxZFoQG8=
 github.com/wagoodman/go-progress v0.0.0-20230925121702-07e42b3cdba0/go.mod h1:jLXFoL31zFaHKAAyZUh+sxiTDFe1L1ZHrcK2T1itVKA=
+github.com/weaveworks/procspy v0.0.0-20150706124340-cb970aa190c3 h1:UC4iN/yCDCObTBhKzo34/R2U6qptTPmqbzG6UiQVMUQ=
+github.com/weaveworks/procspy v0.0.0-20150706124340-cb970aa190c3/go.mod h1:cJTfuBcxkdbj8Mabk4PPdaf0AXv9TYEJmkFxKcWxYY4=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM=

pkg/config/config.go

@@ -90,6 +90,9 @@ type Config struct {
 	UpdateDataPeriod               time.Duration                        `mapstructure:"updateDataPeriod"`
 	WorkerChannelSize              int                                  `mapstructure:"workerChannelSize"`
 	WorkerPoolSize                 int                                  `mapstructure:"workerPoolSize"`
+	// Host sensor configuration
+	EnableHostSensor   bool          `mapstructure:"hostSensorEnabled"`
+	HostSensorInterval time.Duration `mapstructure:"hostSensorInterval"`
 }
 
 // FIMConfig defines the configuration for File Integrity Monitoring
@@ -191,6 +194,9 @@ func LoadConfig(path string) (Config, error) {
 	viper.SetDefault("fim::periodicConfig::maxFileSize", int64(100*1024*1024))
 	viper.SetDefault("fim::periodicConfig::followSymlinks", false)
 	viper.SetDefault("fim::exporters::stdoutExporter", false)
+	// Host sensor defaults
+	viper.SetDefault("hostSensorEnabled", true)
+	viper.SetDefault("hostSensorInterval", 5*time.Minute)
 
 	viper.AutomaticEnv()
 

pkg/config/config_test.go

@@ -39,6 +39,8 @@ func TestLoadConfig(t *testing.T) {
 				EnableFIM:                      true,
 				EnableNetworkStreaming:         false,
 				EnableEmbeddedSboms:            false,
+				EnableHostSensor:               true,
+				HostSensorInterval:             1 * time.Minute,
 				KubernetesMode:                 true,
 				NetworkStreamingInterval:       2 * time.Minute,
 				InitialDelay:                   2 * time.Minute,

pkg/hostsensormanager/crd_client.go

@@ -0,0 +1,150 @@
+package hostsensormanager
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/kubescape/go-logger"
+	"github.com/kubescape/go-logger/helpers"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/rest"
+)
+
+// CRDClient handles Kubernetes CRD operations
+type CRDClient struct {
+	dynamicClient dynamic.Interface
+	nodeName      string
+}
+
+// NewCRDClient creates a new CRD client
+func NewCRDClient(nodeName string) (*CRDClient, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get in-cluster config: %w", err)
+	}
+
+	dynamicClient, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create dynamic client: %w", err)
+	}
+
+	return &CRDClient{
+		dynamicClient: dynamicClient,
+		nodeName:      nodeName,
+	}, nil
+}
+
+// CreateOrUpdateHostData creates or updates a host data CRD
+func (c *CRDClient) CreateOrUpdateHostData(ctx context.Context, resource string, kind string, spec interface{}) error {
+	gvr := schema.GroupVersionResource{
+		Group:    HostDataGroup,
+		Version:  HostDataVersion,
+		Resource: resource,
+	}
+
+	// Create the unstructured object
+	unstructuredObj := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": fmt.Sprintf("%s/%s", HostDataGroup, HostDataVersion),
+			"kind":       kind,
+			"metadata": map[string]interface{}{
+				"name": c.nodeName,
+			},
+			"spec": spec,
+			"status": map[string]interface{}{
+				"lastSensed": metav1.Now().UTC().Format(time.RFC3339),
+			},
+		},
+	}
+
+	// Try to get existing resource
+	_, err := c.dynamicClient.Resource(gvr).Get(ctx, c.nodeName, metav1.GetOptions{})
+	if err != nil {
+		// Resource doesn't exist, create it
+		logger.L().Debug("creating new host data CRD",
+			helpers.String("kind", kind),
+			helpers.String("nodeName", c.nodeName))
+		_, err = c.dynamicClient.Resource(gvr).Create(ctx, unstructuredObj, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to create %s CRD: %w", kind, err)
+		}
+		logger.L().Info("created host data CRD",
+			helpers.String("kind", kind),
+			helpers.String("nodeName", c.nodeName))
+		return nil
+	}
+
+	// Resource exists, update it using patch
+	logger.L().Debug("updating existing host data CRD",
+		helpers.String("kind", kind),
+		helpers.String("nodeName", c.nodeName))
+
+	// Create patch data
+	patchData, err := json.Marshal(map[string]interface{}{
+		"spec": spec,
+		"status": Status{
+			LastSensed: metav1.Now(),
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal patch data: %w", err)
+	}
+
+	_, err = c.dynamicClient.Resource(gvr).Patch(ctx, c.nodeName, types.MergePatchType, patchData, metav1.PatchOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to patch %s CRD: %w", kind, err)
+	}
+
+	logger.L().Debug("updated host data CRD",
+		helpers.String("kind", kind),
+		helpers.String("nodeName", c.nodeName))
+	return nil
+}
+
+// UpdateStatus updates the status of a host data CRD with an error
+func (c *CRDClient) UpdateStatus(ctx context.Context, resource string, errorMsg string) error {
+	gvr := schema.GroupVersionResource{
+		Group:    HostDataGroup,
+		Version:  HostDataVersion,
+		Resource: resource,
+	}
+
+	patchData, err := json.Marshal(map[string]interface{}{
+		"status": Status{
+			LastSensed: metav1.Now(),
+			Error:      errorMsg,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal patch data: %w", err)
+	}
+
+	_, err = c.dynamicClient.Resource(gvr).Patch(ctx, c.nodeName, types.MergePatchType, patchData, metav1.PatchOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to update status: %w", err)
+	}
+
+	return nil
+}
+
+// toUnstructured converts a typed object to unstructured
+func toUnstructured(obj interface{}) (*unstructured.Unstructured, error) {
+	data, err := json.Marshal(obj)
+	if err != nil {
+		return nil, err
+	}
+
+	var unstructuredObj unstructured.Unstructured
+	err = json.Unmarshal(data, &unstructuredObj.Object)
+	if err != nil {
+		return nil, err
+	}
+
+	return &unstructuredObj, nil
+}