fix(rule-engine): Check process state before evaluation

Events can arrive in non-deterministic order since they are published by independent providers. The side effect of this can translate in form of a missing process state. Imagine OpenProcess event arriving from the process before the
process' own creation event. To mitigate this bad behaviour, we always check the process state before evaluating  out of order events.

Author: rabbitstack

pkg/rules/engine.go

@@ -210,7 +210,7 @@ func (e *Engine) Compile() (*config.RulesCompileResult, error) {
 	for c, f := range filters {
 		var ss *sequenceState
 		if f.IsSequence() {
-			ss = newSequenceState(f, c)
+			ss = newSequenceState(f, c, e.psnap)
 		}
 		fltr := newCompiledFilter(f, c, ss)
 		if ss != nil {

pkg/rules/sequence.go

@@ -28,6 +28,7 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/kevent"
 	"github.com/rabbitstack/fibratus/pkg/kevent/kparams"
 	"github.com/rabbitstack/fibratus/pkg/kevent/ktypes"
+	"github.com/rabbitstack/fibratus/pkg/ps"
 	"github.com/rabbitstack/fibratus/pkg/util/atomic"
 	log "github.com/sirupsen/logrus"
 	"sort"
@@ -114,9 +115,11 @@ type sequenceState struct {
 	states map[fsm.State]bool
 	// smu guards the states map
 	smu sync.RWMutex
+
+	psnap ps.Snapshotter
 }
 
-func newSequenceState(f filter.Filter, c *config.FilterConfig) *sequenceState {
+func newSequenceState(f filter.Filter, c *config.FilterConfig, psnap ps.Snapshotter) *sequenceState {
 	ss := &sequenceState{
 		filter:        f,
 		seq:           f.GetSequence(),
@@ -129,6 +132,7 @@ func newSequenceState(f filter.Filter, c *config.FilterConfig) *sequenceState {
 		spanDeadlines: make(map[fsm.State]*time.Timer),
 		initialState:  sequenceInitialState,
 		inDeadline:    atomic.MakeBool(false),
+		psnap:         psnap,
 	}
 
 	ss.initFSM()
@@ -480,6 +484,10 @@ func (s *sequenceState) runSequence(e *kevent.Kevent) bool {
 						if !evt.ContainsMeta(kevent.RuleSequenceOOOKey) {
 							continue
 						}
+						// try to initialize process state before evaluating the event
+						if evt.PS == nil {
+							_, evt.PS = s.psnap.Find(evt.PID)
+						}
 						matches = s.filter.RunSequence(evt, seqID, s.partials, false)
 						// transition the state machine
 						if matches {

pkg/rules/sequence_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/kevent"
 	"github.com/rabbitstack/fibratus/pkg/kevent/kparams"
 	"github.com/rabbitstack/fibratus/pkg/kevent/ktypes"
+	"github.com/rabbitstack/fibratus/pkg/ps"
 	pstypes "github.com/rabbitstack/fibratus/pkg/ps/types"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -50,7 +51,7 @@ func TestSequenceState(t *testing.T) {
 
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	assert.Equal(t, 0, ss.currentState())
 	assert.True(t, ss.isInitialState())
@@ -190,7 +191,7 @@ func TestSimpleSequence(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	var tests = []struct {
 		evts    []*kevent.Kevent
@@ -276,7 +277,7 @@ func TestSimpleSequenceMultiplePartials(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	// create random matches which don't satisfy the sequence link
 	for i, pid := range []uint32{2343, 1024, 11122, 3450, 12319} {
@@ -382,7 +383,7 @@ func TestSimpleSequenceDeadline(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Type:      ktypes.CreateProcess,
@@ -453,7 +454,7 @@ func TestComplexSequence(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Seq:       1,
@@ -546,7 +547,7 @@ func TestSequenceOOO(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Type:      ktypes.CreateFile,
@@ -606,7 +607,7 @@ func TestSequenceGC(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e := &kevent.Kevent{
 		Type:      ktypes.OpenProcess,
@@ -755,7 +756,7 @@ func TestSequenceExpire(t *testing.T) {
 			f := filter.New(tt.expr, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 			require.NoError(t, f.Compile())
 
-			ss := newSequenceState(f, tt.c)
+			ss := newSequenceState(f, tt.c, new(ps.SnapshotterMock))
 			for _, evt := range tt.evts {
 				if evt.IsTerminateProcess() {
 					ss.expire(evt)
@@ -787,7 +788,7 @@ func TestSequenceBoundFields(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Type:      ktypes.CreateProcess,
@@ -882,7 +883,7 @@ func TestSequenceBoundFieldsWithFunctions(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true, EnableRegistryKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Type:     ktypes.CreateFile,
@@ -942,7 +943,7 @@ func TestIsExpressionEvaluable(t *testing.T) {
 	`, &config.Config{Kstream: config.KstreamConfig{EnableFileIOKevents: true}, Filters: &config.Filters{}})
 	require.NoError(t, f.Compile())
 
-	ss := newSequenceState(f, c)
+	ss := newSequenceState(f, c, new(ps.SnapshotterMock))
 
 	e1 := &kevent.Kevent{
 		Type: ktypes.CreateProcess,