diff --git a/go.mod b/go.mod index 613714b..be86e5b 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 @@ -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 6c04225..2db4120 100644 --- a/go.sum +++ b/go.sum @@ -45,6 +45,10 @@ 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/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= @@ -210,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= @@ -248,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= @@ -268,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= @@ -295,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= @@ -311,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= diff --git a/procmon.go b/procmon.go index 1cb35b4..19b840a 100644 --- a/procmon.go +++ b/procmon.go @@ -3,8 +3,9 @@ package main import ( "errors" "fmt" + "math/bits" "strconv" - "sync" + "sync/atomic" "syscall" "time" @@ -17,13 +18,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 +// 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 @@ -32,12 +34,15 @@ type ProcessMonitor struct { DNSProxy *DNSProxy WorkingDirectory string DisableFileMonitoring bool - Events map[int]*Event - mutex sync.RWMutex - records int64 - droppedEvents int - receiveOverflow int + // 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 { @@ -52,37 +57,22 @@ 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 - lastRecord int64 + 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{}) *Event { - p.mutex.Lock() - p.records++ - event, found := p.Events[sequence] - if !found { - 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() - } - +func (event *Event) addRecord(eventMap map[string]interface{}) { // Tags value, found := eventMap["tags"] @@ -136,13 +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" { + // 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() - return event } func getValue(key string, eventMap map[string]interface{}) string { @@ -154,69 +142,62 @@ 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() +// 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 (p *ProcessMonitor) dropIdleEventsLocked() { - dropped := 0 - for sequence, event := range p.Events { - if event.lastRecord <= p.records-pendingEventWindow { - delete(p.Events, sequence) - dropped++ +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) } - 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)) +// 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") } -func (p *ProcessMonitor) markEventSent(event *Event) { - p.mutex.Lock() - event.SentForProcessing = true - p.mutex.Unlock() +// 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)) } -func isEventReady(event *Event) bool { - if event.SentForProcessing { +// 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 7eb189b..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" @@ -166,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) @@ -174,14 +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 { - if isReceiveOverflow(err) { - p.noteReceiveOverflow() - continue + // 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 } - return errors.Wrap(err, "receive failed") + continue } // Messages from 1300-2999 are valid audit messages. @@ -190,14 +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 event := p.handleRecord(rawEvent.Type, message); event != nil { - go eventHandler.HandleEvent(event) + 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)) } + } +} +// 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 b57aab3..7c61132 100644 --- a/procmon_linux_test.go +++ b/procmon_linux_test.go @@ -5,9 +5,12 @@ package main import ( "errors" + "fmt" "os" + "sync" "syscall" "testing" + "time" "github.com/elastic/go-libaudit/v2" "github.com/elastic/go-libaudit/v2/auparse" @@ -80,6 +83,7 @@ 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) { @@ -88,6 +92,10 @@ func (f *fakeNetlink) Receive(bool, libaudit.NetlinkParser) ([]syscall.NetlinkMe } 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 } @@ -97,34 +105,149 @@ func (f *fakeNetlink) Receive(bool, libaudit.NetlinkParser) ([]syscall.NetlinkMe }}, nil } -func (f *fakeNetlink) Send(syscall.NetlinkMessage) (uint32, error) { return 0, nil } -func (f *fakeNetlink) Close() error { return 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 TestReceiveContinuesAfterOverflow(t *testing.T) { +func TestReceiveKeepsGoingUntilTheClientIsClosed(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.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:42): proctitle=707974686F6E33`}, - {msgType: auparse.AUDIT_EOE, text: `audit(1700000000.123:42): `}, + {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() }} - 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) + t.Fatalf("receive should stop only when the client is closed, got %v", err) } if len(netlink.steps) != 0 { - t.Fatalf("records after an overflow were not read: %d left", len(netlink.steps)) + 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") } - if p.records != 2 { - t.Fatalf("expected the SYSCALL and PROCTITLE records to be processed, got %d", p.records) + + // 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): } - if len(p.Events) != 0 { - t.Fatalf("expected the event to be retired on EOE, got %d pending", len(p.Events)) + + 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 403f585..bd390ab 100644 --- a/procmon_test.go +++ b/procmon_test.go @@ -4,317 +4,382 @@ 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" - - netEventPart2 := make(map[string]interface{}) - netEventPart2["addr"] = "2.2.2.2" - netEventPart2["port"] = "443" - - processMonitor.PrepareEvent(1, netEventPart1) - processMonitor.PrepareEvent(1, netEventPart2) - - isReady := isEventReady(processMonitor.Events[1]) - - if !isReady { + if !isEventReady(netEvent) { t.Errorf("Event ready expected") } - fileEventPart1 := make(map[string]interface{}) - fileEventPart1["tags"] = []string{"filemon"} - fileEventPart1["syscall"] = "open" - - fileEventPart2 := make(map[string]interface{}) - fileEventPart2["nametype"] = "CREATE" - fileEventPart2["name"] = "file1.txt" + 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"}}) - fileEventPart3 := make(map[string]interface{}) - fileEventPart3["cwd"] = "/dir" + if !isEventReady(fileEvent) { + t.Errorf("Event ready expected") + } +} - // Testing procmon.go#93 [Process Start] - fileEventPart4 := make(map[string]interface{}) - fileEventPart4["argc"] = "2" - fileEventPart4["a0"] = "argv1" - fileEventPart4["a1"] = "argv2" +type auditRecord struct { + msgType auparse.AuditMessageType + text string +} - // 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 ===== +// recordingDispatch collects the events an auditStream dispatches. +type recordingDispatch struct { + mu sync.Mutex + events []Event +} - processMonitor.PrepareEvent(2, fileEventPart1) - processMonitor.PrepareEvent(2, fileEventPart2) - processMonitor.PrepareEvent(2, fileEventPart3) +func (r *recordingDispatch) dispatch(event *Event) { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, *event) +} - processMonitor.PrepareEvent(2, fileEventPart4) - processMonitor.PrepareEvent(2, fileEventPart5) +func (r *recordingDispatch) dispatched() []Event { + r.mu.Lock() + defer r.mu.Unlock() + return append([]Event(nil), r.events...) +} - isReady = isEventReady(processMonitor.Events[2]) +// countingStream wraps an auditStream and counts every delivered event. +type countingStream struct { + *auditStream + delivered atomic.Int64 +} - if !isReady { - t.Errorf("Event ready expected") - } +func (c *countingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + c.delivered.Add(1) + c.auditStream.ReassemblyComplete(msgs) } -func parseAuditRecord(t *testing.T, msgType auparse.AuditMessageType, text string) *auparse.AuditMessage { +func newTestReassembler(t *testing.T, maxInFlight int, timeout time.Duration) (*libaudit.Reassembler, *countingStream, *recordingDispatch, *ProcessMonitor) { t.Helper() - message, err := auparse.Parse(msgType, text) + 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("parse %v: %v", msgType, err) + t.Fatalf("new reassembler: %v", err) } - return message + return r, stream, rec, p } -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) +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) } } - 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") - } +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)} } -type auditRecord struct { - msgType auparse.AuditMessageType - text string +func proctitleRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_PROCTITLE, fmt.Sprintf(`audit(1700000000.000:%d): proctitle=7465737400`, seq)} } -// 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 eoeRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_EOE, fmt.Sprintf(`audit(1700000000.000:%d): `, seq)} } -func TestHandleRecordDispatchesDetectionsOnce(t *testing.T) { +func TestReassemblyDispatchesDetectionsOnce(t *testing.T) { tests := []struct { name string records []auditRecord - check func(*Event) bool + 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{ - {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): `}, + 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), }, - check: func(e *Event) bool { - return e.EventType == processMonitorTag && len(e.ProcessArguments) == 2 && e.Path == "/home/runner/work" + 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{ - {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): `}, + 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), }, - check: func(e *Event) bool { + want: func(e Event) bool { return e.EventType == fileMonitorTag && e.FileName == "out.txt" && e.Path == "/home/runner/work/repo" }, }, { - name: "sendto with an address", + name: "connect over IPv6", 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): `}, + 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), }, - check: func(e *Event) bool { - return e.EventType == netMonitorTag && e.IPAddress == "192.0.2.53" && e.Port == "53" + 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" }, }, } 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)) + r, _, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, tt.records) + + // 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.check(dispatched[0]) { - t.Fatalf("unexpected dispatched event %+v", dispatched[0]) + if !tt.want(got[0]) { + t.Fatalf("unexpected dispatched event %+v", got[0]) } - if len(p.Events) != 0 { - t.Fatalf("expected no pending events after EOE, got %d", len(p.Events)) + r.Close() + if again := rec.dispatched(); len(again) != 1 { + t.Fatalf("the final flush must not dispatch again, got %d", len(again)) } }) } } -func TestPrepareEventDropsIdleEvents(t *testing.T) { - p := &ProcessMonitor{Events: make(map[int]*Event)} - part := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} +func TestReassemblyDropsSendWithoutAddress(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) - last := 5 * pendingEventWindow - for sequence := 1; sequence <= last; sequence++ { - p.PrepareEvent(sequence, part) - } + // 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 len(p.Events) > 2*pendingEventWindow+1 { - t.Fatalf("pending events not bounded: %d", len(p.Events)) + if stream.delivered.Load() != 1 { + t.Fatalf("expected the event to be delivered once, got %d", stream.delivered.Load()) } - if p.droppedEvents == 0 { - t.Fatalf("expected dropped events to be counted") + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a send without an address must not be dispatched, got %+v", got) } - if _, ok := p.Events[last]; !ok { - t.Fatalf("the newest event must be kept") +} + +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")}) } - if _, ok := p.Events[1]; ok { - t.Fatalf("the oldest event should have been dropped") + + // 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 TestPrepareEventKeepsActiveEventsDuringSweep(t *testing.T) { - p := &ProcessMonitor{Events: make(map[int]*Event)} - p.PrepareEvent(1, map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) +func TestReassemblyGivesUpOnEventsAfterTimeout(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, 50*time.Millisecond) - 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{}{}) - } + // 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") } - if p.droppedEvents == 0 { - t.Fatalf("expected the sweep to run") + time.Sleep(100 * time.Millisecond) + if err := r.Maintain(); err != nil { + t.Fatalf("maintain: %v", err) } - event, ok := p.Events[1] - if !ok || event.EventType != netMonitorTag || event.Syscall != "connect" { - t.Fatalf("an active event lost its data: %+v", event) + + 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 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) +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) } - // 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") + stream.EventsLost(3) + if got := p.sequenceGaps.Load(); got != 3 { + t.Fatalf("expected a gap of 3, got %d", got) } - 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 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 TestPrepareEventLosesEventCompletedAfterEviction(t *testing.T) { - p := &ProcessMonitor{Events: make(map[int]*Event)} - p.PrepareEvent(1, map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) +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) + } + }) + } +} - idle := map[string]interface{}{"tags": []string{"netmon"}, "syscall": "sendto"} - for sequence := 2; sequence <= 2+3*pendingEventWindow; sequence++ { - p.PrepareEvent(sequence, idle) +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()) + } } - if _, ok := p.Events[1]; ok { - t.Fatalf("an event idle for longer than the window should be dropped") + want := []int64{1, 2, 4, 8, 16} + if fmt.Sprint(logged) != fmt.Sprint(want) { + t.Fatalf("logged at %v, want %v", logged, want) } - // 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) + // 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 TestIsReceiveOverflow(t *testing.T) { +func TestNoteReceiveError(t *testing.T) { cases := []struct { - err error - want bool + err error + wantOverflow int64 + wantOther int64 }{ - {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}, + {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 { - if got := isReceiveOverflow(c.err); got != c.want { - t.Errorf("isReceiveOverflow(%v) = %v, want %v", c.err, got, c.want) + 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) } } }