From 029d768e12b0091dfbe597b4d810ecbaef4fb9db Mon Sep 17 00:00:00 2001 From: venkata-stepsecurity Date: Tue, 27 Oct 2026 19:05:54 -0700 Subject: [PATCH] fix(procmon): retire audit events on EOE and keep receiving on overflow 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. On a workload that streams over localhost TCP the agent grew by about 1.4 GiB per minute until the runner ran out of memory (harden-runner#709). Remove an event when its EOE record arrives. Events that never get one are dropped once 8192 records arrive without touching them, which also bounds the map. When the netlink socket overflows (ENOBUFS), count the loss and keep receiving instead of stopping audit monitoring for the rest of the job, and log if the receive loop does stop. --- procmon.go | 109 +++++++++++++++--- procmon_linux.go | 16 +-- procmon_linux_test.go | 63 +++++++++++ procmon_test.go | 254 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 419 insertions(+), 23 deletions(-) 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) + } + } +}