From ad2fe96dc8a34ab638e27b30135340c1c65b634a Mon Sep 17 00:00:00 2001 From: venkata-stepsecurity Date: Wed, 28 Oct 2026 13:54:38 -0700 Subject: [PATCH 1/3] fix(procmon): group audit records with go-libaudit's reassembler Audit events were kept in ProcessMonitor.Events forever. send() on a connected socket is audited as sendto without an address, so those events never became ready and were never released, and the agent grew by about 1.4 GiB per minute on a localhost-streaming workload (harden-runner#709). Under the same load the netlink socket overflowed (ENOBUFS), receive returned and audit monitoring stopped for the rest of the job. Use go-libaudit's Reassembler, which Auditbeat uses, with Auditbeat's defaults: at most 50 events in flight, a 2 second timeout and maintenance every 500 ms. Every event is delivered once, complete or evicted, so nothing is kept after delivery. The first qualifying PATH is kept, as before. Like Auditbeat, receive stops only when the client is closed: other errors, including ENOBUFS, are counted and receiving continues, and records with an invalid header are dropped. Sequence gaps and records whose fields cannot be parsed are counted and logged on powers of two. Bump go-libaudit from v2.3.2 to v2.6.2: v2.3.2 panics in parseSockaddr on a SOCKADDR record shorter than expected (elastic/go-libaudit#152), which stops the agent. This replaces the event map and idle sweep from #501, reverted in the previous commit. (cherry picked from commit 018a9e6dbf8ecb1d135b90f411eb7eec79bb2afd) --- go.mod | 2 +- go.sum | 2 + procmon.go | 150 +++++++++++----- procmon_linux.go | 67 +++++-- procmon_linux_test.go | 186 +++++++++++++++++++ procmon_test.go | 405 +++++++++++++++++++++++++++++++++++++----- 6 files changed, 710 insertions(+), 102 deletions(-) diff --git a/go.mod b/go.mod index 613714b..fd019f3 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.27.3 require ( github.com/coreos/go-iptables v0.6.0 - github.com/elastic/go-libaudit/v2 v2.3.2 + github.com/elastic/go-libaudit/v2 v2.6.2 github.com/florianl/go-nflog/v2 v2.0.1 github.com/google/gopacket v1.1.19 github.com/jarcoal/httpmock v1.3.0 diff --git a/go.sum b/go.sum index 6c04225..9bbefa6 100644 --- a/go.sum +++ b/go.sum @@ -45,6 +45,8 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/elastic/go-libaudit/v2 v2.3.2 h1:qWNcA3nkwNEGh1UBDbDTVF55KR6SM1W2Ji1LGDqFEpw= github.com/elastic/go-libaudit/v2 v2.3.2/go.mod h1:+ZE0czqmbqtnRkl0fNgpI+HvVVRo/ZMJdcXv/PaKcOo= +github.com/elastic/go-libaudit/v2 v2.6.2 h1:1PM6wVBTJHJQYsKl8jfA9/Aw9pFty5uUezPiUfKtOI4= +github.com/elastic/go-libaudit/v2 v2.6.2/go.mod h1:8205nkf2oSrXFlO4H5j8/cyVMoSF3Y7jt+FjgS4ubQU= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/florianl/go-nflog/v2 v2.0.1 h1:8csWIqFQ2vDaZJkxezY3dXDB7bEFg0VRFsYd2Bzj3II= diff --git a/procmon.go b/procmon.go index e84c5a2..19b840a 100644 --- a/procmon.go +++ b/procmon.go @@ -1,10 +1,15 @@ package main import ( + "errors" "fmt" + "math/bits" "strconv" - "sync" + "sync/atomic" + "syscall" "time" + + "github.com/elastic/go-libaudit/v2/auparse" ) const ( @@ -13,6 +18,15 @@ const ( processMonitorTag = "procmon" ) +// Audit records are grouped into events by go-libaudit's reassembler, with the +// same limits Auditbeat uses by default: at most 50 events in flight, and an +// event without its last record is given up after 2 seconds. +const ( + reassemblerMaxInFlight = 50 + reassemblerTimeout = 2 * time.Second + reassemblerMaintain = 500 * time.Millisecond +) + type ProcessMonitor struct { CorrelationId string Repo string @@ -20,8 +34,15 @@ type ProcessMonitor struct { DNSProxy *DNSProxy WorkingDirectory string DisableFileMonitoring bool - Events map[int]*Event - mutex sync.RWMutex + + // dispatchEvent replaces the event handler in tests + dispatchEvent func(*Event) + + sequenceGaps atomic.Int64 + receiveOverflow atomic.Int64 + receiveErrors atomic.Int64 + invalidRecords atomic.Int64 + recordErrors atomic.Int64 } type Process struct { @@ -36,61 +57,54 @@ type Process struct { } type Event struct { - FileName string - Path string - Syscall string - Exe string - IPAddress string - Port string - Pid string - ProcessArguments []string - PPid string - Euid string - Timestamp time.Time - EventType string - Status string - SentForProcessing bool + FileName string + Path string + Syscall string + Exe string + IPAddress string + Port string + Pid string + ProcessArguments []string + PPid string + Euid string + Timestamp time.Time + EventType string + Status string } -func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interface{}) { - p.mutex.Lock() - _, found := p.Events[sequence] - if !found { - p.Events[sequence] = &Event{} - } - +func (event *Event) addRecord(eventMap map[string]interface{}) { // Tags value, found := eventMap["tags"] 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 +117,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)])) } } @@ -112,12 +126,11 @@ 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) + // the first qualifying PATH wins, as when events were dispatched on it + if (nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL") && event.FileName == "" { + event.FileName = getValue("name", eventMap) } } - - p.mutex.Unlock() } func getValue(key string, eventMap map[string]interface{}) string { @@ -129,17 +142,62 @@ func getValue(key string, eventMap map[string]interface{}) string { return "" } -func (p *ProcessMonitor) markEventSent(event *Event) { - p.mutex.Lock() - event.SentForProcessing = true - p.mutex.Unlock() +// auditStream receives complete events from the reassembler and dispatches +// the ones the event handler uses. Every event is delivered once, including +// incomplete ones that are evicted, so nothing is kept after delivery. +type auditStream struct { + monitor *ProcessMonitor + dispatch func(*Event) } -func isEventReady(event *Event) bool { - if event.SentForProcessing { +func (s *auditStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + event := &Event{} + for _, msg := range msgs { + eventMap := msg.ToMapStr() + if parseErr, found := eventMap["error"]; found { + noteLoss(&s.monitor.recordErrors, 1, fmt.Sprintf("audit record fields not parsed: %v", parseErr)) + } + event.addRecord(eventMap) + } + if isEventReady(event) { + s.dispatch(event) + } +} + +// 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. +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 + if count <= 0 || count > 1<<24 { + return + } + noteLoss(&s.monitor.sequenceGaps, int64(count), "gaps in audit sequence numbers") +} + +// noteReceiveError counts a failed receive. A full netlink socket (ENOBUFS) +// means the kernel dropped messages; the socket stays usable either way. +func (p *ProcessMonitor) noteReceiveError(err error) { + if errors.Is(err, syscall.ENOBUFS) { + noteLoss(&p.receiveOverflow, 1, "audit netlink buffer overflowed, kernel dropped messages") + return + } + noteLoss(&p.receiveErrors, 1, fmt.Sprintf("audit receive error: %v", err)) +} + +// noteLoss adds n to the counter and logs when the total reaches the next +// power of two, so a sustained loss does not flood the log. It reports whether +// it logged. +func noteLoss(counter *atomic.Int64, n int64, what string) bool { + total := counter.Add(n) + if bits.Len64(uint64(total)) == bits.Len64(uint64(total-n)) { return false } + WriteLog(fmt.Sprintf("[procmon] %s, total=%d", what, total)) + return true +} +func isEventReady(event *Event) bool { switch event.EventType { case netMonitorTag: if event.IPAddress != "" && event.Port != "" { diff --git a/procmon_linux.go b/procmon_linux.go index 4730ce1..e985016 100644 --- a/procmon_linux.go +++ b/procmon_linux.go @@ -10,6 +10,9 @@ import ( "io/ioutil" "os" "strings" + "sync" + "syscall" + "time" "github.com/elastic/go-libaudit/v2" "github.com/elastic/go-libaudit/v2/auparse" @@ -151,7 +154,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 { @@ -164,7 +169,6 @@ func fileMonitorAuditRule(workingDirectory, goarch string) string { func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error { - p.Events = make(map[int]*Event) eventHandler := EventHandler{CorrelationId: p.CorrelationId, Repo: p.Repo, ApiClient: p.ApiClient, DNSProxy: p.DNSProxy} eventHandler.ProcessConnectionMap = make(map[string]bool) eventHandler.ProcessFileMap = make(map[string]bool) @@ -172,10 +176,41 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error { eventHandler.ProcessMap = make(map[string]*Process) eventHandler.FileOverwriteCounterMap = make(map[string]int) + dispatch := p.dispatchEvent + if dispatch == nil { + dispatch = func(event *Event) { go eventHandler.HandleEvent(event) } + } + + reassembler, err := libaudit.NewReassembler(reassemblerMaxInFlight, reassemblerTimeout, &auditStream{monitor: p, dispatch: dispatch}) + if err != nil { + return errors.Wrap(err, "create reassembler") + } + done := make(chan struct{}) + var maintaining sync.WaitGroup + maintaining.Add(1) + go func() { + defer maintaining.Done() + maintainReassembler(reassembler, done) + }() + // stop maintenance before the final flush, so no callback runs after return + defer func() { + close(done) + maintaining.Wait() + reassembler.Close() + }() + for { rawEvent, err := r.Receive(false) if err != nil { - return errors.Wrap(err, "receive failed") + // EBADF means the client was closed + if errors.Is(err, syscall.EBADF) { + return errors.Wrap(err, "receive failed") + } + p.noteReceiveError(err) + if !errors.Is(err, syscall.ENOBUFS) { + time.Sleep(10 * time.Millisecond) // do not spin on a persistent error + } + continue } // Messages from 1300-2999 are valid audit messages. @@ -184,18 +219,26 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error { continue } - message, err := auparse.Parse(rawEvent.Type, string(rawEvent.Data)) - if err != nil { - return errors.Wrap(err, "parse failed.") + if err := reassembler.Push(rawEvent.Type, rawEvent.Data); err != nil { + noteLoss(&p.invalidRecords, 1, fmt.Sprintf("dropped audit record with an invalid header: %v", err)) } - 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)]) +// 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{}) { + ticker := time.NewTicker(reassemblerMaintain) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + if err := r.Maintain(); err != nil { + return + } } - } } diff --git a/procmon_linux_test.go b/procmon_linux_test.go index a6189fe..7c61132 100644 --- a/procmon_linux_test.go +++ b/procmon_linux_test.go @@ -4,8 +4,16 @@ package main import ( + "errors" + "fmt" "os" + "sync" + "syscall" "testing" + "time" + + "github.com/elastic/go-libaudit/v2" + "github.com/elastic/go-libaudit/v2/auparse" ) func Test_fileMonitorAuditRule(t *testing.T) { @@ -65,3 +73,181 @@ 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 + block chan struct{} // Receive waits on it, then reports the client closed +} + +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.block != nil { + <-step.block + return nil, syscall.EBADF + } + 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) SendNoWait(syscall.NetlinkMessage) (uint32, error) { return 0, nil } +func (f *fakeNetlink) Close() error { return nil } + +func TestReceiveKeepsGoingUntilTheClientIsClosed(t *testing.T) { + netlink := &fakeNetlink{steps: []fakeNetlinkStep{ + {err: syscall.ENOBUFS}, + {err: syscall.EINVAL}, + {msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:43): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=101 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:43): saddr=020001BBC000020A0000000000000000`}, + {msgType: auparse.AUDIT_CWD, text: `not an audit record`}, + {err: syscall.ENOBUFS}, + {msgType: auparse.AUDIT_PROCTITLE, text: `audit(1700000000.123:43): proctitle=6375726C`}, + {msgType: auparse.AUDIT_EOE, text: `audit(1700000000.123:43): `}, + }} + var mu sync.Mutex + var dispatched []Event + p := &ProcessMonitor{dispatchEvent: func(e *Event) { + mu.Lock() + dispatched = append(dispatched, *e) + mu.Unlock() + }} + + err := p.receive(&libaudit.AuditClient{Netlink: netlink}) + + if !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop only when the client is closed, got %v", err) + } + if len(netlink.steps) != 0 { + t.Fatalf("not every message was read: %d left", len(netlink.steps)) + } + if got := p.receiveOverflow.Load(); got != 2 { + t.Fatalf("expected 2 overflows counted, got %d", got) + } + if got := p.receiveErrors.Load(); got != 1 { + t.Fatalf("expected 1 other receive error counted, got %d", got) + } + if got := p.invalidRecords.Load(); got != 1 { + t.Fatalf("expected the unparsable record to be dropped, got %d", got) + } + mu.Lock() + defer mu.Unlock() + if len(dispatched) != 1 || dispatched[0].IPAddress != "192.0.2.10" || dispatched[0].Port != "443" { + t.Fatalf("expected the connect to be dispatched once, got %+v", dispatched) + } +} + +func TestReceiveDeliversTimedOutEventsWhileIdle(t *testing.T) { + connect := func(seq int, last bool) []fakeNetlinkStep { + steps := []fakeNetlinkStep{ + {msgType: auparse.AUDIT_SYSCALL, text: fmt.Sprintf(`audit(1700000000.123:%d): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=%d 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"`, seq, seq)}, + {msgType: auparse.AUDIT_SOCKADDR, text: fmt.Sprintf(`audit(1700000000.123:%d): saddr=020001BBC000020A0000000000000000`, seq)}, + } + if last { + steps = append(steps, fakeNetlinkStep{msgType: auparse.AUDIT_PROCTITLE, text: fmt.Sprintf(`audit(1700000000.123:%d): proctitle=6375726C`, seq)}) + } + return steps + } + release := make(chan struct{}) + var steps []fakeNetlinkStep + steps = append(steps, connect(79, true)...) // complete, delivered right away + steps = append(steps, connect(80, false)...) // never completes + steps = append(steps, connect(81, true)...) // complete, but behind 80 + steps = append(steps, fakeNetlinkStep{block: release}) + netlink := &fakeNetlink{steps: steps} + + var mu sync.Mutex + var pids []string + p := &ProcessMonitor{dispatchEvent: func(e *Event) { + mu.Lock() + pids = append(pids, e.Pid) + mu.Unlock() + }} + delivered := func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), pids...) + } + + result := make(chan error, 1) + go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }() + + // 79 completes on its PROCTITLE, well before the timeout + completeBy := time.Now().Add(reassemblerTimeout / 2) + for len(delivered()) < 1 && time.Now().Before(completeBy) { + time.Sleep(10 * time.Millisecond) + } + if got := delivered(); fmt.Sprint(got) != "[79]" { + t.Fatalf("expected 79 to be delivered when it completed, got %v", got) + } + + // while receive is blocked, only the maintenance goroutine can deliver 80 and 81 + deadline := time.Now().Add(reassemblerTimeout + 2*time.Second) + for len(delivered()) < 3 && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + if got := delivered(); fmt.Sprint(got) != "[79 80 81]" { + t.Fatalf("expected 79, then 80 and 81 after the timeout, got %v", got) + } + + close(release) + if err := <-result; !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop when the client is closed, got %v", err) + } + if got := delivered(); len(got) != 3 { + t.Fatalf("no event should be delivered twice, got %v", got) + } +} + +func TestReceiveWaitsForMaintenanceBeforeReturning(t *testing.T) { + release := make(chan struct{}) + netlink := &fakeNetlink{steps: []fakeNetlinkStep{ + // an incomplete connect that only the maintenance goroutine can deliver + {msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:95): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=95 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:95): saddr=020001BBC000020A0000000000000000`}, + {block: release}, + }} + inCallback := make(chan struct{}) + finishCallback := make(chan struct{}) + p := &ProcessMonitor{dispatchEvent: func(*Event) { + close(inCallback) + <-finishCallback + }} + + result := make(chan error, 1) + go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }() + + select { + case <-inCallback: + case <-time.After(reassemblerTimeout + 2*time.Second): + t.Fatalf("maintenance did not deliver the timed out event") + } + + // the client closes while the maintenance callback is still running + close(release) + select { + case err := <-result: + t.Fatalf("receive returned (%v) while a maintenance callback was running", err) + case <-time.After(200 * time.Millisecond): + } + + close(finishCallback) + 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 c110e86..bd390ab 100644 --- a/procmon_test.go +++ b/procmon_test.go @@ -1,66 +1,385 @@ package main import ( + "errors" + "fmt" + "os" + "reflect" + "sync" + "sync/atomic" + "syscall" "testing" + "time" + + "github.com/elastic/go-libaudit/v2" + "github.com/elastic/go-libaudit/v2/auparse" ) -func TestProcessMonitor_PrepareEvent(t *testing.T) { - processMonitor := &ProcessMonitor{} - processMonitor.Events = make(map[int]*Event) +func TestEvent_addRecord(t *testing.T) { + netEvent := &Event{} + netEvent.addRecord(map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) + netEvent.addRecord(map[string]interface{}{"addr": "2.2.2.2", "port": "443"}) - netEventPart1 := make(map[string]interface{}) - netEventPart1["tags"] = []string{"netmon"} - netEventPart1["syscall"] = "connect" + if !isEventReady(netEvent) { + t.Errorf("Event ready expected") + } - netEventPart2 := make(map[string]interface{}) - netEventPart2["addr"] = "2.2.2.2" - netEventPart2["port"] = "443" + fileEvent := &Event{} + fileEvent.addRecord(map[string]interface{}{"tags": []string{"filemon"}, "syscall": "open"}) + fileEvent.addRecord(map[string]interface{}{"nametype": "CREATE", "name": "file1.txt"}) + fileEvent.addRecord(map[string]interface{}{"cwd": "/dir"}) + fileEvent.addRecord(map[string]interface{}{"argc": "2", "a0": "argv1", "a1": "argv2"}) + // argc that cannot be converted to an int is logged and ignored + fileEvent.addRecord(map[string]interface{}{"argc": []string{"failed"}}) + + if !isEventReady(fileEvent) { + t.Errorf("Event ready expected") + } +} - processMonitor.PrepareEvent(1, netEventPart1) - processMonitor.PrepareEvent(1, netEventPart2) +type auditRecord struct { + msgType auparse.AuditMessageType + text string +} - isReady := isEventReady(processMonitor.Events[1]) +// recordingDispatch collects the events an auditStream dispatches. +type recordingDispatch struct { + mu sync.Mutex + events []Event +} - if !isReady { - t.Errorf("Event ready expected") +func (r *recordingDispatch) dispatch(event *Event) { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, *event) +} + +func (r *recordingDispatch) dispatched() []Event { + r.mu.Lock() + defer r.mu.Unlock() + return append([]Event(nil), r.events...) +} + +// countingStream wraps an auditStream and counts every delivered event. +type countingStream struct { + *auditStream + delivered atomic.Int64 +} + +func (c *countingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + c.delivered.Add(1) + c.auditStream.ReassemblyComplete(msgs) +} + +func newTestReassembler(t *testing.T, maxInFlight int, timeout time.Duration) (*libaudit.Reassembler, *countingStream, *recordingDispatch, *ProcessMonitor) { + t.Helper() + p := &ProcessMonitor{} + rec := &recordingDispatch{} + stream := &countingStream{auditStream: &auditStream{monitor: p, dispatch: rec.dispatch}} + r, err := libaudit.NewReassembler(maxInFlight, timeout, stream) + if err != nil { + t.Fatalf("new reassembler: %v", err) + } + return r, stream, rec, p +} + +func pushRecords(t *testing.T, r *libaudit.Reassembler, 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) + } } +} - fileEventPart1 := make(map[string]interface{}) - fileEventPart1["tags"] = []string{"filemon"} - fileEventPart1["syscall"] = "open" +func syscallRecord(seq, syscallNr int, comm, exe, key string) auditRecord { + return auditRecord{auparse.AUDIT_SYSCALL, fmt.Sprintf(`audit(1700000000.000:%d): arch=c000003e syscall=%d success=yes exit=0 a0=3 a1=7f00 a2=10 a3=0 items=1 ppid=1 pid=%d auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="%s" exe="%s" key="%s"`, seq, syscallNr, 100+seq, comm, exe, key)} +} - fileEventPart2 := make(map[string]interface{}) - fileEventPart2["nametype"] = "CREATE" - fileEventPart2["name"] = "file1.txt" +func proctitleRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_PROCTITLE, fmt.Sprintf(`audit(1700000000.000:%d): proctitle=7465737400`, seq)} +} - fileEventPart3 := make(map[string]interface{}) - fileEventPart3["cwd"] = "/dir" +func eoeRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_EOE, fmt.Sprintf(`audit(1700000000.000:%d): `, seq)} +} - // Testing procmon.go#93 [Process Start] - fileEventPart4 := make(map[string]interface{}) - fileEventPart4["argc"] = "2" - fileEventPart4["a0"] = "argv1" - fileEventPart4["a1"] = "argv2" +func TestReassemblyDispatchesDetectionsOnce(t *testing.T) { + tests := []struct { + name string + records []auditRecord + want func(Event) bool + }{ + { + name: "connect", + records: []auditRecord{ + syscallRecord(43, 42, "curl", "/usr/bin/curl", "netmon"), + // 192.0.2.10:443 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:43): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(43), + eoeRecord(43), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.Syscall == "connect" && e.IPAddress == "192.0.2.10" && e.Port == "443" && + e.Pid == "143" && e.PPid == "1" && e.Exe == "/usr/bin/curl" && e.Euid == "1001" && e.Status == "success" && + e.Timestamp.Equal(time.Unix(1700000000, 0)) + }, + }, + { + name: "sendto with an address", + records: []auditRecord{ + syscallRecord(52, 44, "dig", "/usr/bin/dig", "netmon"), + // 192.0.2.53:53 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:52): saddr=02000035C0000235000000000000000000`}, + proctitleRecord(52), + eoeRecord(52), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.IPAddress == "192.0.2.53" && e.Port == "53" + }, + }, + { + name: "process start", + records: []auditRecord{ + syscallRecord(50, 59, "ls", "/usr/bin/ls", "procmon"), + {auparse.AUDIT_EXECVE, `audit(1700000000.000:50): argc=2 a0="ls" a1="-l"`}, + {auparse.AUDIT_CWD, `audit(1700000000.000:50): cwd="/home/runner/work"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:50): item=0 name="/usr/bin/ls" inode=10 dev=08:01 mode=0100755 ouid=0 ogid=0 rdev=00:00 nametype=NORMAL cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(50), + eoeRecord(50), + }, + want: func(e Event) bool { + return e.EventType == processMonitorTag && reflect.DeepEqual(e.ProcessArguments, []string{"ls", "-l"}) && e.Path == "/home/runner/work" && + e.Pid == "150" && e.Exe == "/usr/bin/ls" + }, + }, + { + name: "file write", + records: []auditRecord{ + syscallRecord(51, 257, "bash", "/usr/bin/bash", "filemon"), + {auparse.AUDIT_CWD, `audit(1700000000.000:51): cwd="/home/runner/work/repo"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:51): item=1 name="out.txt" inode=11 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=CREATE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(51), + eoeRecord(51), + }, + want: func(e Event) bool { + return e.EventType == fileMonitorTag && e.FileName == "out.txt" && e.Path == "/home/runner/work/repo" + }, + }, + { + name: "connect over IPv6", + records: []auditRecord{ + syscallRecord(54, 42, "curl", "/usr/bin/curl", "netmon"), + // [2001:db8::1]:443 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:54): saddr=0A0001BB0000000020010DB800000000000000000000000100000000`}, + proctitleRecord(54), + eoeRecord(54), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.IPAddress == "2001:db8::1" && e.Port == "443" + }, + }, + { + // rename has PARENT, DELETE and CREATE paths; the first qualifying one is kept + name: "rename keeps the source file", + records: []auditRecord{ + syscallRecord(53, 82, "mv", "/usr/bin/mv", "filemon"), + {auparse.AUDIT_CWD, `audit(1700000000.000:53): cwd="/home/runner/work/repo"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=0 name="/home/runner/work/repo" inode=5 dev=08:01 mode=040755 ouid=1001 ogid=1001 rdev=00:00 nametype=PARENT cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=1 name="/home/runner/work/repo" inode=5 dev=08:01 mode=040755 ouid=1001 ogid=1001 rdev=00:00 nametype=PARENT cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=2 name="Dockerfile" inode=12 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=DELETE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=3 name="tmpfile" inode=12 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=CREATE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(53), + eoeRecord(53), + }, + want: func(e Event) bool { + return e.EventType == fileMonitorTag && e.FileName == "Dockerfile" && e.Path == "/home/runner/work/repo" + }, + }, + } - // In procmon.go [Process Start] section, - // we are converting `argc` to INT. - // Because of below code, we will hit the error, - // in which we are unable to convert argc to INT. - // So, main intention was to cover that error handling block. - fileEventPart5 := make(map[string]interface{}) - fileEventPart5["argc"] = []string{"failed"} - // ===== END ===== + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r, _, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, tt.records) - processMonitor.PrepareEvent(2, fileEventPart1) - processMonitor.PrepareEvent(2, fileEventPart2) - processMonitor.PrepareEvent(2, fileEventPart3) + // delivered when the event completes, before any timeout or flush + got := rec.dispatched() + if len(got) != 1 { + t.Fatalf("expected one dispatch, got %d", len(got)) + } + if !tt.want(got[0]) { + t.Fatalf("unexpected dispatched event %+v", got[0]) + } + r.Close() + if again := rec.dispatched(); len(again) != 1 { + t.Fatalf("the final flush must not dispatch again, got %d", len(again)) + } + }) + } +} - processMonitor.PrepareEvent(2, fileEventPart4) - processMonitor.PrepareEvent(2, fileEventPart5) +func TestReassemblyDropsSendWithoutAddress(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) - isReady = isEventReady(processMonitor.Events[2]) + // 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)}) - if !isReady { - t.Errorf("Event ready expected") + if stream.delivered.Load() != 1 { + t.Fatalf("expected the event to be delivered once, got %d", stream.delivered.Load()) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a send without an address must not be dispatched, got %+v", got) + } +} + +func TestReassemblyDoesNotKeepIncompleteEvents(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // 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")}) + } + + // 50 is Auditbeat's default; compared as a literal so a change to the limit is caught + if inFlight := events - stream.delivered.Load(); inFlight > 50 { + t.Fatalf("expected at most 50 events in flight, got %d", inFlight) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("incomplete sends must not be dispatched, got %d", len(got)) + } +} + +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{ + syscallRecord(60, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:60): saddr=020001BBC000020A0000000000000000`}, + }) + if stream.delivered.Load() != 0 { + t.Fatalf("event delivered before it completed or timed out") + } + + time.Sleep(100 * time.Millisecond) + if err := r.Maintain(); err != nil { + t.Fatalf("maintain: %v", err) + } + + if stream.delivered.Load() != 1 { + t.Fatalf("expected the timed out event to be delivered, got %d", stream.delivered.Load()) + } + if got := rec.dispatched(); len(got) != 1 || got[0].IPAddress != "192.0.2.10" { + t.Fatalf("expected the connect to be dispatched with what arrived, got %+v", got) + } +} + +func TestReassemblyCountsLostEvents(t *testing.T) { + r, _, _, 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)}) + + if got := p.sequenceGaps.Load(); got != 4 { + t.Fatalf("expected a gap of 4, got %d", got) + } +} + +func TestEventsLostIgnoresImpossibleGaps(t *testing.T) { + p := &ProcessMonitor{} + stream := &auditStream{monitor: p, dispatch: func(*Event) {}} + + // what go-libaudit v2.3.2 reports when a record arrives after its event was given up + stream.EventsLost(int(^uint32(0))) + stream.EventsLost(0) + if got := p.sequenceGaps.Load(); got != 0 { + t.Fatalf("impossible gaps must be ignored, got %d", got) + } + + stream.EventsLost(3) + if got := p.sequenceGaps.Load(); got != 3 { + t.Fatalf("expected a gap of 3, got %d", got) + } +} + +func TestReassemblyCountsRecordFieldErrors(t *testing.T) { + r, _, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // a valid header with an address that cannot be decoded + pushRecords(t, r, []auditRecord{ + syscallRecord(90, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:90): saddr=ZZZZ`}, + proctitleRecord(90), + }) + + if got := p.recordErrors.Load(); got != 1 { + t.Fatalf("expected one record with unparsed fields, got %d", got) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a connect without an address must not be dispatched, got %+v", got) + } +} + +func TestReassemblyHandlesShortSocketAddresses(t *testing.T) { + // go-libaudit before v2.6.0 panicked on these (elastic/go-libaudit#152) + // 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{ + syscallRecord(91, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:91): saddr=` + saddr}, + proctitleRecord(91), + }) + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a connect without a usable address must not be dispatched, got %+v", got) + } + if got := p.recordErrors.Load(); got != 1 { + t.Fatalf("expected the record to be counted as not parsed, got %d", got) + } + }) + } +} + +func TestNoteLossLogsOnPowersOfTwo(t *testing.T) { + var counter atomic.Int64 + var logged []int64 + for i := 0; i < 20; i++ { + if noteLoss(&counter, 1, "test") { + logged = append(logged, counter.Load()) + } + } + want := []int64{1, 2, 4, 8, 16} + if fmt.Sprint(logged) != fmt.Sprint(want) { + t.Fatalf("logged at %v, want %v", logged, want) + } + + // a batch that crosses a power of two logs once + if !noteLoss(&counter, 20, "test") { + t.Fatalf("expected a log when the total goes from 20 to 40") + } +} + +func TestNoteReceiveError(t *testing.T) { + cases := []struct { + err error + wantOverflow int64 + wantOther int64 + }{ + {syscall.ENOBUFS, 1, 0}, + {os.NewSyscallError("recvfrom", syscall.ENOBUFS), 1, 0}, + {fmt.Errorf("receive: %w", syscall.ENOBUFS), 1, 0}, + {syscall.EINVAL, 0, 1}, + {errors.New("no buffer space available"), 0, 1}, + } + for _, c := range cases { + p := &ProcessMonitor{} + p.noteReceiveError(c.err) + if p.receiveOverflow.Load() != c.wantOverflow || p.receiveErrors.Load() != c.wantOther { + t.Errorf("noteReceiveError(%v): overflow=%d other=%d, want %d %d", c.err, p.receiveOverflow.Load(), p.receiveErrors.Load(), c.wantOverflow, c.wantOther) + } } } From 068ce96ca84d50540e67e2a479ba1c934db95f20 Mon Sep 17 00:00:00 2001 From: venkata-stepsecurity Date: Wed, 28 Oct 2026 14:00:34 -0700 Subject: [PATCH 2/3] build: add go-libaudit v2.6.2 tool dependency to go.mod and go.sum go-libaudit v2.6.2 pins github.com/elastic/go-licenser in rule/tools.go behind the tools build tag. It is not built into the agent, but go mod vendor needs its checksum. (cherry picked from commit 3ac31ccb3e9c6824d508d61a5396f1d5fd5d93fc) --- go.mod | 1 + go.sum | 11 +++++++++++ 2 files changed, 12 insertions(+) diff --git a/go.mod b/go.mod index fd019f3..be86e5b 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/docker/distribution v2.8.1+incompatible // indirect github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/elastic/go-licenser v0.4.1 // indirect github.com/go-ini/ini v1.67.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect diff --git a/go.sum b/go.sum index 9bbefa6..2db4120 100644 --- a/go.sum +++ b/go.sum @@ -47,6 +47,8 @@ github.com/elastic/go-libaudit/v2 v2.3.2 h1:qWNcA3nkwNEGh1UBDbDTVF55KR6SM1W2Ji1L github.com/elastic/go-libaudit/v2 v2.3.2/go.mod h1:+ZE0czqmbqtnRkl0fNgpI+HvVVRo/ZMJdcXv/PaKcOo= github.com/elastic/go-libaudit/v2 v2.6.2 h1:1PM6wVBTJHJQYsKl8jfA9/Aw9pFty5uUezPiUfKtOI4= github.com/elastic/go-libaudit/v2 v2.6.2/go.mod h1:8205nkf2oSrXFlO4H5j8/cyVMoSF3Y7jt+FjgS4ubQU= +github.com/elastic/go-licenser v0.4.1 h1:1xDURsc8pL5zYT9R29425J3vkHdt4RT5TNEMeRN48x4= +github.com/elastic/go-licenser v0.4.1/go.mod h1:V56wHMpmdURfibNBggaSBfqgPxyT1Tldns1i87iTEvU= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/florianl/go-nflog/v2 v2.0.1 h1:8csWIqFQ2vDaZJkxezY3dXDB7bEFg0VRFsYd2Bzj3II= @@ -212,6 +214,7 @@ github.com/yashtewari/glob-intersection v0.2.0 h1:8iuHdN88yYuCzCdjt0gDe+6bAhUwBe github.com/yashtewari/glob-intersection v0.2.0/go.mod h1:LK7pIC3piUjovexikBbJ26Yml7g8xa5bsjfx2v1fwok= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 h1:RbKq8BG0FI8OiXhBfcRtqqHcZcka+gU3cskNuf05R18= @@ -250,9 +253,12 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= +golang.org/x/lint v0.0.0-20210508222113-6edffad5e616/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro= golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= @@ -270,11 +276,13 @@ golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -297,6 +305,8 @@ golang.org/x/sys v0.0.0-20210216163648-f7da38b97c65/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210305230114-8fe3ee5dd75b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210525143221-35b2ab0089ea/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20211102192858-4dd72447c267/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= @@ -313,6 +323,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo= golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= From b14af87e1d8ebfb9d094796a74356dd15758cffd Mon Sep 17 00:00:00 2001 From: venkata-stepsecurity Date: Thu, 29 Oct 2026 19:05:38 -0700 Subject: [PATCH 3/3] 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. (cherry picked from commit efd33daf4bdaa42bc8170e7a7058cc21fb6b2e96) --- 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),