fix(procmon): retire audit events on EOE and keep receiving on overflow

procmon.go

@@ -1,10 +1,14 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"sync"
+	"syscall"
 	"time"
+
+	"github.com/elastic/go-libaudit/v2/auparse"
 )
 
 const (
@@ -13,6 +17,14 @@ const (
 	processMonitorTag = "procmon"
 )
 
+// pendingEventWindow is how many audit records may arrive without touching an
+// event before it is dropped. Events are retired on their EOE record; this
+// catches the ones that never get one, such as single-record events. It also
+// bounds the map: at most this many events can have been touched recently.
+// An event whose remaining records arrive after it was dropped is lost, since
+// the late records alone do not make it ready.
+const pendingEventWindow = 8192
+
 type ProcessMonitor struct {
 	CorrelationId         string
 	Repo                  string
@@ -22,6 +34,10 @@ type ProcessMonitor struct {
 	DisableFileMonitoring bool
 	Events                map[int]*Event
 	mutex                 sync.RWMutex
+
+	records         int64
+	droppedEvents   int
+	receiveOverflow int
 }
 
 type Process struct {
@@ -50,13 +66,21 @@ type Event struct {
 	EventType         string
 	Status            string
 	SentForProcessing bool
+	lastRecord        int64
 }
 
-func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interface{}) {
+func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interface{}) *Event {
 	p.mutex.Lock()
-	_, found := p.Events[sequence]
+	p.records++
+	event, found := p.Events[sequence]
 	if !found {
-		p.Events[sequence] = &Event{}
+		event = &Event{}
+		p.Events[sequence] = event
+	}
+	// refreshed before the sweep, so the event being prepared is never dropped
+	event.lastRecord = p.records
+	if !found && len(p.Events) > 2*pendingEventWindow {
+		p.dropIdleEventsLocked()
 	}
 
 	// Tags
@@ -65,32 +89,32 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac
 	if found {
 		tags := value.([]string)
 		for _, tag := range tags {
-			p.Events[sequence].EventType = tag
-			p.Events[sequence].Syscall = getValue("syscall", eventMap)
-			p.Events[sequence].Exe = getValue("exe", eventMap)
-			p.Events[sequence].Pid = getValue("pid", eventMap)
-			p.Events[sequence].PPid = getValue("ppid", eventMap)
-			p.Events[sequence].Euid = getValue("euid", eventMap)
+			event.EventType = tag
+			event.Syscall = getValue("syscall", eventMap)
+			event.Exe = getValue("exe", eventMap)
+			event.Pid = getValue("pid", eventMap)
+			event.PPid = getValue("ppid", eventMap)
+			event.Euid = getValue("euid", eventMap)
 			timestamp, err := time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", getValue("@timestamp", eventMap))
 			if err != nil {
 				timestamp = time.Now().UTC()
 			}
-			p.Events[sequence].Timestamp = timestamp
-			p.Events[sequence].Status = getValue("result", eventMap)
+			event.Timestamp = timestamp
+			event.Status = getValue("result", eventMap)
 		}
 	}
 
 	// Current working directory
 	_, found = eventMap["cwd"]
 	if found {
-		p.Events[sequence].Path = getValue("cwd", eventMap)
+		event.Path = getValue("cwd", eventMap)
 	}
 
 	// Connect
 	_, found = eventMap["addr"]
 	if found {
-		p.Events[sequence].IPAddress = getValue("addr", eventMap)
-		p.Events[sequence].Port = getValue("port", eventMap)
+		event.IPAddress = getValue("addr", eventMap)
+		event.Port = getValue("port", eventMap)
 	}
 
 	// Process start
@@ -103,7 +127,7 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac
 			WriteLog(fmt.Sprintf("could not parse argc:%v", argc))
 		}
 		for i := 0; i < argCount; i++ {
-			p.Events[sequence].ProcessArguments = append(p.Events[sequence].ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)]))
+			event.ProcessArguments = append(event.ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)]))
 		}
 	}
 
@@ -113,11 +137,12 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac
 	if found {
 		nameTypeStr := fmt.Sprintf("%v", nameType)
 		if nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL" {
-			p.Events[sequence].FileName = getValue("name", eventMap)
+			event.FileName = getValue("name", eventMap)
 		}
 	}
 
 	p.mutex.Unlock()
+	return event
 }
 
 func getValue(key string, eventMap map[string]interface{}) string {
@@ -129,6 +154,58 @@ func getValue(key string, eventMap map[string]interface{}) string {
 	return ""
 }
 
+// handleRecord adds one audit record to its event and returns the event once
+// it is ready to dispatch. The EOE record ends a multi-record event, so the
+// event is removed then.
+func (p *ProcessMonitor) handleRecord(msgType auparse.AuditMessageType, message *auparse.AuditMessage) *Event {
+	sequence := int(message.Sequence)
+	if msgType == auparse.AUDIT_EOE {
+		p.retireEvent(sequence)
+		return nil
+	}
+
+	event := p.PrepareEvent(sequence, message.ToMapStr())
+	if !isEventReady(event) {
+		return nil
+	}
+	p.markEventSent(event)
+	return event
+}
+
+// retireEvent removes a sequence once its EOE record arrives. Handlers keep
+// their own pointer, so an event already dispatched is unaffected.
+func (p *ProcessMonitor) retireEvent(sequence int) {
+	p.mutex.Lock()
+	delete(p.Events, sequence)
+	p.mutex.Unlock()
+}
+
+func (p *ProcessMonitor) dropIdleEventsLocked() {
+	dropped := 0
+	for sequence, event := range p.Events {
+		if event.lastRecord <= p.records-pendingEventWindow {
+			delete(p.Events, sequence)
+			dropped++
+		}
+	}
+	p.droppedEvents += dropped
+	WriteLog(fmt.Sprintf("[procmon] dropped %d audit events without EOE, total=%d", dropped, p.droppedEvents))
+}
+
+// isReceiveOverflow reports whether the kernel dropped audit messages because
+// the netlink socket was full. The socket stays usable, so receiving continues.
+func isReceiveOverflow(err error) bool {
+	return errors.Is(err, syscall.ENOBUFS)
+}
+
+func (p *ProcessMonitor) noteReceiveOverflow() {
+	p.receiveOverflow++
+	// log on 1, 2, 4, 8, ... so a sustained overflow does not flood the log
+	if p.receiveOverflow&(p.receiveOverflow-1) == 0 {
+		WriteLog(fmt.Sprintf("[procmon] audit messages lost, netlink buffer full, total=%d", p.receiveOverflow))
+	}
+}
+
 func (p *ProcessMonitor) markEventSent(event *Event) {
 	p.mutex.Lock()
 	event.SentForProcessing = true

procmon_linux.go

@@ -151,7 +151,9 @@ func (p *ProcessMonitor) MonitorProcesses(errc chan error) {
 	WriteLog("receive called")
 	WriteLog("\n")
 
-	p.receive(client)
+	if err := p.receive(client); err != nil {
+		WriteLog(fmt.Sprintf("[procmon] receive stopped: %v", err))
+	}
 }
 
 func fileMonitorAuditRule(workingDirectory, goarch string) string {
@@ -175,6 +177,10 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error {
 	for {
 		rawEvent, err := r.Receive(false)
 		if err != nil {
+			if isReceiveOverflow(err) {
+				p.noteReceiveOverflow()
+				continue
+			}
 			return errors.Wrap(err, "receive failed")
 		}
 
@@ -188,12 +194,8 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error {
 		if err != nil {
 			return errors.Wrap(err, "parse failed.")
 		}
-		eventMap := message.ToMapStr()
-
-		p.PrepareEvent(int(message.Sequence), eventMap)
-		if isEventReady(p.Events[int(message.Sequence)]) {
-			p.markEventSent(p.Events[int(message.Sequence)])
-			go eventHandler.HandleEvent(p.Events[int(message.Sequence)])
+		if event := p.handleRecord(rawEvent.Type, message); event != nil {
+			go eventHandler.HandleEvent(event)
 		}
 
 	}

procmon_linux_test.go

@@ -4,8 +4,13 @@
 package main
 
 import (
+	"errors"
 	"os"
+	"syscall"
 	"testing"
+
+	"github.com/elastic/go-libaudit/v2"
+	"github.com/elastic/go-libaudit/v2/auparse"
 )
 
 func Test_fileMonitorAuditRule(t *testing.T) {
@@ -65,3 +70,61 @@ func Test_getProcessExe(t *testing.T) {
 	}
 
 }
+
+// fakeNetlink replays audit records and errors to AuditClient.Receive.
+type fakeNetlink struct {
+	steps []fakeNetlinkStep
+}
+
+type fakeNetlinkStep struct {
+	msgType auparse.AuditMessageType
+	text    string
+	err     error
+}
+
+func (f *fakeNetlink) Receive(bool, libaudit.NetlinkParser) ([]syscall.NetlinkMessage, error) {
+	if len(f.steps) == 0 {
+		return nil, syscall.EBADF
+	}
+	step := f.steps[0]
+	f.steps = f.steps[1:]
+	if step.err != nil {
+		return nil, step.err
+	}
+	return []syscall.NetlinkMessage{{
+		Header: syscall.NlMsghdr{Type: uint16(step.msgType)},
+		Data:   []byte(step.text),
+	}}, nil
+}
+
+func (f *fakeNetlink) Send(syscall.NetlinkMessage) (uint32, error) { return 0, nil }
+func (f *fakeNetlink) Close() error                                { return nil }
+
+func TestReceiveContinuesAfterOverflow(t *testing.T) {
+	netlink := &fakeNetlink{steps: []fakeNetlinkStep{
+		{err: syscall.ENOBUFS},
+		{msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:42): arch=c000003e syscall=44 success=yes exit=121 a0=3 a1=7f00 a2=79 a3=0 items=0 ppid=1 pid=100 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="python3" exe="/usr/bin/python3.12" key="netmon"`},
+		{err: syscall.ENOBUFS},
+		{msgType: auparse.AUDIT_PROCTITLE, text: `audit(1700000000.123:42): proctitle=707974686F6E33`},
+		{msgType: auparse.AUDIT_EOE, text: `audit(1700000000.123:42): `},
+	}}
+	p := &ProcessMonitor{}
+
+	err := p.receive(&libaudit.AuditClient{Netlink: netlink})
+
+	if !errors.Is(err, syscall.EBADF) {
+		t.Fatalf("receive should stop only on the terminal error, got %v", err)
+	}
+	if p.receiveOverflow != 2 {
+		t.Fatalf("expected 2 overflows counted, got %d", p.receiveOverflow)
+	}
+	if len(netlink.steps) != 0 {
+		t.Fatalf("records after an overflow were not read: %d left", len(netlink.steps))
+	}
+	if p.records != 2 {
+		t.Fatalf("expected the SYSCALL and PROCTITLE records to be processed, got %d", p.records)
+	}
+	if len(p.Events) != 0 {
+		t.Fatalf("expected the event to be retired on EOE, got %d pending", len(p.Events))
+	}
+}