diff --git a/procmon.go b/procmon.go index e84c5a2..1cb35b4 100644 --- a/procmon.go +++ b/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 diff --git a/procmon_linux.go b/procmon_linux.go index 4730ce1..7eb189b 100644 --- a/procmon_linux.go +++ b/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) } } diff --git a/procmon_linux_test.go b/procmon_linux_test.go index a6189fe..b57aab3 100644 --- a/procmon_linux_test.go +++ b/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)) + } +} diff --git a/procmon_test.go b/procmon_test.go index c110e86..403f585 100644 --- a/procmon_test.go +++ b/procmon_test.go @@ -1,7 +1,13 @@ package main import ( + "errors" + "fmt" + "os" + "syscall" "testing" + + "github.com/elastic/go-libaudit/v2/auparse" ) func TestProcessMonitor_PrepareEvent(t *testing.T) { @@ -64,3 +70,251 @@ func TestProcessMonitor_PrepareEvent(t *testing.T) { t.Errorf("Event ready expected") } } + +func parseAuditRecord(t *testing.T, msgType auparse.AuditMessageType, text string) *auparse.AuditMessage { + t.Helper() + message, err := auparse.Parse(msgType, text) + if err != nil { + t.Fatalf("parse %v: %v", msgType, err) + } + return message +} + +func TestHandleRecordRetiresSendWithoutAddressOnEOE(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + + // send() on a connected socket: sendto with no SOCKADDR record + records := []struct { + msgType auparse.AuditMessageType + text string + }{ + {auparse.AUDIT_SYSCALL, `audit(1700000000.123:42): arch=c000003e syscall=44 success=yes exit=121 a0=3 a1=7f0000000000 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"`}, + {auparse.AUDIT_PROCTITLE, `audit(1700000000.123:42): proctitle=707974686F6E33`}, + } + for _, r := range records { + if event := p.handleRecord(r.msgType, parseAuditRecord(t, r.msgType, r.text)); event != nil { + t.Fatalf("send without an address must not be dispatched, got %+v", event) + } + } + if event, ok := p.Events[42]; !ok || event.EventType != netMonitorTag || event.Syscall != "sendto" { + t.Fatalf("expected a pending sendto event before EOE, got %+v", p.Events[42]) + } + + p.handleRecord(auparse.AUDIT_EOE, parseAuditRecord(t, auparse.AUDIT_EOE, `audit(1700000000.123:42): `)) + if len(p.Events) != 0 { + t.Fatalf("expected no pending events after EOE, got %d", len(p.Events)) + } +} + +func TestHandleRecordDispatchesConnectBeforeEOE(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + + syscallRecord := `audit(1700000000.456:43): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd00000000 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"` + if event := p.handleRecord(auparse.AUDIT_SYSCALL, parseAuditRecord(t, auparse.AUDIT_SYSCALL, syscallRecord)); event != nil { + t.Fatalf("connect is not ready before its address, got %+v", event) + } + + // 192.0.2.10:443 + sockaddr := `audit(1700000000.456:43): saddr=020001BBC000020A0000000000000000` + event := p.handleRecord(auparse.AUDIT_SOCKADDR, parseAuditRecord(t, auparse.AUDIT_SOCKADDR, sockaddr)) + if event == nil { + t.Fatalf("expected connect to be dispatched once its address arrives") + } + if event.EventType != netMonitorTag || event.Syscall != "connect" || event.IPAddress != "192.0.2.10" || event.Port != "443" || !event.SentForProcessing { + t.Fatalf("unexpected dispatched event %+v", event) + } + if again := p.handleRecord(auparse.AUDIT_PROCTITLE, parseAuditRecord(t, auparse.AUDIT_PROCTITLE, `audit(1700000000.456:43): proctitle=6375726C`)); again != nil { + t.Fatalf("connect must be dispatched only once") + } + + p.handleRecord(auparse.AUDIT_EOE, parseAuditRecord(t, auparse.AUDIT_EOE, `audit(1700000000.456:43): `)) + if len(p.Events) != 0 { + t.Fatalf("expected no pending events after EOE, got %d", len(p.Events)) + } + if event.IPAddress != "192.0.2.10" { + t.Fatalf("retiring the sequence must not change the dispatched event") + } +} + +type auditRecord struct { + msgType auparse.AuditMessageType + text string +} + +// feedRecords passes records through handleRecord and returns the dispatched events. +func feedRecords(t *testing.T, p *ProcessMonitor, records []auditRecord) []*Event { + t.Helper() + var dispatched []*Event + for _, r := range records { + if event := p.handleRecord(r.msgType, parseAuditRecord(t, r.msgType, r.text)); event != nil { + // copy, so later records for the sequence do not hide what was dispatched + snapshot := *event + dispatched = append(dispatched, &snapshot) + } + } + return dispatched +} + +func TestHandleRecordDispatchesDetectionsOnce(t *testing.T) { + tests := []struct { + name string + records []auditRecord + check func(*Event) bool + }{ + { + name: "process start", + records: []auditRecord{ + {auparse.AUDIT_SYSCALL, `audit(1700000001.000:50): arch=c000003e syscall=59 success=yes exit=0 a0=55d0 a1=55d1 a2=55d2 a3=0 items=2 ppid=1 pid=200 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="ls" exe="/usr/bin/ls" key="procmon"`}, + {auparse.AUDIT_EXECVE, `audit(1700000001.000:50): argc=2 a0="ls" a1="-l"`}, + {auparse.AUDIT_CWD, `audit(1700000001.000:50): cwd="/home/runner/work"`}, + {auparse.AUDIT_PATH, `audit(1700000001.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`}, + {auparse.AUDIT_PROCTITLE, `audit(1700000001.000:50): proctitle=6C73002D6C`}, + {auparse.AUDIT_EOE, `audit(1700000001.000:50): `}, + }, + check: func(e *Event) bool { + return e.EventType == processMonitorTag && len(e.ProcessArguments) == 2 && e.Path == "/home/runner/work" + }, + }, + { + name: "file write", + records: []auditRecord{ + {auparse.AUDIT_SYSCALL, `audit(1700000002.000:51): arch=c000003e syscall=257 success=yes exit=3 a0=ffffff9c a1=55d0 a2=241 a3=1b6 items=2 ppid=1 pid=201 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="bash" exe="/usr/bin/bash" key="filemon"`}, + {auparse.AUDIT_CWD, `audit(1700000002.000:51): cwd="/home/runner/work/repo"`}, + {auparse.AUDIT_PATH, `audit(1700000002.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`}, + {auparse.AUDIT_PROCTITLE, `audit(1700000002.000:51): proctitle=62617368`}, + {auparse.AUDIT_EOE, `audit(1700000002.000:51): `}, + }, + check: func(e *Event) bool { + return e.EventType == fileMonitorTag && e.FileName == "out.txt" && e.Path == "/home/runner/work/repo" + }, + }, + { + name: "sendto with an address", + records: []auditRecord{ + {auparse.AUDIT_SYSCALL, `audit(1700000003.000:52): arch=c000003e syscall=44 success=yes exit=40 a0=5 a1=7f00 a2=28 a3=0 items=0 ppid=1 pid=202 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="dig" exe="/usr/bin/dig" key="netmon"`}, + // 192.0.2.53:53 + {auparse.AUDIT_SOCKADDR, `audit(1700000003.000:52): saddr=02000035C0000235000000000000000000`}, + {auparse.AUDIT_PROCTITLE, `audit(1700000003.000:52): proctitle=646967`}, + {auparse.AUDIT_EOE, `audit(1700000003.000:52): `}, + }, + check: func(e *Event) bool { + return e.EventType == netMonitorTag && e.IPAddress == "192.0.2.53" && e.Port == "53" + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + dispatched := feedRecords(t, p, tt.records) + if len(dispatched) != 1 { + t.Fatalf("expected one dispatch, got %d", len(dispatched)) + } + if !tt.check(dispatched[0]) { + t.Fatalf("unexpected dispatched event %+v", dispatched[0]) + } + if len(p.Events) != 0 { + t.Fatalf("expected no pending events after EOE, got %d", len(p.Events)) + } + }) + } +} + +func TestPrepareEventDropsIdleEvents(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + part := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} + + last := 5 * pendingEventWindow + for sequence := 1; sequence <= last; sequence++ { + p.PrepareEvent(sequence, part) + } + + if len(p.Events) > 2*pendingEventWindow+1 { + t.Fatalf("pending events not bounded: %d", len(p.Events)) + } + if p.droppedEvents == 0 { + t.Fatalf("expected dropped events to be counted") + } + if _, ok := p.Events[last]; !ok { + t.Fatalf("the newest event must be kept") + } + if _, ok := p.Events[1]; ok { + t.Fatalf("the oldest event should have been dropped") + } +} + +func TestPrepareEventKeepsActiveEventsDuringSweep(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + p.PrepareEvent(1, map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) + + idle := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} + for sequence := 2; sequence <= 2+4*pendingEventWindow; sequence++ { + p.PrepareEvent(sequence, idle) + if sequence%1000 == 0 { + // sequence 1 keeps receiving records, so it is never idle + p.PrepareEvent(1, map[string]interface{}{}) + } + } + + if p.droppedEvents == 0 { + t.Fatalf("expected the sweep to run") + } + event, ok := p.Events[1] + if !ok || event.EventType != netMonitorTag || event.Syscall != "connect" { + t.Fatalf("an active event lost its data: %+v", event) + } +} + +func TestPrepareEventLateRecordWhileSweeping(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + idle := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} + for sequence := 2; sequence <= 2*pendingEventWindow+1; sequence++ { + p.PrepareEvent(sequence, idle) + } + + // a late record for an unknown sequence triggers the sweep while it is prepared + event := p.PrepareEvent(1, map[string]interface{}{"addr": "192.0.2.10", "port": "443"}) + if p.droppedEvents == 0 { + t.Fatalf("expected the sweep to run") + } + if event == nil || event.IPAddress != "192.0.2.10" || p.Events[1] != event { + t.Fatalf("the event being prepared must survive the sweep, got %+v", event) + } +} + +func TestPrepareEventLosesEventCompletedAfterEviction(t *testing.T) { + p := &ProcessMonitor{Events: make(map[int]*Event)} + p.PrepareEvent(1, map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) + + idle := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} + for sequence := 2; sequence <= 2+3*pendingEventWindow; sequence++ { + p.PrepareEvent(sequence, idle) + } + if _, ok := p.Events[1]; ok { + t.Fatalf("an event idle for longer than the window should be dropped") + } + + // documented trade-off: the late address alone does not make the event ready + event := p.PrepareEvent(1, map[string]interface{}{"addr": "192.0.2.10", "port": "443"}) + if isEventReady(event) { + t.Fatalf("a recreated event without its syscall record must not be dispatched: %+v", event) + } +} + +func TestIsReceiveOverflow(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {syscall.ENOBUFS, true}, + {os.NewSyscallError("recvfrom", syscall.ENOBUFS), true}, + {fmt.Errorf("receive: %w", syscall.ENOBUFS), true}, + {syscall.EBADF, false}, + {errors.New("no buffer space available"), false}, + } + for _, c := range cases { + if got := isReceiveOverflow(c.err); got != c.want { + t.Errorf("isReceiveOverflow(%v) = %v, want %v", c.err, got, c.want) + } + } +}