From efd33daf4bdaa42bc8170e7a7058cc21fb6b2e96 Mon Sep 17 00:00:00 2001 From: venkata-stepsecurity Date: Thu, 29 Oct 2026 19:05:38 -0700 Subject: [PATCH] fix(procmon): only reassemble records of audited syscall events go-libaudit's reassembler delivers events in sequence order. Records that are not part of an audited syscall, such as BPF program loads and unloads or ANOM_PROMISCUOUS, can arrive alone without a record that completes an event, so each one held every later event until it timed out (up to 2 s, or until 50 events were in flight). Handlers that look up the parent process in /proc can miss it once it has exited. Pass a record to the reassembler only if it is a SYSCALL record or belongs to an event whose SYSCALL record was passed. Every record the agent uses follows the SYSCALL record. Records that arrive after their event was delivered are dropped instead of starting a new event. Pushing, maintenance and the final flush take the same lock, so an event cannot be given up between a record being admitted and being pushed. --- procmon.go | 62 ++++++++++++- procmon_linux.go | 16 ++-- procmon_linux_test.go | 31 +++++++ procmon_test.go | 205 ++++++++++++++++++++++++++++++++++++++---- 4 files changed, 292 insertions(+), 22 deletions(-) diff --git a/procmon.go b/procmon.go index 19b840a..696ac44 100644 --- a/procmon.go +++ b/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 diff --git a/procmon_linux.go b/procmon_linux.go index e985016..d0ec761 100644 --- a/procmon_linux.go +++ b/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,15 +220,18 @@ 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 { @@ -235,7 +239,7 @@ func maintainReassembler(r *libaudit.Reassembler, done <-chan struct{}) { case <-done: return case <-ticker.C: - if err := r.Maintain(); err != nil { + if err := s.maintain(r); err != nil { return } } diff --git a/procmon_linux_test.go b/procmon_linux_test.go index 7c61132..3e18aca 100644 --- a/procmon_linux_test.go +++ b/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) + } +} diff --git a/procmon_test.go b/procmon_test.go index bd390ab..5e8ccf0 100644 --- a/procmon_test.go +++ b/procmon_test.go @@ -83,12 +83,15 @@ func newTestReassembler(t *testing.T, maxInFlight int, timeout time.Duration) (* return r, stream, rec, p } -func pushRecords(t *testing.T, r *libaudit.Reassembler, records []auditRecord) { +// pushRecords passes records to the reassembler the way receive does. +func pushRecords(t *testing.T, r *libaudit.Reassembler, stream *countingStream, records []auditRecord) { t.Helper() for _, rec := range records { - if err := r.Push(rec.msgType, []byte(rec.text)); err != nil { - t.Fatalf("push %v: %v", rec.msgType, err) + msg, err := auparse.Parse(rec.msgType, rec.text) + if err != nil { + t.Fatalf("parse %v: %v", rec.msgType, err) } + stream.push(r, msg) } } @@ -200,8 +203,8 @@ func TestReassemblyDispatchesDetectionsOnce(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - r, _, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) - pushRecords(t, r, tt.records) + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, stream, tt.records) // delivered when the event completes, before any timeout or flush got := rec.dispatched() @@ -219,11 +222,183 @@ func TestReassemblyDispatchesDetectionsOnce(t *testing.T) { } } +func TestReassemblySkipsRecordsOutsideSyscallEvents(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + pushRecords(t, r, stream, []auditRecord{ + // a BPF program unload, logged on its own, with no record that completes it + {auparse.AUDIT_BPF, `audit(1700000000.000:30): prog-id=12 op=UNLOAD`}, + // a BPF record logged during an audited syscall, before its SYSCALL record + {auparse.AUDIT_BPF, `audit(1700000000.000:31): prog-id=13 op=LOAD`}, + syscallRecord(31, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:31): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(31), + eoeRecord(31), + syscallRecord(32, 59, "ls", "/usr/bin/ls", "procmon"), + {auparse.AUDIT_EXECVE, `audit(1700000000.000:32): argc=1 a0="ls"`}, + {auparse.AUDIT_CWD, `audit(1700000000.000:32): cwd="/home/runner/work"`}, + proctitleRecord(32), + eoeRecord(32), + }) + + // both are delivered as they complete, not held behind the BPF record + got := rec.dispatched() + if len(got) != 2 || got[0].IPAddress != "192.0.2.10" || got[1].Exe != "/usr/bin/ls" { + t.Fatalf("expected the connect and the process start right away, got %+v", got) + } + if delivered := stream.delivered.Load(); delivered != 2 { + t.Fatalf("records outside syscall events must not reach the reassembler, %d events delivered", delivered) + } + stream.mu.Lock() + defer stream.mu.Unlock() + if len(stream.syscallEvents) != 0 { + t.Fatalf("delivered events must be forgotten, %d left", len(stream.syscallEvents)) + } +} + +func TestReassemblyCompletesOnEOEWithoutProctitle(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(34, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:34): saddr=020001BBC000020A0000000000000000`}, + eoeRecord(34), + }) + + if got := rec.dispatched(); len(got) != 1 || got[0].IPAddress != "192.0.2.10" { + t.Fatalf("expected the connect to be dispatched on its EOE, got %+v", got) + } +} + +func TestReassemblyDropsRecordsAfterTheEventIsDelivered(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, 50*time.Millisecond) + + // a connect given up before its last records arrive + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(33, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:33): saddr=020001BBC000020A0000000000000000`}, + }) + time.Sleep(100 * time.Millisecond) + if err := r.Maintain(); err != nil { + t.Fatalf("maintain: %v", err) + } + pushRecords(t, r, stream, []auditRecord{proctitleRecord(33), eoeRecord(33)}) + r.Close() + + if delivered := stream.delivered.Load(); delivered != 1 { + t.Fatalf("late records must not start a new event, %d events delivered", delivered) + } + if got := rec.dispatched(); len(got) != 1 { + t.Fatalf("expected the connect to be dispatched once, got %+v", got) + } +} + +// blockingStream holds the delivery of one event until released. +type blockingStream struct { + *auditStream + seq uint32 + entered chan struct{} + release chan struct{} + enterOne sync.Once +} + +func (b *blockingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + if msgs[0].Sequence == b.seq { + b.enterOne.Do(func() { close(b.entered) }) + <-b.release + } + b.auditStream.ReassemblyComplete(msgs) +} + +func TestLateRecordWaitsForTheEventToBeGivenUp(t *testing.T) { + rec := &recordingDispatch{} + stream := &auditStream{monitor: &ProcessMonitor{}, dispatch: rec.dispatch} + blocking := &blockingStream{auditStream: stream, seq: 40, entered: make(chan struct{}), release: make(chan struct{})} + r, err := libaudit.NewReassembler(reassemblerMaxInFlight, 50*time.Millisecond, blocking) + if err != nil { + t.Fatalf("new reassembler: %v", err) + } + push := func(records ...auditRecord) { + for _, record := range records { + msg, err := auparse.Parse(record.msgType, record.text) + if err != nil { + t.Errorf("parse %v: %v", record.msgType, err) + return + } + stream.push(r, msg) + } + } + + var releaseOnce sync.Once + releaseCallback := func() { releaseOnce.Do(func() { close(blocking.release) }) } + maintained := make(chan struct{}) + t.Cleanup(func() { + releaseCallback() + <-maintained + }) + + // a connect whose address arrives just as it is given up + push(syscallRecord(40, 42, "curl", "/usr/bin/curl", "netmon")) + time.Sleep(100 * time.Millisecond) + go func() { + defer close(maintained) + stream.maintain(r) + }() + select { + case <-blocking.entered: + case <-time.After(5 * time.Second): + t.Fatalf("maintenance did not give up the event") + } + + pushed := make(chan struct{}) + go func() { + push(auditRecord{auparse.AUDIT_SOCKADDR, `audit(1700000000.000:40): saddr=020001BBC000020A0000000000000000`}) + close(pushed) + }() + select { + case <-pushed: + t.Fatalf("a record was pushed while its event was being given up") + case <-time.After(100 * time.Millisecond): + } + releaseCallback() + select { + case <-pushed: + case <-time.After(5 * time.Second): + t.Fatalf("the late record was not pushed after the event was given up") + } + + // the late record did not start a new event that holds this one back + push( + syscallRecord(41, 42, "curl", "/usr/bin/curl", "netmon"), + auditRecord{auparse.AUDIT_SOCKADDR, `audit(1700000000.000:41): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(41), + ) + got := rec.dispatched() + if len(got) != 1 || got[0].Pid != "141" { + t.Fatalf("expected only the second connect to be dispatched, right away, got %+v", got) + } +} + +func TestSyscallEventsAreBounded(t *testing.T) { + r, stream, _, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + for seq := 1; seq <= 10000; seq++ { + pushRecords(t, r, stream, []auditRecord{syscallRecord(seq, 44, "python3", "/usr/bin/python3.12", "netmon")}) + } + + stream.mu.Lock() + defer stream.mu.Unlock() + // at most the events the reassembler holds, plus the one being added + if n := len(stream.syscallEvents); n > reassemblerMaxInFlight+1 { + t.Fatalf("expected at most %d tracked events, got %d", reassemblerMaxInFlight+1, n) + } +} + func TestReassemblyDropsSendWithoutAddress(t *testing.T) { r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) // send() on a connected socket: sendto with no SOCKADDR record - pushRecords(t, r, []auditRecord{syscallRecord(42, 44, "python3", "/usr/bin/python3.12", "netmon"), proctitleRecord(42), eoeRecord(42)}) + pushRecords(t, r, stream, []auditRecord{syscallRecord(42, 44, "python3", "/usr/bin/python3.12", "netmon"), proctitleRecord(42), eoeRecord(42)}) if stream.delivered.Load() != 1 { t.Fatalf("expected the event to be delivered once, got %d", stream.delivered.Load()) @@ -239,7 +414,7 @@ func TestReassemblyDoesNotKeepIncompleteEvents(t *testing.T) { // records that never complete (no PROCTITLE or EOE) must not pile up const events = 10000 for seq := 1; seq <= events; seq++ { - pushRecords(t, r, []auditRecord{syscallRecord(seq, 44, "python3", "/usr/bin/python3.12", "netmon")}) + pushRecords(t, r, stream, []auditRecord{syscallRecord(seq, 44, "python3", "/usr/bin/python3.12", "netmon")}) } // 50 is Auditbeat's default; compared as a literal so a change to the limit is caught @@ -255,7 +430,7 @@ func TestReassemblyGivesUpOnEventsAfterTimeout(t *testing.T) { r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, 50*time.Millisecond) // a connect whose last records never arrive - pushRecords(t, r, []auditRecord{ + pushRecords(t, r, stream, []auditRecord{ syscallRecord(60, 42, "curl", "/usr/bin/curl", "netmon"), {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:60): saddr=020001BBC000020A0000000000000000`}, }) @@ -277,11 +452,11 @@ func TestReassemblyGivesUpOnEventsAfterTimeout(t *testing.T) { } func TestReassemblyCountsLostEvents(t *testing.T) { - r, _, _, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + r, stream, _, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) // sequences 71 to 74 never arrive - pushRecords(t, r, []auditRecord{syscallRecord(70, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(70)}) - pushRecords(t, r, []auditRecord{syscallRecord(75, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(75)}) + pushRecords(t, r, stream, []auditRecord{syscallRecord(70, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(70)}) + pushRecords(t, r, stream, []auditRecord{syscallRecord(75, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(75)}) if got := p.sequenceGaps.Load(); got != 4 { t.Fatalf("expected a gap of 4, got %d", got) @@ -306,10 +481,10 @@ func TestEventsLostIgnoresImpossibleGaps(t *testing.T) { } func TestReassemblyCountsRecordFieldErrors(t *testing.T) { - r, _, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + r, stream, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) // a valid header with an address that cannot be decoded - pushRecords(t, r, []auditRecord{ + pushRecords(t, r, stream, []auditRecord{ syscallRecord(90, 42, "curl", "/usr/bin/curl", "netmon"), {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:90): saddr=ZZZZ`}, proctitleRecord(90), @@ -328,8 +503,8 @@ func TestReassemblyHandlesShortSocketAddresses(t *testing.T) { // too short for the family, for AF_INET and for AF_INET6 for _, saddr := range []string{"ZZ", "02", "", "0200", "02000050", "0A00", "0A0001BB"} { t.Run(fmt.Sprintf("saddr=%q", saddr), func(t *testing.T) { - r, _, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) - pushRecords(t, r, []auditRecord{ + r, stream, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, stream, []auditRecord{ syscallRecord(91, 42, "curl", "/usr/bin/curl", "netmon"), {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:91): saddr=` + saddr}, proctitleRecord(91),