diff --git a/.github/workflows/code-review.yml b/.github/workflows/code-review.yml new file mode 100644 index 0000000..2315798 --- /dev/null +++ b/.github/workflows/code-review.yml @@ -0,0 +1,24 @@ +name: Code Review +on: + pull_request: +permissions: + contents: read +jobs: + code-review: + runs-on: ubuntu-latest + permissions: + contents: read + pull-requests: read + steps: + - name: Harden Runner + uses: step-security/harden-runner@f086349bfa2bd1361f7909c78558e816508cdc10 # v2.8.0 + with: + disable-sudo: true + egress-policy: block + allowed-endpoints: > + api.github.com:443 + int.api.stepsecurity.io:443 + + - name: Code Review + uses: step-security/ai-codewise@int + diff --git a/apiclient.go b/apiclient.go index 465f009..fe64e84 100644 --- a/apiclient.go +++ b/apiclient.go @@ -123,7 +123,7 @@ func (apiclient *ApiClient) getGlobalFeatureFlags() GlobalFeatureFlags { u, err := url.Parse(apiclient.APIURL) if err != nil { - return GlobalFeatureFlags{} + return defaultGlobalFeatureFlags() } u.Path = path.Join(u.Path, "global-feature-flags") @@ -138,27 +138,27 @@ func (apiclient *ApiClient) getGlobalFeatureFlags() GlobalFeatureFlags { if err != nil { fmt.Println("Error creating request:", err) - return GlobalFeatureFlags{} + return defaultGlobalFeatureFlags() } resp, err := apiclient.Client.Do(req) if err != nil { fmt.Println("Error sending request:", err) - return GlobalFeatureFlags{} + return defaultGlobalFeatureFlags() } body, err := io.ReadAll(resp.Body) if err != nil { fmt.Println("Error reading response body:", err) - return GlobalFeatureFlags{} + return defaultGlobalFeatureFlags() } var globalFeatureFlags GlobalFeatureFlags err = json.Unmarshal(body, &globalFeatureFlags) if err != nil { fmt.Println("Error unmarshalling response body:", err) - return GlobalFeatureFlags{} + return defaultGlobalFeatureFlags() } return globalFeatureFlags diff --git a/apiclient_test.go b/apiclient_test.go index acd43a5..159b9bd 100644 --- a/apiclient_test.go +++ b/apiclient_test.go @@ -100,3 +100,47 @@ func Test_getGithubMetaDomains_ErrorAfterRetries(t *testing.T) { t.Fatalf("server called %d times, want 3", got) } } + +func Test_getGlobalFeatureFlags(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/global-feature-flags" { + fmt.Fprint(w, `not json`) + return + } + fmt.Fprint(w, `{"agent_type":"agent-oss","revert_github_dns_experiment":false}`) + })) + defer server.Close() + + closedServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + closedServer.Close() + + tests := []struct { + name string + apiURL string + want GlobalFeatureFlags + }{ + {name: "fetched flags", apiURL: server.URL + "/v1", want: GlobalFeatureFlags{AgentType: AgentTypeOSS}}, + {name: "invalid url", apiURL: "%zz", want: defaultGlobalFeatureFlags()}, + {name: "request error", apiURL: closedServer.URL, want: defaultGlobalFeatureFlags()}, + {name: "invalid response", apiURL: server.URL + "/other", want: defaultGlobalFeatureFlags()}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + apiclient := &ApiClient{Client: newTestHTTPClient(), APIURL: tt.apiURL} + + if got := apiclient.getGlobalFeatureFlags(); got != tt.want { + t.Fatalf("getGlobalFeatureFlags() = %+v, want %+v", got, tt.want) + } + }) + } +} + +func Test_GetGlobalFeatureFlags_DefaultsWhenUninitialized(t *testing.T) { + savedManager := globalManager + globalManager = nil + defer func() { globalManager = savedManager }() + + if got := GetGlobalFeatureFlags(); !got.RevertGithubDNSExperiment { + t.Fatalf("GetGlobalFeatureFlags() = %+v, want RevertGithubDNSExperiment true", got) + } +} diff --git a/global_feature_flags.go b/global_feature_flags.go index eca7ce5..b63fc44 100644 --- a/global_feature_flags.go +++ b/global_feature_flags.go @@ -36,7 +36,7 @@ func InitGlobalFeatureFlags(apiURL string, apiClient *ApiClient) { globalOnce.Do(func() { globalManager = &GlobalFeatureFlagManager{ apiURL: apiURL, - flags: GlobalFeatureFlags{}, // Default values + flags: defaultGlobalFeatureFlags(), apiClient: apiClient, } // Fetch initial flags and handle errors gracefully @@ -68,10 +68,10 @@ func RefreshGlobalFeatureFlags() error { } // GetGlobalFeatureFlags returns the current global feature flags. -// This method never returns nil. It returns a default-initialized struct if manager is not initialized. +// This method never returns nil. It returns default flags if manager is not initialized. func GetGlobalFeatureFlags() GlobalFeatureFlags { if globalManager == nil { - return GlobalFeatureFlags{} // Return default-initialized struct + return defaultGlobalFeatureFlags() } globalManager.mutex.RLock() defer globalManager.mutex.RUnlock() @@ -87,3 +87,10 @@ func IsCustomDetectionRulesEnabled() bool { flags := GetGlobalFeatureFlags() return flags.EnableCustomDetectionRules } + +// defaultGlobalFeatureFlags returns the flags used when they could not be fetched. +func defaultGlobalFeatureFlags() GlobalFeatureFlags { + return GlobalFeatureFlags{ + RevertGithubDNSExperiment: true, + } +} 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 e84c5a2..696ac44 100644 --- a/procmon.go +++ b/procmon.go @@ -1,10 +1,17 @@ package main import ( + "errors" "fmt" + "math/bits" "strconv" "sync" + "sync/atomic" + "syscall" "time" + + "github.com/elastic/go-libaudit/v2" + "github.com/elastic/go-libaudit/v2/auparse" ) const ( @@ -13,6 +20,15 @@ const ( processMonitorTag = "procmon" ) +// Audit records are grouped into events by go-libaudit's reassembler, with the +// same limits Auditbeat uses by default: at most 50 events in flight, and an +// event without its last record is given up after 2 seconds. +const ( + reassemblerMaxInFlight = 50 + reassemblerTimeout = 2 * time.Second + reassemblerMaintain = 500 * time.Millisecond +) + type ProcessMonitor struct { CorrelationId string Repo string @@ -20,8 +36,15 @@ type ProcessMonitor struct { DNSProxy *DNSProxy WorkingDirectory string DisableFileMonitoring bool - Events map[int]*Event - mutex sync.RWMutex + + // dispatchEvent replaces the event handler in tests + dispatchEvent func(*Event) + + sequenceGaps atomic.Int64 + receiveOverflow atomic.Int64 + receiveErrors atomic.Int64 + invalidRecords atomic.Int64 + recordErrors atomic.Int64 } type Process struct { @@ -36,61 +59,54 @@ type Process struct { } type Event struct { - FileName string - Path string - Syscall string - Exe string - IPAddress string - Port string - Pid string - ProcessArguments []string - PPid string - Euid string - Timestamp time.Time - EventType string - Status string - SentForProcessing bool + FileName string + Path string + Syscall string + Exe string + IPAddress string + Port string + Pid string + ProcessArguments []string + PPid string + Euid string + Timestamp time.Time + EventType string + Status string } -func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interface{}) { - p.mutex.Lock() - _, found := p.Events[sequence] - if !found { - p.Events[sequence] = &Event{} - } - +func (event *Event) addRecord(eventMap map[string]interface{}) { // Tags value, found := eventMap["tags"] if found { tags := value.([]string) for _, tag := range tags { - p.Events[sequence].EventType = tag - p.Events[sequence].Syscall = getValue("syscall", eventMap) - p.Events[sequence].Exe = getValue("exe", eventMap) - p.Events[sequence].Pid = getValue("pid", eventMap) - p.Events[sequence].PPid = getValue("ppid", eventMap) - p.Events[sequence].Euid = getValue("euid", eventMap) + event.EventType = tag + event.Syscall = getValue("syscall", eventMap) + event.Exe = getValue("exe", eventMap) + event.Pid = getValue("pid", eventMap) + event.PPid = getValue("ppid", eventMap) + event.Euid = getValue("euid", eventMap) timestamp, err := time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", getValue("@timestamp", eventMap)) if err != nil { timestamp = time.Now().UTC() } - p.Events[sequence].Timestamp = timestamp - p.Events[sequence].Status = getValue("result", eventMap) + event.Timestamp = timestamp + event.Status = getValue("result", eventMap) } } // Current working directory _, found = eventMap["cwd"] if found { - p.Events[sequence].Path = getValue("cwd", eventMap) + event.Path = getValue("cwd", eventMap) } // Connect _, found = eventMap["addr"] if found { - p.Events[sequence].IPAddress = getValue("addr", eventMap) - p.Events[sequence].Port = getValue("port", eventMap) + event.IPAddress = getValue("addr", eventMap) + event.Port = getValue("port", eventMap) } // Process start @@ -103,7 +119,7 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac WriteLog(fmt.Sprintf("could not parse argc:%v", argc)) } for i := 0; i < argCount; i++ { - p.Events[sequence].ProcessArguments = append(p.Events[sequence].ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)])) + event.ProcessArguments = append(event.ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)])) } } @@ -112,12 +128,11 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac if found { nameTypeStr := fmt.Sprintf("%v", nameType) - if nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL" { - p.Events[sequence].FileName = getValue("name", eventMap) + // the first qualifying PATH wins, as when events were dispatched on it + if (nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL") && event.FileName == "" { + event.FileName = getValue("name", eventMap) } } - - p.mutex.Unlock() } func getValue(key string, eventMap map[string]interface{}) string { @@ -129,17 +144,120 @@ func getValue(key string, eventMap map[string]interface{}) string { return "" } -func (p *ProcessMonitor) markEventSent(event *Event) { - p.mutex.Lock() - event.SentForProcessing = true - p.mutex.Unlock() +// auditStream receives complete events from the reassembler and dispatches +// the ones the event handler uses. Every event is delivered once, including +// incomplete ones that are evicted, so nothing is kept after delivery. +type auditStream struct { + monitor *ProcessMonitor + dispatch func(*Event) + + // ops serializes push with maintain and flush, so an event cannot be + // given up between a record being admitted and being pushed + ops sync.Mutex + + // syscallEvents holds the sequence numbers of the events in the + // reassembler, which all start with a SYSCALL record (see push) + mu sync.Mutex + syscallEvents map[uint32]struct{} } -func isEventReady(event *Event) bool { - if event.SentForProcessing { +// push passes a record to the reassembler if it belongs to an audited +// syscall. The records the agent uses (EXECVE, SOCKADDR, CWD, PATH, +// PROCTITLE) follow the event's SYSCALL record. Other records, such as BPF or +// ANOM_PROMISCUOUS, can arrive alone without the records that complete an +// event, and since the reassembler delivers events in sequence order, each +// one would hold back every later event until it times out. +func (s *auditStream) push(r *libaudit.Reassembler, msg *auparse.AuditMessage) { + s.ops.Lock() + defer s.ops.Unlock() + + s.mu.Lock() + if s.syscallEvents == nil { + s.syscallEvents = make(map[uint32]struct{}) + } + _, found := s.syscallEvents[msg.Sequence] + if !found && msg.RecordType == auparse.AUDIT_SYSCALL { + s.syscallEvents[msg.Sequence] = struct{}{} + found = true + } + s.mu.Unlock() + + if found { + r.PushMessage(msg) + } +} + +// maintain gives up on events whose last record did not arrive in time. +func (s *auditStream) maintain(r *libaudit.Reassembler) error { + s.ops.Lock() + defer s.ops.Unlock() + return r.Maintain() +} + +// flush delivers the events still in the reassembler and closes it. +func (s *auditStream) flush(r *libaudit.Reassembler) { + s.ops.Lock() + defer s.ops.Unlock() + r.Close() +} + +func (s *auditStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + if len(msgs) == 0 { + return + } + // records that arrive after this, such as the EOE after a PROCTITLE, + // are not passed on + s.mu.Lock() + delete(s.syscallEvents, msgs[0].Sequence) + s.mu.Unlock() + + event := &Event{} + for _, msg := range msgs { + eventMap := msg.ToMapStr() + if parseErr, found := eventMap["error"]; found { + noteLoss(&s.monitor.recordErrors, 1, fmt.Sprintf("audit record fields not parsed: %v", parseErr)) + } + event.addRecord(eventMap) + } + if isEventReady(event) { + s.dispatch(event) + } +} + +// EventsLost is called for gaps in the audit sequence numbers. A gap is not +// always a loss, since events this agent skips use sequence numbers too. +func (s *auditStream) EventsLost(count int) { + // go-libaudit v2.3.2 underflows when a record arrives after its event was + // given up, which shows up as a huge count + if count <= 0 || count > 1<<24 { + return + } + noteLoss(&s.monitor.sequenceGaps, int64(count), "gaps in audit sequence numbers") +} + +// noteReceiveError counts a failed receive. A full netlink socket (ENOBUFS) +// means the kernel dropped messages; the socket stays usable either way. +func (p *ProcessMonitor) noteReceiveError(err error) { + if errors.Is(err, syscall.ENOBUFS) { + noteLoss(&p.receiveOverflow, 1, "audit netlink buffer overflowed, kernel dropped messages") + return + } + noteLoss(&p.receiveErrors, 1, fmt.Sprintf("audit receive error: %v", err)) +} + +// noteLoss adds n to the counter and logs when the total reaches the next +// power of two, so a sustained loss does not flood the log. It reports whether +// it logged. +func noteLoss(counter *atomic.Int64, n int64, what string) bool { + total := counter.Add(n) + if bits.Len64(uint64(total)) == bits.Len64(uint64(total-n)) { return false } + WriteLog(fmt.Sprintf("[procmon] %s, total=%d", what, total)) + return true +} +func isEventReady(event *Event) bool { switch event.EventType { case netMonitorTag: if event.IPAddress != "" && event.Port != "" { diff --git a/procmon_linux.go b/procmon_linux.go index 4730ce1..d0ec761 100644 --- a/procmon_linux.go +++ b/procmon_linux.go @@ -10,6 +10,9 @@ import ( "io/ioutil" "os" "strings" + "sync" + "syscall" + "time" "github.com/elastic/go-libaudit/v2" "github.com/elastic/go-libaudit/v2/auparse" @@ -151,7 +154,9 @@ func (p *ProcessMonitor) MonitorProcesses(errc chan error) { WriteLog("receive called") WriteLog("\n") - p.receive(client) + if err := p.receive(client); err != nil { + WriteLog(fmt.Sprintf("[procmon] receive stopped: %v", err)) + } } func fileMonitorAuditRule(workingDirectory, goarch string) string { @@ -164,7 +169,6 @@ func fileMonitorAuditRule(workingDirectory, goarch string) string { func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error { - p.Events = make(map[int]*Event) eventHandler := EventHandler{CorrelationId: p.CorrelationId, Repo: p.Repo, ApiClient: p.ApiClient, DNSProxy: p.DNSProxy} eventHandler.ProcessConnectionMap = make(map[string]bool) eventHandler.ProcessFileMap = make(map[string]bool) @@ -172,10 +176,42 @@ 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) } + } + + stream := &auditStream{monitor: p, dispatch: dispatch} + reassembler, err := libaudit.NewReassembler(reassemblerMaxInFlight, reassemblerTimeout, stream) + 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(stream, reassembler, done) + }() + // stop maintenance before the final flush, so no callback runs after return + defer func() { + close(done) + maintaining.Wait() + stream.flush(reassembler) + }() + for { rawEvent, err := r.Receive(false) if err != nil { - return errors.Wrap(err, "receive failed") + // EBADF means the client was closed + if errors.Is(err, syscall.EBADF) { + return errors.Wrap(err, "receive failed") + } + p.noteReceiveError(err) + if !errors.Is(err, syscall.ENOBUFS) { + time.Sleep(10 * time.Millisecond) // do not spin on a persistent error + } + continue } // Messages from 1300-2999 are valid audit messages. @@ -186,16 +222,27 @@ func (p *ProcessMonitor) receive(r *libaudit.AuditClient) error { message, err := auparse.Parse(rawEvent.Type, string(rawEvent.Data)) if err != nil { - return errors.Wrap(err, "parse failed.") + noteLoss(&p.invalidRecords, 1, fmt.Sprintf("dropped audit record with an invalid header: %v", err)) + continue } - eventMap := message.ToMapStr() + stream.push(reassembler, message) + } +} - p.PrepareEvent(int(message.Sequence), eventMap) - if isEventReady(p.Events[int(message.Sequence)]) { - p.markEventSent(p.Events[int(message.Sequence)]) - go eventHandler.HandleEvent(p.Events[int(message.Sequence)]) +// maintainReassembler gives up on events whose last record did not arrive in +// time, as Auditbeat does, until done is closed. +func maintainReassembler(s *auditStream, r *libaudit.Reassembler, done <-chan struct{}) { + ticker := time.NewTicker(reassemblerMaintain) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + if err := s.maintain(r); err != nil { + return + } } - } } diff --git a/procmon_linux_test.go b/procmon_linux_test.go index a6189fe..3e18aca 100644 --- a/procmon_linux_test.go +++ b/procmon_linux_test.go @@ -4,8 +4,16 @@ package main import ( + "errors" + "fmt" "os" + "sync" + "syscall" "testing" + "time" + + "github.com/elastic/go-libaudit/v2" + "github.com/elastic/go-libaudit/v2/auparse" ) func Test_fileMonitorAuditRule(t *testing.T) { @@ -65,3 +73,212 @@ func Test_getProcessExe(t *testing.T) { } } + +// fakeNetlink replays audit records and errors to AuditClient.Receive. +type fakeNetlink struct { + steps []fakeNetlinkStep +} + +type fakeNetlinkStep struct { + msgType auparse.AuditMessageType + text string + err error + block chan struct{} // Receive waits on it, then reports the client closed +} + +func (f *fakeNetlink) Receive(bool, libaudit.NetlinkParser) ([]syscall.NetlinkMessage, error) { + if len(f.steps) == 0 { + return nil, syscall.EBADF + } + step := f.steps[0] + f.steps = f.steps[1:] + if step.block != nil { + <-step.block + return nil, syscall.EBADF + } + if step.err != nil { + return nil, step.err + } + return []syscall.NetlinkMessage{{ + Header: syscall.NlMsghdr{Type: uint16(step.msgType)}, + Data: []byte(step.text), + }}, nil +} + +func (f *fakeNetlink) Send(syscall.NetlinkMessage) (uint32, error) { return 0, nil } +func (f *fakeNetlink) SendNoWait(syscall.NetlinkMessage) (uint32, error) { return 0, nil } +func (f *fakeNetlink) Close() error { return nil } + +func TestReceiveKeepsGoingUntilTheClientIsClosed(t *testing.T) { + netlink := &fakeNetlink{steps: []fakeNetlinkStep{ + {err: syscall.ENOBUFS}, + {err: syscall.EINVAL}, + {msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:43): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=101 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="curl" exe="/usr/bin/curl" key="netmon"`}, + {msgType: auparse.AUDIT_SOCKADDR, text: `audit(1700000000.123:43): saddr=020001BBC000020A0000000000000000`}, + {msgType: auparse.AUDIT_CWD, text: `not an audit record`}, + {err: syscall.ENOBUFS}, + {msgType: auparse.AUDIT_PROCTITLE, text: `audit(1700000000.123:43): proctitle=6375726C`}, + {msgType: auparse.AUDIT_EOE, text: `audit(1700000000.123:43): `}, + }} + var mu sync.Mutex + var dispatched []Event + p := &ProcessMonitor{dispatchEvent: func(e *Event) { + mu.Lock() + dispatched = append(dispatched, *e) + mu.Unlock() + }} + + err := p.receive(&libaudit.AuditClient{Netlink: netlink}) + + if !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop only when the client is closed, got %v", err) + } + if len(netlink.steps) != 0 { + t.Fatalf("not every message was read: %d left", len(netlink.steps)) + } + if got := p.receiveOverflow.Load(); got != 2 { + t.Fatalf("expected 2 overflows counted, got %d", got) + } + if got := p.receiveErrors.Load(); got != 1 { + t.Fatalf("expected 1 other receive error counted, got %d", got) + } + if got := p.invalidRecords.Load(); got != 1 { + t.Fatalf("expected the unparsable record to be dropped, got %d", got) + } + mu.Lock() + defer mu.Unlock() + if len(dispatched) != 1 || dispatched[0].IPAddress != "192.0.2.10" || dispatched[0].Port != "443" { + t.Fatalf("expected the connect to be dispatched once, got %+v", dispatched) + } +} + +func TestReceiveDeliversTimedOutEventsWhileIdle(t *testing.T) { + connect := func(seq int, last bool) []fakeNetlinkStep { + steps := []fakeNetlinkStep{ + {msgType: auparse.AUDIT_SYSCALL, text: fmt.Sprintf(`audit(1700000000.123:%d): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=%d auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="curl" exe="/usr/bin/curl" key="netmon"`, seq, seq)}, + {msgType: auparse.AUDIT_SOCKADDR, text: fmt.Sprintf(`audit(1700000000.123:%d): saddr=020001BBC000020A0000000000000000`, seq)}, + } + if last { + steps = append(steps, fakeNetlinkStep{msgType: auparse.AUDIT_PROCTITLE, text: fmt.Sprintf(`audit(1700000000.123:%d): proctitle=6375726C`, seq)}) + } + return steps + } + release := make(chan struct{}) + var steps []fakeNetlinkStep + steps = append(steps, connect(79, true)...) // complete, delivered right away + steps = append(steps, connect(80, false)...) // never completes + steps = append(steps, connect(81, true)...) // complete, but behind 80 + steps = append(steps, fakeNetlinkStep{block: release}) + netlink := &fakeNetlink{steps: steps} + + var mu sync.Mutex + var pids []string + p := &ProcessMonitor{dispatchEvent: func(e *Event) { + mu.Lock() + pids = append(pids, e.Pid) + mu.Unlock() + }} + delivered := func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), pids...) + } + + result := make(chan error, 1) + go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }() + + // 79 completes on its PROCTITLE, well before the timeout + completeBy := time.Now().Add(reassemblerTimeout / 2) + for len(delivered()) < 1 && time.Now().Before(completeBy) { + time.Sleep(10 * time.Millisecond) + } + if got := delivered(); fmt.Sprint(got) != "[79]" { + t.Fatalf("expected 79 to be delivered when it completed, got %v", got) + } + + // while receive is blocked, only the maintenance goroutine can deliver 80 and 81 + deadline := time.Now().Add(reassemblerTimeout + 2*time.Second) + for len(delivered()) < 3 && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + if got := delivered(); fmt.Sprint(got) != "[79 80 81]" { + t.Fatalf("expected 79, then 80 and 81 after the timeout, got %v", got) + } + + close(release) + if err := <-result; !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop when the client is closed, got %v", err) + } + if got := delivered(); len(got) != 3 { + t.Fatalf("no event should be delivered twice, got %v", got) + } +} + +func TestReceiveWaitsForMaintenanceBeforeReturning(t *testing.T) { + release := make(chan struct{}) + netlink := &fakeNetlink{steps: []fakeNetlinkStep{ + // an incomplete connect that only the maintenance goroutine can deliver + {msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:95): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=95 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="curl" exe="/usr/bin/curl" key="netmon"`}, + {msgType: auparse.AUDIT_SOCKADDR, text: `audit(1700000000.123:95): saddr=020001BBC000020A0000000000000000`}, + {block: release}, + }} + inCallback := make(chan struct{}) + finishCallback := make(chan struct{}) + p := &ProcessMonitor{dispatchEvent: func(*Event) { + close(inCallback) + <-finishCallback + }} + + result := make(chan error, 1) + go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }() + + select { + case <-inCallback: + case <-time.After(reassemblerTimeout + 2*time.Second): + t.Fatalf("maintenance did not deliver the timed out event") + } + + // the client closes while the maintenance callback is still running + close(release) + select { + case err := <-result: + t.Fatalf("receive returned (%v) while a maintenance callback was running", err) + case <-time.After(200 * time.Millisecond): + } + + close(finishCallback) + if err := <-result; !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop when the client is closed, got %v", err) + } +} + +func TestReceiveDoesNotHoldEventsBehindRecordsOutsideSyscalls(t *testing.T) { + release := make(chan struct{}) + netlink := &fakeNetlink{steps: []fakeNetlinkStep{ + // logged on its own, with no record that completes it + {msgType: auparse.AUDIT_BPF, text: `audit(1700000000.123:96): prog-id=12 op=UNLOAD`}, + {msgType: auparse.AUDIT_SYSCALL, text: `audit(1700000000.123:97): arch=c000003e syscall=42 success=yes exit=0 a0=3 a1=7ffd a2=10 a3=0 items=0 ppid=1 pid=97 auid=4294967295 uid=1001 gid=1001 euid=1001 suid=1001 fsuid=1001 egid=1001 sgid=1001 fsgid=1001 tty=(none) ses=4294967295 comm="curl" exe="/usr/bin/curl" key="netmon"`}, + {msgType: auparse.AUDIT_SOCKADDR, text: `audit(1700000000.123:97): saddr=020001BBC000020A0000000000000000`}, + {msgType: auparse.AUDIT_PROCTITLE, text: `audit(1700000000.123:97): proctitle=6375726C`}, + {block: release}, + }} + dispatched := make(chan *Event, 1) + p := &ProcessMonitor{dispatchEvent: func(e *Event) { dispatched <- e }} + + result := make(chan error, 1) + go func() { result <- p.receive(&libaudit.AuditClient{Netlink: netlink}) }() + + select { + case e := <-dispatched: + if e.IPAddress != "192.0.2.10" { + t.Fatalf("unexpected event %+v", e) + } + case <-time.After(reassemblerTimeout / 2): + t.Fatalf("the connect was held behind the BPF record") + } + + close(release) + if err := <-result; !errors.Is(err, syscall.EBADF) { + t.Fatalf("receive should stop when the client is closed, got %v", err) + } +} diff --git a/procmon_test.go b/procmon_test.go index c110e86..5e8ccf0 100644 --- a/procmon_test.go +++ b/procmon_test.go @@ -1,66 +1,560 @@ package main import ( + "errors" + "fmt" + "os" + "reflect" + "sync" + "sync/atomic" + "syscall" "testing" + "time" + + "github.com/elastic/go-libaudit/v2" + "github.com/elastic/go-libaudit/v2/auparse" ) -func TestProcessMonitor_PrepareEvent(t *testing.T) { - processMonitor := &ProcessMonitor{} - processMonitor.Events = make(map[int]*Event) +func TestEvent_addRecord(t *testing.T) { + netEvent := &Event{} + netEvent.addRecord(map[string]interface{}{"tags": []string{"netmon"}, "syscall": "connect"}) + netEvent.addRecord(map[string]interface{}{"addr": "2.2.2.2", "port": "443"}) - netEventPart1 := make(map[string]interface{}) - netEventPart1["tags"] = []string{"netmon"} - netEventPart1["syscall"] = "connect" + if !isEventReady(netEvent) { + t.Errorf("Event ready expected") + } - netEventPart2 := make(map[string]interface{}) - netEventPart2["addr"] = "2.2.2.2" - netEventPart2["port"] = "443" + fileEvent := &Event{} + fileEvent.addRecord(map[string]interface{}{"tags": []string{"filemon"}, "syscall": "open"}) + fileEvent.addRecord(map[string]interface{}{"nametype": "CREATE", "name": "file1.txt"}) + fileEvent.addRecord(map[string]interface{}{"cwd": "/dir"}) + fileEvent.addRecord(map[string]interface{}{"argc": "2", "a0": "argv1", "a1": "argv2"}) + // argc that cannot be converted to an int is logged and ignored + fileEvent.addRecord(map[string]interface{}{"argc": []string{"failed"}}) - processMonitor.PrepareEvent(1, netEventPart1) - processMonitor.PrepareEvent(1, netEventPart2) + if !isEventReady(fileEvent) { + t.Errorf("Event ready expected") + } +} - isReady := isEventReady(processMonitor.Events[1]) +type auditRecord struct { + msgType auparse.AuditMessageType + text string +} - if !isReady { - t.Errorf("Event ready expected") +// recordingDispatch collects the events an auditStream dispatches. +type recordingDispatch struct { + mu sync.Mutex + events []Event +} + +func (r *recordingDispatch) dispatch(event *Event) { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, *event) +} + +func (r *recordingDispatch) dispatched() []Event { + r.mu.Lock() + defer r.mu.Unlock() + return append([]Event(nil), r.events...) +} + +// countingStream wraps an auditStream and counts every delivered event. +type countingStream struct { + *auditStream + delivered atomic.Int64 +} + +func (c *countingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + c.delivered.Add(1) + c.auditStream.ReassemblyComplete(msgs) +} + +func newTestReassembler(t *testing.T, maxInFlight int, timeout time.Duration) (*libaudit.Reassembler, *countingStream, *recordingDispatch, *ProcessMonitor) { + t.Helper() + p := &ProcessMonitor{} + rec := &recordingDispatch{} + stream := &countingStream{auditStream: &auditStream{monitor: p, dispatch: rec.dispatch}} + r, err := libaudit.NewReassembler(maxInFlight, timeout, stream) + if err != nil { + t.Fatalf("new reassembler: %v", err) } + return r, stream, rec, p +} - fileEventPart1 := make(map[string]interface{}) - fileEventPart1["tags"] = []string{"filemon"} - fileEventPart1["syscall"] = "open" +// pushRecords passes records to the reassembler the way receive does. +func pushRecords(t *testing.T, r *libaudit.Reassembler, stream *countingStream, records []auditRecord) { + t.Helper() + for _, rec := range records { + msg, err := auparse.Parse(rec.msgType, rec.text) + if err != nil { + t.Fatalf("parse %v: %v", rec.msgType, err) + } + stream.push(r, msg) + } +} - fileEventPart2 := make(map[string]interface{}) - fileEventPart2["nametype"] = "CREATE" - fileEventPart2["name"] = "file1.txt" +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)} +} - fileEventPart3 := make(map[string]interface{}) - fileEventPart3["cwd"] = "/dir" +func proctitleRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_PROCTITLE, fmt.Sprintf(`audit(1700000000.000:%d): proctitle=7465737400`, seq)} +} - // Testing procmon.go#93 [Process Start] - fileEventPart4 := make(map[string]interface{}) - fileEventPart4["argc"] = "2" - fileEventPart4["a0"] = "argv1" - fileEventPart4["a1"] = "argv2" +func eoeRecord(seq int) auditRecord { + return auditRecord{auparse.AUDIT_EOE, fmt.Sprintf(`audit(1700000000.000:%d): `, seq)} +} - // 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 ===== +func TestReassemblyDispatchesDetectionsOnce(t *testing.T) { + tests := []struct { + name string + records []auditRecord + want func(Event) bool + }{ + { + name: "connect", + records: []auditRecord{ + syscallRecord(43, 42, "curl", "/usr/bin/curl", "netmon"), + // 192.0.2.10:443 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:43): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(43), + eoeRecord(43), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.Syscall == "connect" && e.IPAddress == "192.0.2.10" && e.Port == "443" && + e.Pid == "143" && e.PPid == "1" && e.Exe == "/usr/bin/curl" && e.Euid == "1001" && e.Status == "success" && + e.Timestamp.Equal(time.Unix(1700000000, 0)) + }, + }, + { + name: "sendto with an address", + records: []auditRecord{ + syscallRecord(52, 44, "dig", "/usr/bin/dig", "netmon"), + // 192.0.2.53:53 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:52): saddr=02000035C0000235000000000000000000`}, + proctitleRecord(52), + eoeRecord(52), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.IPAddress == "192.0.2.53" && e.Port == "53" + }, + }, + { + name: "process start", + records: []auditRecord{ + syscallRecord(50, 59, "ls", "/usr/bin/ls", "procmon"), + {auparse.AUDIT_EXECVE, `audit(1700000000.000:50): argc=2 a0="ls" a1="-l"`}, + {auparse.AUDIT_CWD, `audit(1700000000.000:50): cwd="/home/runner/work"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:50): item=0 name="/usr/bin/ls" inode=10 dev=08:01 mode=0100755 ouid=0 ogid=0 rdev=00:00 nametype=NORMAL cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(50), + eoeRecord(50), + }, + want: func(e Event) bool { + return e.EventType == processMonitorTag && reflect.DeepEqual(e.ProcessArguments, []string{"ls", "-l"}) && e.Path == "/home/runner/work" && + e.Pid == "150" && e.Exe == "/usr/bin/ls" + }, + }, + { + name: "file write", + records: []auditRecord{ + syscallRecord(51, 257, "bash", "/usr/bin/bash", "filemon"), + {auparse.AUDIT_CWD, `audit(1700000000.000:51): cwd="/home/runner/work/repo"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:51): item=1 name="out.txt" inode=11 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=CREATE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(51), + eoeRecord(51), + }, + want: func(e Event) bool { + return e.EventType == fileMonitorTag && e.FileName == "out.txt" && e.Path == "/home/runner/work/repo" + }, + }, + { + name: "connect over IPv6", + records: []auditRecord{ + syscallRecord(54, 42, "curl", "/usr/bin/curl", "netmon"), + // [2001:db8::1]:443 + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:54): saddr=0A0001BB0000000020010DB800000000000000000000000100000000`}, + proctitleRecord(54), + eoeRecord(54), + }, + want: func(e Event) bool { + return e.EventType == netMonitorTag && e.IPAddress == "2001:db8::1" && e.Port == "443" + }, + }, + { + // rename has PARENT, DELETE and CREATE paths; the first qualifying one is kept + name: "rename keeps the source file", + records: []auditRecord{ + syscallRecord(53, 82, "mv", "/usr/bin/mv", "filemon"), + {auparse.AUDIT_CWD, `audit(1700000000.000:53): cwd="/home/runner/work/repo"`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=0 name="/home/runner/work/repo" inode=5 dev=08:01 mode=040755 ouid=1001 ogid=1001 rdev=00:00 nametype=PARENT cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=1 name="/home/runner/work/repo" inode=5 dev=08:01 mode=040755 ouid=1001 ogid=1001 rdev=00:00 nametype=PARENT cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=2 name="Dockerfile" inode=12 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=DELETE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + {auparse.AUDIT_PATH, `audit(1700000000.000:53): item=3 name="tmpfile" inode=12 dev=08:01 mode=0100644 ouid=1001 ogid=1001 rdev=00:00 nametype=CREATE cap_fp=0 cap_fi=0 cap_fe=0 cap_fver=0 cap_frootid=0`}, + proctitleRecord(53), + eoeRecord(53), + }, + want: func(e Event) bool { + return e.EventType == fileMonitorTag && e.FileName == "Dockerfile" && e.Path == "/home/runner/work/repo" + }, + }, + } - processMonitor.PrepareEvent(2, fileEventPart1) - processMonitor.PrepareEvent(2, fileEventPart2) - processMonitor.PrepareEvent(2, fileEventPart3) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, stream, tt.records) - processMonitor.PrepareEvent(2, fileEventPart4) - processMonitor.PrepareEvent(2, fileEventPart5) + // delivered when the event completes, before any timeout or flush + got := rec.dispatched() + if len(got) != 1 { + t.Fatalf("expected one dispatch, got %d", len(got)) + } + if !tt.want(got[0]) { + t.Fatalf("unexpected dispatched event %+v", got[0]) + } + r.Close() + if again := rec.dispatched(); len(again) != 1 { + t.Fatalf("the final flush must not dispatch again, got %d", len(again)) + } + }) + } +} - isReady = isEventReady(processMonitor.Events[2]) +func TestReassemblySkipsRecordsOutsideSyscallEvents(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) - if !isReady { - t.Errorf("Event ready expected") + pushRecords(t, r, stream, []auditRecord{ + // a BPF program unload, logged on its own, with no record that completes it + {auparse.AUDIT_BPF, `audit(1700000000.000:30): prog-id=12 op=UNLOAD`}, + // a BPF record logged during an audited syscall, before its SYSCALL record + {auparse.AUDIT_BPF, `audit(1700000000.000:31): prog-id=13 op=LOAD`}, + syscallRecord(31, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:31): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(31), + eoeRecord(31), + syscallRecord(32, 59, "ls", "/usr/bin/ls", "procmon"), + {auparse.AUDIT_EXECVE, `audit(1700000000.000:32): argc=1 a0="ls"`}, + {auparse.AUDIT_CWD, `audit(1700000000.000:32): cwd="/home/runner/work"`}, + proctitleRecord(32), + eoeRecord(32), + }) + + // both are delivered as they complete, not held behind the BPF record + got := rec.dispatched() + if len(got) != 2 || got[0].IPAddress != "192.0.2.10" || got[1].Exe != "/usr/bin/ls" { + t.Fatalf("expected the connect and the process start right away, got %+v", got) + } + if delivered := stream.delivered.Load(); delivered != 2 { + t.Fatalf("records outside syscall events must not reach the reassembler, %d events delivered", delivered) + } + stream.mu.Lock() + defer stream.mu.Unlock() + if len(stream.syscallEvents) != 0 { + t.Fatalf("delivered events must be forgotten, %d left", len(stream.syscallEvents)) + } +} + +func TestReassemblyCompletesOnEOEWithoutProctitle(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(34, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:34): saddr=020001BBC000020A0000000000000000`}, + eoeRecord(34), + }) + + if got := rec.dispatched(); len(got) != 1 || got[0].IPAddress != "192.0.2.10" { + t.Fatalf("expected the connect to be dispatched on its EOE, got %+v", got) + } +} + +func TestReassemblyDropsRecordsAfterTheEventIsDelivered(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, 50*time.Millisecond) + + // a connect given up before its last records arrive + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(33, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:33): saddr=020001BBC000020A0000000000000000`}, + }) + time.Sleep(100 * time.Millisecond) + if err := r.Maintain(); err != nil { + t.Fatalf("maintain: %v", err) + } + pushRecords(t, r, stream, []auditRecord{proctitleRecord(33), eoeRecord(33)}) + r.Close() + + if delivered := stream.delivered.Load(); delivered != 1 { + t.Fatalf("late records must not start a new event, %d events delivered", delivered) + } + if got := rec.dispatched(); len(got) != 1 { + t.Fatalf("expected the connect to be dispatched once, got %+v", got) + } +} + +// blockingStream holds the delivery of one event until released. +type blockingStream struct { + *auditStream + seq uint32 + entered chan struct{} + release chan struct{} + enterOne sync.Once +} + +func (b *blockingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) { + if msgs[0].Sequence == b.seq { + b.enterOne.Do(func() { close(b.entered) }) + <-b.release + } + b.auditStream.ReassemblyComplete(msgs) +} + +func TestLateRecordWaitsForTheEventToBeGivenUp(t *testing.T) { + rec := &recordingDispatch{} + stream := &auditStream{monitor: &ProcessMonitor{}, dispatch: rec.dispatch} + blocking := &blockingStream{auditStream: stream, seq: 40, entered: make(chan struct{}), release: make(chan struct{})} + r, err := libaudit.NewReassembler(reassemblerMaxInFlight, 50*time.Millisecond, blocking) + if err != nil { + t.Fatalf("new reassembler: %v", err) + } + push := func(records ...auditRecord) { + for _, record := range records { + msg, err := auparse.Parse(record.msgType, record.text) + if err != nil { + t.Errorf("parse %v: %v", record.msgType, err) + return + } + stream.push(r, msg) + } + } + + var releaseOnce sync.Once + releaseCallback := func() { releaseOnce.Do(func() { close(blocking.release) }) } + maintained := make(chan struct{}) + t.Cleanup(func() { + releaseCallback() + <-maintained + }) + + // a connect whose address arrives just as it is given up + push(syscallRecord(40, 42, "curl", "/usr/bin/curl", "netmon")) + time.Sleep(100 * time.Millisecond) + go func() { + defer close(maintained) + stream.maintain(r) + }() + select { + case <-blocking.entered: + case <-time.After(5 * time.Second): + t.Fatalf("maintenance did not give up the event") + } + + pushed := make(chan struct{}) + go func() { + push(auditRecord{auparse.AUDIT_SOCKADDR, `audit(1700000000.000:40): saddr=020001BBC000020A0000000000000000`}) + close(pushed) + }() + select { + case <-pushed: + t.Fatalf("a record was pushed while its event was being given up") + case <-time.After(100 * time.Millisecond): + } + releaseCallback() + select { + case <-pushed: + case <-time.After(5 * time.Second): + t.Fatalf("the late record was not pushed after the event was given up") + } + + // the late record did not start a new event that holds this one back + push( + syscallRecord(41, 42, "curl", "/usr/bin/curl", "netmon"), + auditRecord{auparse.AUDIT_SOCKADDR, `audit(1700000000.000:41): saddr=020001BBC000020A0000000000000000`}, + proctitleRecord(41), + ) + got := rec.dispatched() + if len(got) != 1 || got[0].Pid != "141" { + t.Fatalf("expected only the second connect to be dispatched, right away, got %+v", got) + } +} + +func TestSyscallEventsAreBounded(t *testing.T) { + r, stream, _, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + for seq := 1; seq <= 10000; seq++ { + pushRecords(t, r, stream, []auditRecord{syscallRecord(seq, 44, "python3", "/usr/bin/python3.12", "netmon")}) + } + + stream.mu.Lock() + defer stream.mu.Unlock() + // at most the events the reassembler holds, plus the one being added + if n := len(stream.syscallEvents); n > reassemblerMaxInFlight+1 { + t.Fatalf("expected at most %d tracked events, got %d", reassemblerMaxInFlight+1, n) + } +} + +func TestReassemblyDropsSendWithoutAddress(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // send() on a connected socket: sendto with no SOCKADDR record + pushRecords(t, r, stream, []auditRecord{syscallRecord(42, 44, "python3", "/usr/bin/python3.12", "netmon"), proctitleRecord(42), eoeRecord(42)}) + + if stream.delivered.Load() != 1 { + t.Fatalf("expected the event to be delivered once, got %d", stream.delivered.Load()) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a send without an address must not be dispatched, got %+v", got) + } +} + +func TestReassemblyDoesNotKeepIncompleteEvents(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // records that never complete (no PROCTITLE or EOE) must not pile up + const events = 10000 + for seq := 1; seq <= events; seq++ { + pushRecords(t, r, stream, []auditRecord{syscallRecord(seq, 44, "python3", "/usr/bin/python3.12", "netmon")}) + } + + // 50 is Auditbeat's default; compared as a literal so a change to the limit is caught + if inFlight := events - stream.delivered.Load(); inFlight > 50 { + t.Fatalf("expected at most 50 events in flight, got %d", inFlight) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("incomplete sends must not be dispatched, got %d", len(got)) + } +} + +func TestReassemblyGivesUpOnEventsAfterTimeout(t *testing.T) { + r, stream, rec, _ := newTestReassembler(t, reassemblerMaxInFlight, 50*time.Millisecond) + + // a connect whose last records never arrive + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(60, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:60): saddr=020001BBC000020A0000000000000000`}, + }) + if stream.delivered.Load() != 0 { + t.Fatalf("event delivered before it completed or timed out") + } + + time.Sleep(100 * time.Millisecond) + if err := r.Maintain(); err != nil { + t.Fatalf("maintain: %v", err) + } + + if stream.delivered.Load() != 1 { + t.Fatalf("expected the timed out event to be delivered, got %d", stream.delivered.Load()) + } + if got := rec.dispatched(); len(got) != 1 || got[0].IPAddress != "192.0.2.10" { + t.Fatalf("expected the connect to be dispatched with what arrived, got %+v", got) + } +} + +func TestReassemblyCountsLostEvents(t *testing.T) { + r, stream, _, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // sequences 71 to 74 never arrive + pushRecords(t, r, stream, []auditRecord{syscallRecord(70, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(70)}) + pushRecords(t, r, stream, []auditRecord{syscallRecord(75, 42, "curl", "/usr/bin/curl", "netmon"), proctitleRecord(75)}) + + if got := p.sequenceGaps.Load(); got != 4 { + t.Fatalf("expected a gap of 4, got %d", got) + } +} + +func TestEventsLostIgnoresImpossibleGaps(t *testing.T) { + p := &ProcessMonitor{} + stream := &auditStream{monitor: p, dispatch: func(*Event) {}} + + // what go-libaudit v2.3.2 reports when a record arrives after its event was given up + stream.EventsLost(int(^uint32(0))) + stream.EventsLost(0) + if got := p.sequenceGaps.Load(); got != 0 { + t.Fatalf("impossible gaps must be ignored, got %d", got) + } + + stream.EventsLost(3) + if got := p.sequenceGaps.Load(); got != 3 { + t.Fatalf("expected a gap of 3, got %d", got) + } +} + +func TestReassemblyCountsRecordFieldErrors(t *testing.T) { + r, stream, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + + // a valid header with an address that cannot be decoded + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(90, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:90): saddr=ZZZZ`}, + proctitleRecord(90), + }) + + if got := p.recordErrors.Load(); got != 1 { + t.Fatalf("expected one record with unparsed fields, got %d", got) + } + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a connect without an address must not be dispatched, got %+v", got) + } +} + +func TestReassemblyHandlesShortSocketAddresses(t *testing.T) { + // go-libaudit before v2.6.0 panicked on these (elastic/go-libaudit#152) + // too short for the family, for AF_INET and for AF_INET6 + for _, saddr := range []string{"ZZ", "02", "", "0200", "02000050", "0A00", "0A0001BB"} { + t.Run(fmt.Sprintf("saddr=%q", saddr), func(t *testing.T) { + r, stream, rec, p := newTestReassembler(t, reassemblerMaxInFlight, reassemblerTimeout) + pushRecords(t, r, stream, []auditRecord{ + syscallRecord(91, 42, "curl", "/usr/bin/curl", "netmon"), + {auparse.AUDIT_SOCKADDR, `audit(1700000000.000:91): saddr=` + saddr}, + proctitleRecord(91), + }) + if got := rec.dispatched(); len(got) != 0 { + t.Fatalf("a connect without a usable address must not be dispatched, got %+v", got) + } + if got := p.recordErrors.Load(); got != 1 { + t.Fatalf("expected the record to be counted as not parsed, got %d", got) + } + }) + } +} + +func TestNoteLossLogsOnPowersOfTwo(t *testing.T) { + var counter atomic.Int64 + var logged []int64 + for i := 0; i < 20; i++ { + if noteLoss(&counter, 1, "test") { + logged = append(logged, counter.Load()) + } + } + want := []int64{1, 2, 4, 8, 16} + if fmt.Sprint(logged) != fmt.Sprint(want) { + t.Fatalf("logged at %v, want %v", logged, want) + } + + // a batch that crosses a power of two logs once + if !noteLoss(&counter, 20, "test") { + t.Fatalf("expected a log when the total goes from 20 to 40") + } +} + +func TestNoteReceiveError(t *testing.T) { + cases := []struct { + err error + wantOverflow int64 + wantOther int64 + }{ + {syscall.ENOBUFS, 1, 0}, + {os.NewSyscallError("recvfrom", syscall.ENOBUFS), 1, 0}, + {fmt.Errorf("receive: %w", syscall.ENOBUFS), 1, 0}, + {syscall.EINVAL, 0, 1}, + {errors.New("no buffer space available"), 0, 1}, + } + for _, c := range cases { + p := &ProcessMonitor{} + p.noteReceiveError(c.err) + if p.receiveOverflow.Load() != c.wantOverflow || p.receiveErrors.Load() != c.wantOther { + t.Errorf("noteReceiveError(%v): overflow=%d other=%d, want %d %d", c.err, p.receiveOverflow.Load(), p.receiveErrors.Load(), c.wantOverflow, c.wantOther) + } } }