chore: open some more syscalls from tarian-detector

- add 2nd queue for tarian-detection event
- skip eBPF events from this process ID
- send events to tarian-server with Go routine

cmd/tarianctl/cmd/install/defaultValues.go

@@ -43,7 +43,7 @@ func natsHelmDefaultValues(natsValuesFile string) error {
 				},
 				FileStorage: StorageOpts{
 					Enabled: true,
-					Size:    "50Mi",
+					Size:    "200Mi",
 				},
 			},
 		},

cmd/tarianctl/cmd/install/defaultValues_test.go

@@ -37,8 +37,8 @@ func TestNatsHelmDefaultValues(t *testing.T) {
 		t.Errorf("natsValuesFile does not contain 'size: 128Mi' in MemStorage")
 	}
 
-	if !containsSubstring(string(yamlContent), "size: 50Mi") {
-		t.Errorf("natsValuesFile does not contain 'size: 50Mi' in FileStorage")
+	if !containsSubstring(string(yamlContent), "size: 200Mi") {
+		t.Errorf("natsValuesFile does not contain 'size: 200Mi' in FileStorage")
 	}
 }
 

dev/nats-helm-values.yaml

@@ -10,4 +10,4 @@ nats:
 
     fileStorage:
       enabled: true
-      size: 50Mi
+      size: 200Mi

go.mod

@@ -119,4 +119,4 @@ require (
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 )
 
-replace github.com/intelops/tarian-detector => github.com/andylibrian/tarian-detector v0.0.0-20240328042403-0ad0b29f56cf
+replace github.com/intelops/tarian-detector => github.com/andylibrian/tarian-detector v0.0.0-20240401225603-ad8e890004a8

go.sum

@@ -41,8 +41,8 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
-github.com/andylibrian/tarian-detector v0.0.0-20240328042403-0ad0b29f56cf h1:sCIBD/c64HW9pW41ws1hDzFSwiQ53etRTl35zzYQzuw=
-github.com/andylibrian/tarian-detector v0.0.0-20240328042403-0ad0b29f56cf/go.mod h1:dXcRWq8AHABseHsjcnM8iJqwXCGX+dGGOR8kiXw1acY=
+github.com/andylibrian/tarian-detector v0.0.0-20240401225603-ad8e890004a8 h1:qHPCT8fRp50sbB1s2NWsP2B6wPRDqFJOjeYwbMMIuKk=
+github.com/andylibrian/tarian-detector v0.0.0-20240401225603-ad8e890004a8/go.mod h1:dXcRWq8AHABseHsjcnM8iJqwXCGX+dGGOR8kiXw1acY=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=

pkg/nodeagent/nodeagent.go

@@ -220,6 +220,11 @@ func (n *NodeAgent) loopTarianDetectorReadEvents(ctx context.Context) error {
 				continue
 			}
 
+			thisPID := event["processId"].(uint32)
+			if thisPID == 1 {
+				continue
+			}
+
 			pid := event["hostProcessId"].(uint32)
 
 			// Retrieve the container ID.
@@ -238,9 +243,6 @@ func (n *NodeAgent) loopTarianDetectorReadEvents(ctx context.Context) error {
 				continue
 			}
 
-			// TODO: sys_execve_entry could be added here
-			// But for kubectl exec, the detected entry comm is still the wrapper: runc:init
-			// With sys_execve_exit, the comm is the target process
 			detectionDataType := event["eventId"].(string)
 			if detectionDataType == "sys_execve_entry" {
 				execEvent, err2 := n.execEventFromTarianDetector(event, containerID, pod)
@@ -255,7 +257,7 @@ func (n *NodeAgent) loopTarianDetectorReadEvents(ctx context.Context) error {
 					}
 				}
 
-				n.logger.WithField("execEvent", execEvent).WithField("event", event).Info("DEBUG")
+				n.logger.WithField("execEvent", execEvent).WithField("event", event).Info("=============== DEBUG")
 			}
 
 			byteData, err := json.Marshal(event)
@@ -264,7 +266,7 @@ func (n *NodeAgent) loopTarianDetectorReadEvents(ctx context.Context) error {
 				continue
 			}
 
-			n.SendDetectionEventToClusterAgent(detectionDataType, string(byteData))
+			go n.SendDetectionEventToClusterAgent(detectionDataType, string(byteData))
 		}
 	}
 }
@@ -387,10 +389,10 @@ func (n *NodeAgent) handleExecEvent(evt *ExecEvent) error {
 
 		if registerProcess {
 			n.logger.WithField("comm", evt).Debug("violated process detected, going to register")
-			n.RegisterViolationsAsNewConstraint(violation)
+			go n.RegisterViolationsAsNewConstraint(violation)
 		} else {
 			n.logger.WithField("comm", evt).Debug("violated process detected")
-			n.ReportViolationsToClusterAgent(violation)
+			go n.ReportViolationsToClusterAgent(violation)
 		}
 	}
 
@@ -418,7 +420,7 @@ func (n *NodeAgent) SendDetectionEventToClusterAgent(detectionDataType, detectio
 	if err != nil {
 		n.logger.Error("error while sending detection events", "err", err)
 	} else {
-		n.logger.Debug("ingest event response", "response", resp)
+		n.logger.WithField("response", resp).Trace("ingest event response", "response", resp)
 	}
 }
 

pkg/server/event_server.go

@@ -14,9 +14,10 @@ import (
 // EventServer handles gRPC calls related to event ingestion and retrieval.
 type EventServer struct {
 	tarianpb.UnimplementedEventServer
-	eventStore     store.EventStore
-	ingestionQueue protoqueue.QueuePublisher
-	logger         *logrus.Logger
+	eventStore                      store.EventStore
+	ingestionQueue                  protoqueue.QueuePublisher
+	ingestionQueueForEventDetection protoqueue.QueuePublisher
+	logger                          *logrus.Logger
 }
 
 // NewEventServer creates a new EventServer instance.
@@ -28,11 +29,12 @@ type EventServer struct {
 //
 // Returns:
 // - *EventServer: A new instance of EventServer.
-func NewEventServer(logger *logrus.Logger, s store.EventStore, ingestionQueue protoqueue.QueuePublisher) *EventServer {
+func NewEventServer(logger *logrus.Logger, s store.EventStore, ingestionQueue protoqueue.QueuePublisher, ingestionQueueForEventDetection protoqueue.QueuePublisher) *EventServer {
 	return &EventServer{
-		eventStore:     s,
-		ingestionQueue: ingestionQueue,
-		logger:         logger,
+		eventStore:                      s,
+		ingestionQueue:                  ingestionQueue,
+		ingestionQueueForEventDetection: ingestionQueueForEventDetection,
+		logger:                          logger,
 	}
 }
 
@@ -55,7 +57,12 @@ func (es *EventServer) IngestEvent(ctx context.Context, request *tarianpb.Ingest
 		return nil, status.Error(codes.InvalidArgument, "required event is empty")
 	}
 
-	err := es.ingestionQueue.Publish(request.GetEvent())
+	var err error
+	if event.Type == tarianpb.EventTypeDetection {
+		err = es.ingestionQueueForEventDetection.Publish(request.GetEvent())
+	} else {
+		err = es.ingestionQueue.Publish(request.GetEvent())
+	}
 
 	if err != nil {
 		es.logger.WithError(err).Error("error while handling ingest event")

pkg/server/ingestion_worker.go

@@ -10,9 +10,10 @@ import (
 
 // IngestionWorker handles the ingestion of events from a message queue.
 type IngestionWorker struct {
-	eventStore     store.EventStore
-	IngestionQueue protoqueue.QueueSubscriber
-	logger         *logrus.Logger
+	eventStore                      store.EventStore
+	IngestionQueue                  protoqueue.QueueSubscriber
+	ingestionQueueForEventDetection protoqueue.QueueSubscriber
+	logger                          *logrus.Logger
 }
 
 // NewIngestionWorker creates a new IngestionWorker instance.
@@ -24,11 +25,12 @@ type IngestionWorker struct {
 //
 // Returns:
 // - *IngestionWorker: A new instance of IngestionWorker.
-func NewIngestionWorker(logger *logrus.Logger, eventStore store.EventStore, queueSubscriber protoqueue.QueueSubscriber) *IngestionWorker {
+func NewIngestionWorker(logger *logrus.Logger, eventStore store.EventStore, queueSubscriber protoqueue.QueueSubscriber, queueSubscriberForEventDetection protoqueue.QueueSubscriber) *IngestionWorker {
 	return &IngestionWorker{
-		eventStore:     eventStore,
-		IngestionQueue: queueSubscriber,
-		logger:         logger,
+		eventStore:                      eventStore,
+		IngestionQueue:                  queueSubscriber,
+		ingestionQueueForEventDetection: queueSubscriberForEventDetection,
+		logger:                          logger,
 	}
 }
 
@@ -40,8 +42,14 @@ func NewIngestionWorker(logger *logrus.Logger, eventStore store.EventStore, queu
 // - If it is a valid event, it updates the server timestamp and stores the event in the event store.
 // - If there are errors during processing, they are logged.
 func (iw *IngestionWorker) Start() {
+	go iw.loopConsumeQueue(iw.IngestionQueue)
+	go iw.loopConsumeQueueEventDetection(iw.ingestionQueueForEventDetection)
+}
+
+func (iw *IngestionWorker) loopConsumeQueue(queue protoqueue.QueueSubscriber) {
 	for {
-		msg, err := iw.IngestionQueue.NextMessage(&tarianpb.Event{})
+		msg, err := queue.NextMessage(&tarianpb.Event{})
+		iw.logger.WithField("event", msg).Infof("loopConsumeQueue: got message")
 		if err != nil {
 			iw.logger.WithError(err).Error("error while processing event")
 			continue
@@ -61,3 +69,26 @@ func (iw *IngestionWorker) Start() {
 		}
 	}
 }
+
+func (iw *IngestionWorker) loopConsumeQueueEventDetection(queue protoqueue.QueueSubscriber) {
+	for {
+		msg, err := queue.NextMessage(&tarianpb.Event{})
+		if err != nil {
+			// iw.logger.WithError(err).Error("error while processing event")
+			continue
+		}
+
+		event, ok := msg.(*tarianpb.Event)
+		if !ok {
+			// iw.logger.WithError(err).Error("error while processing event")
+			continue
+		}
+
+		event.ServerTimestamp = timestamppb.Now()
+		_ = iw.eventStore.Add(event)
+
+		// if err != nil {
+		// iw.logger.WithError(err).Error("error while processing event")
+		// }
+	}
+}