fix(procmon): only reassemble records of audited syscall events

procmon.go

@@ -5,10 +5,12 @@ import (
 	"fmt"
 	"math/bits"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"time"
 
+	"github.com/elastic/go-libaudit/v2"
 	"github.com/elastic/go-libaudit/v2/auparse"
 )
 
@@ -148,9 +150,67 @@ func getValue(key string, eventMap map[string]interface{}) string {
 type auditStream struct {
 	monitor  *ProcessMonitor
 	dispatch func(*Event)
+
+	// ops serializes push with maintain and flush, so an event cannot be
+	// given up between a record being admitted and being pushed
+	ops sync.Mutex
+
+	// syscallEvents holds the sequence numbers of the events in the
+	// reassembler, which all start with a SYSCALL record (see push)
+	mu            sync.Mutex
+	syscallEvents map[uint32]struct{}
+}
+
+// push passes a record to the reassembler if it belongs to an audited
+// syscall. The records the agent uses (EXECVE, SOCKADDR, CWD, PATH,
+// PROCTITLE) follow the event's SYSCALL record. Other records, such as BPF or
+// ANOM_PROMISCUOUS, can arrive alone without the records that complete an
+// event, and since the reassembler delivers events in sequence order, each
+// one would hold back every later event until it times out.
+func (s *auditStream) push(r *libaudit.Reassembler, msg *auparse.AuditMessage) {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+
+	s.mu.Lock()
+	if s.syscallEvents == nil {
+		s.syscallEvents = make(map[uint32]struct{})
+	}
+	_, found := s.syscallEvents[msg.Sequence]
+	if !found && msg.RecordType == auparse.AUDIT_SYSCALL {
+		s.syscallEvents[msg.Sequence] = struct{}{}
+		found = true
+	}
+	s.mu.Unlock()
+
+	if found {
+		r.PushMessage(msg)
+	}
+}
+
+// maintain gives up on events whose last record did not arrive in time.
+func (s *auditStream) maintain(r *libaudit.Reassembler) error {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+	return r.Maintain()
+}
+
+// flush delivers the events still in the reassembler and closes it.
+func (s *auditStream) flush(r *libaudit.Reassembler) {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+	r.Close()
 }
 
 func (s *auditStream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
+	if len(msgs) == 0 {
+		return
+	}
+	// records that arrive after this, such as the EOE after a PROCTITLE,
+	// are not passed on
+	s.mu.Lock()
+	delete(s.syscallEvents, msgs[0].Sequence)
+	s.mu.Unlock()
+
 	event := &Event{}
 	for _, msg := range msgs {
 		eventMap := msg.ToMapStr()
@@ -165,7 +225,7 @@ func (s *auditStream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
 }
 
 // EventsLost is called for gaps in the audit sequence numbers. A gap is not
-// always a loss, since record types this agent skips use sequence numbers too.
+// always a loss, since events this agent skips use sequence numbers too.
 func (s *auditStream) EventsLost(count int) {
 	// go-libaudit v2.3.2 underflows when a record arrives after its event was
 	// given up, which shows up as a huge count

procmon_linux.go

@@ -181,7 +181,8 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error {
 		dispatch = func(event *Event) { go eventHandler.HandleEvent(event) }
 	}
 
-	reassembler, err := libaudit.NewReassembler(reassemblerMaxInFlight, reassemblerTimeout, &auditStream{monitor: p, dispatch: dispatch})
+	stream := &auditStream{monitor: p, dispatch: dispatch}
+	reassembler, err := libaudit.NewReassembler(reassemblerMaxInFlight, reassemblerTimeout, stream)
 	if err != nil {
 		return errors.Wrap(err, "create reassembler")
 	}
@@ -190,13 +191,13 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error {
 	maintaining.Add(1)
 	go func() {
 		defer maintaining.Done()
-		maintainReassembler(reassembler, done)
+		maintainReassembler(stream, reassembler, done)
 	}()
 	// stop maintenance before the final flush, so no callback runs after return
 	defer func() {
 		close(done)
 		maintaining.Wait()
-		reassembler.Close()
+		stream.flush(reassembler)
 	}()
 
 	for {
@@ -219,23 +220,26 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error {
 			continue
 		}
 
-		if err := reassembler.Push(rawEvent.Type, rawEvent.Data); err != nil {
+		message, err := auparse.Parse(rawEvent.Type, string(rawEvent.Data))
+		if err != nil {
 			noteLoss(&p.invalidRecords, 1, fmt.Sprintf("dropped audit record with an invalid header: %v", err))
+			continue
 		}
+		stream.push(reassembler, message)
 	}
 }
 
 // maintainReassembler gives up on events whose last record did not arrive in
 // time, as Auditbeat does, until done is closed.
-func maintainReassembler(r *libaudit.Reassembler, done <-chan struct{}) {
+func maintainReassembler(s *auditStream, r *libaudit.Reassembler, done <-chan struct{}) {
 	ticker := time.NewTicker(reassemblerMaintain)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-done:
 			return
 		case <-ticker.C:
-			if err := r.Maintain(); err != nil {
+			if err := s.maintain(r); err != nil {
 				return
 			}
 		}

procmon_linux_test.go

@@ -251,3 +251,34 @@ func TestReceiveWaitsForMaintenanceBeforeReturning(t *testing.T) {
 		t.Fatalf("receive should stop when the client is closed, got %v", err)
 	}
 }
+
+func TestReceiveDoesNotHoldEventsBehindRecordsOutsideSyscalls(t *testing.T) {
+	release := make(chan struct{})
+	netlink := &fakeNetlink{steps: []fakeNetlinkStep{
+		// logged on its own, with no record that completes it
+		{msgType: auparse.AUDIT_BPF, text: `audit(1700000000.123:96): prog-id=12 op=UNLOAD`},
+		{msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:97): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=97 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="curl" exe="/usr/bin/curl" key="netmon"`},
+		{msgType: auparse.AUDIT_SOCKADDR, text: `audit(1700000000.123:97): saddr=020001BBC000020A0000000000000000`},
+		{msgType: auparse.AUDIT_PROCTITLE, text: `audit(1700000000.123:97): proctitle=6375726C`},
+		{block: release},
+	}}
+	dispatched := make(chan *Event, 1)
+	p := &ProcessMonitor{dispatchEvent: func(e *Event) { dispatched <- e }}
+
+	result := make(chan error, 1)
+	go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }()
+
+	select {
+	case e := <-dispatched:
+		if e.IPAddress != "192.0.2.10" {
+			t.Fatalf("unexpected event %+v", e)
+		}
+	case <-time.After(reassemblerTimeout / 2):
+		t.Fatalf("the connect was held behind the BPF record")
+	}
+
+	close(release)
+	if err := <-result; !errors.Is(err, syscall.EBADF) {
+		t.Fatalf("receive should stop when the client is closed, got %v", err)
+	}
+}