feat: add event manager

The event manager manages the state (disabled/enabled) of events
globably on tracee.

pkg/ebpf/event_manager.go

@@ -0,0 +1,50 @@
+package ebpf
+
+import (
+	"sync"
+
+	"github.com/aquasecurity/tracee/pkg/events"
+)
+
+// eventManager is a thread-safe struct that manages the enabled events
+type eventManager struct {
+	mutex  sync.Mutex
+	events map[events.ID]bool
+}
+
+// newEventManager creates a new eventManager
+func newEventManager() *eventManager {
+	return &eventManager{
+		mutex:  sync.Mutex{},
+		events: make(map[events.ID]bool),
+	}
+}
+
+// IsEventEnabled returns true if a given event is enabled
+func (em *eventManager) IsEventEnabled(id events.ID) bool {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	v, ok := em.events[id]
+	if !ok {
+		return false
+	}
+
+	return v
+}
+
+// EnableEvent enables a given event
+func (em *eventManager) EnableEvent(id events.ID) {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	em.events[id] = true
+}
+
+// DisableEvent disables a given event
+func (em *eventManager) DisableEvent(id events.ID) {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	em.events[id] = false
+}

pkg/ebpf/event_manager_test.go

@@ -0,0 +1,104 @@
+package ebpf
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/aquasecurity/tracee/pkg/events"
+	"github.com/aquasecurity/tracee/pkg/policy"
+)
+
+func TestEventManagerEnableRule(t *testing.T) {
+	eventManager := newEventManager()
+
+	assert.False(t, eventManager.IsEventEnabled(events.SecurityBPF))
+	assert.False(t, eventManager.IsEventEnabled(events.SecurityFileOpen))
+	assert.False(t, eventManager.IsEventEnabled(events.SecuritySocketAccept))
+
+	eventManager.EnableEvent(events.SecurityBPF)
+	eventManager.EnableEvent(events.SecurityFileOpen)
+	eventManager.EnableEvent(events.SecuritySocketAccept)
+
+	assert.True(t, eventManager.IsEventEnabled(events.SecurityBPF))
+	assert.True(t, eventManager.IsEventEnabled(events.SecurityFileOpen))
+	assert.True(t, eventManager.IsEventEnabled(events.SecuritySocketAccept))
+}
+
+func TestEventsManagerDisableRule(t *testing.T) {
+	eventManager := newEventManager()
+
+	eventManager.EnableEvent(events.SecurityBPF)
+	eventManager.EnableEvent(events.SecurityFileOpen)
+	eventManager.EnableEvent(events.SecuritySocketAccept)
+
+	assert.True(t, eventManager.IsEventEnabled(events.SecurityBPF))
+	assert.True(t, eventManager.IsEventEnabled(events.SecurityFileOpen))
+	assert.True(t, eventManager.IsEventEnabled(events.SecuritySocketAccept))
+
+	eventManager.DisableEvent(events.SecurityBPF)
+	eventManager.DisableEvent(events.SecurityFileOpen)
+
+	assert.False(t, eventManager.IsEventEnabled(events.SecurityBPF))
+	assert.False(t, eventManager.IsEventEnabled(events.SecurityFileOpen))
+	assert.True(t, eventManager.IsEventEnabled(events.SecuritySocketAccept))
+}
+
+func TestEventsManagerEnableAndDisableConcurrent(t *testing.T) {
+	eventsToEnable := []events.ID{
+		events.SecurityBPF,
+		events.SchedGetPriorityMax,
+		events.SchedProcessExec,
+		events.SchedProcessExit,
+		events.Ptrace,
+	}
+
+	eventsToDisable := []events.ID{
+		events.SecurityBPFMap,
+		events.Openat2,
+		events.SchedProcessFork,
+		events.MagicWrite,
+		events.FileModification,
+	}
+
+	eventManager := newEventManager()
+
+	// activate events
+	for _, e := range eventsToDisable {
+		eventManager.EnableEvent(e)
+	}
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		for i := 0; i < policy.MaxPolicies; i++ {
+			for _, e := range eventsToEnable {
+				eventManager.EnableEvent(e)
+			}
+		}
+		wg.Done()
+	}()
+
+	wg.Add(1)
+	go func() {
+		for i := 0; i < policy.MaxPolicies; i++ {
+			for _, e := range eventsToDisable {
+				eventManager.DisableEvent(e)
+			}
+		}
+		wg.Done()
+	}()
+
+	wg.Wait()
+
+	for i := 0; i < policy.MaxPolicies; i++ {
+		for _, e := range eventsToEnable {
+			assert.True(t, eventManager.IsEventEnabled(e))
+		}
+		for _, e := range eventsToDisable {
+			assert.False(t, eventManager.IsEventEnabled(e))
+		}
+	}
+}

pkg/ebpf/events_pipeline.go

@@ -561,6 +561,13 @@ func (t *Tracee) sinkEvents(ctx context.Context, in <-chan *trace.Event) <-chan
 				continue // might happen during initialization (ctrl+c seg faults)
 			}
 
+			// Is the event disabled?
+			if !t.eventManager.IsEventEnabled(events.ID(event.EventID)) {
+				logger.Debugw("event dropped because it is disabled", "event", event.EventName)
+				t.eventsPool.Put(event)
+				continue
+			}
+
 			// Is the rule disabled for the matched policies?
 			if !t.policyManager.IsRuleEnabled(event.MatchedPoliciesUser, events.ID(event.EventID)) {
 				logger.Debugw("event dropped because matched rule disabled", "event", event.EventName)

pkg/ebpf/policy_manager_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/aquasecurity/tracee/pkg/policy"
 )
 
-func TestEventsManagerEnableRule(t *testing.T) {
+func TestPolicyManagerEnableRule(t *testing.T) {
 	policyManager := newPolicyManager()
 
 	policy1Mached := uint64(0b10)
@@ -33,7 +33,7 @@ func TestEventsManagerEnableRule(t *testing.T) {
 	assert.True(t, policyManager.IsRuleEnabled(policy1And2Mached, events.SecurityBPF))
 }
 
-func TestEventsManagerDisableRule(t *testing.T) {
+func TestPolicyManagerDisableRule(t *testing.T) {
 	policyManager := newPolicyManager()
 
 	policy1Mached := uint64(0b10)
@@ -53,7 +53,7 @@ func TestEventsManagerDisableRule(t *testing.T) {
 	assert.False(t, policyManager.IsRuleEnabled(policy1And2Mached, events.SecurityBPF))
 }
 
-func TestEventsManagerEnableAndDisableConcurrent(t *testing.T) {
+func TestPolicyManagerEnableAndDisableConcurrent(t *testing.T) {
 	eventsToEnable := []events.ID{
 		events.SecurityBPF,
 		events.SchedGetPriorityMax,

pkg/ebpf/tracee.go

@@ -121,6 +121,8 @@ type Tracee struct {
 	streamsManager *streams.StreamsManager
 	// policyManager manages policy state
 	policyManager *policyManager
+	// eventManager manages enabled events
+	eventManager *eventManager
 }
 
 func (t *Tracee) Stats() *metrics.Stats {
@@ -211,6 +213,7 @@ func New(cfg config.Config) (*Tracee, error) {
 	}
 
 	policyManager := newPolicyManager()
+	eventManager := newEventManager()
 
 	// Create Tracee
 
@@ -224,6 +227,7 @@ func New(cfg config.Config) (*Tracee, error) {
 		eventSignatures: make(map[events.ID]bool),
 		streamsManager:  streams.NewStreamsManager(),
 		policyManager:   policyManager,
+		eventManager:    eventManager,
 	}
 
 	// Initialize capabilities rings soon
@@ -253,6 +257,7 @@ func New(cfg config.Config) (*Tracee, error) {
 			utils.SetBit(&emit, uint(p.ID))
 			t.eventsState[e] = events.EventState{Submit: submit, Emit: emit}
 
+			eventManager.EnableEvent(e)
 			policyManager.EnableRule(p.ID, e)
 		}
 	}
@@ -1784,6 +1789,28 @@ func (t *Tracee) Unsubscribe(s *streams.Stream) {
 	t.streamsManager.Unsubscribe(s)
 }
 
+func (t *Tracee) EnableEvent(eventName string) error {
+	id, found := events.Core.GetDefinitionIDByName(eventName)
+	if !found {
+		return errfmt.Errorf("error event not found: %s", eventName)
+	}
+
+	t.eventManager.EnableEvent(id)
+
+	return nil
+}
+
+func (t *Tracee) DisableEvent(eventName string) error {
+	id, found := events.Core.GetDefinitionIDByName(eventName)
+	if !found {
+		return errfmt.Errorf("error event not found: %s", eventName)
+	}
+
+	t.eventManager.DisableEvent(id)
+
+	return nil
+}
+
 // EnableRule enables a rule in the specified policies
 func (t *Tracee) EnableRule(policyNames []string, ruleId string) error {
 	eventID, found := events.Core.GetDefinitionIDByName(ruleId)