Skip to content

fix(procmon): group audit records with go-libaudit's reassembler #503

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
venkata-stepsecurity merged 3 commits into from
Oct 28, 2026
Merged

fix(procmon): group audit records with go-libaudit's reassembler #503

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
25692b0
Revert "Merge pull request #501 from venkata-stepsecurity/venkat/fix-…
venkata-stepsecurity Oct 28, 2026
018a9e6
fix(procmon): group audit records with go-libaudit's reassembler
venkata-stepsecurity Oct 28, 2026
3ac31cc
build: add go-libaudit v2.6.2 tool dependency to go.mod and go.sum
venkata-stepsecurity Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
13 changes: 13 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand All @@ -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=
Expand All @@ -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=
Expand All @@ -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=
Expand Down
171 changes: 76 additions & 95 deletions procmon.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@ package main
import (
"errors"
"fmt"
"math/bits"
"strconv"
"sync"
"sync/atomic"
"syscall"
"time"

Expand All @@ -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
Expand All @@ -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 {
Expand All @@ -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"]

Expand Down Expand Up @@ -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 {
Expand All @@ -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 != "" {
Expand Down
Loading
Loading