events_derived: merge into existing files

Move the derivation pipeline stage to events_pipeline
Move the derivation table initialization to tracee.go

pkg/ebpf/events_pipeline.go

@@ -2,7 +2,7 @@ package ebpf
 
 import (
 	"bytes"
-	gocontext "context"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"strconv"
@@ -11,6 +11,7 @@ import (
 
 	"github.com/aquasecurity/tracee/pkg/bufferdecoder"
 	"github.com/aquasecurity/tracee/pkg/events"
+	"github.com/aquasecurity/tracee/pkg/events/derive"
 	"github.com/aquasecurity/tracee/types/trace"
 )
 
@@ -19,7 +20,7 @@ import (
 const maxStackDepth int = 20
 
 // handleEvents is a high-level function that starts all operations related to events processing
-func (t *Tracee) handleEvents(ctx gocontext.Context) {
+func (t *Tracee) handleEvents(ctx context.Context) {
 	var errcList []<-chan error
 
 	// Source pipeline stage.
@@ -87,7 +88,7 @@ func (t *Tracee) handleEvents(ctx gocontext.Context) {
 // 3) create an internal, to tracee-ebpf, buffer based on the node size.
 
 // queueEvents implements an internal FIFO queue for caching events
-func (t *Tracee) queueEvents(ctx gocontext.Context, in <-chan *trace.Event) (chan *trace.Event, chan error) {
+func (t *Tracee) queueEvents(ctx context.Context, in <-chan *trace.Event) (chan *trace.Event, chan error) {
 	out := make(chan *trace.Event, 10000)
 	errc := make(chan error, 1)
 	done := make(chan struct{}, 1)
@@ -129,7 +130,7 @@ func (t *Tracee) queueEvents(ctx gocontext.Context, in <-chan *trace.Event) (cha
 }
 
 // decodeEvents read the events received from the BPF programs and parse it into trace.Event type
-func (t *Tracee) decodeEvents(outerCtx gocontext.Context) (<-chan *trace.Event, <-chan error) {
+func (t *Tracee) decodeEvents(outerCtx context.Context) (<-chan *trace.Event, <-chan error) {
 	out := make(chan *trace.Event, 10000)
 	errc := make(chan error, 1)
 	go func() {
@@ -234,7 +235,7 @@ func parseContextFlags(flags uint32) trace.ContextFlags {
 	}
 }
 
-func (t *Tracee) processEvents(ctx gocontext.Context, in <-chan *trace.Event) (<-chan *trace.Event, <-chan error) {
+func (t *Tracee) processEvents(ctx context.Context, in <-chan *trace.Event) (<-chan *trace.Event, <-chan error) {
 	out := make(chan *trace.Event, 10000)
 	errc := make(chan error, 1)
 	go func() {
@@ -270,7 +271,41 @@ func (t *Tracee) processEvents(ctx gocontext.Context, in <-chan *trace.Event) (<
 	return out, errc
 }
 
-func (t *Tracee) sinkEvents(ctx gocontext.Context, in <-chan *trace.Event) <-chan error {
+// deriveEvents is the derivation pipeline stage
+func (t *Tracee) deriveEvents(ctx context.Context, in <-chan *trace.Event) (<-chan *trace.Event, <-chan error) {
+	out := make(chan *trace.Event)
+	errc := make(chan error, 1)
+
+	go func() {
+		defer close(out)
+		defer close(errc)
+
+		for {
+			select {
+			case event := <-in:
+				out <- event
+
+				// Derive event before parse its arguments
+				derivatives, errors := derive.DeriveEvent(*event, t.eventDerivations)
+
+				for _, err := range errors {
+					t.handleError(err)
+				}
+
+				for _, derivative := range derivatives {
+					out <- &derivative
+				}
+
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return out, errc
+}
+
+func (t *Tracee) sinkEvents(ctx context.Context, in <-chan *trace.Event) <-chan error {
 	errc := make(chan error, 1)
 
 	go func() {

pkg/ebpf/tracee.go

@@ -33,6 +33,7 @@ import (
 	"github.com/aquasecurity/tracee/pkg/metrics"
 	"github.com/aquasecurity/tracee/pkg/procinfo"
 	"github.com/aquasecurity/tracee/pkg/utils"
+	"github.com/aquasecurity/tracee/pkg/utils/sharedobjs"
 	"github.com/aquasecurity/tracee/types/trace"
 	lru "github.com/hashicorp/golang-lru"
 	"golang.org/x/sys/unix"
@@ -511,6 +512,71 @@ func (t *Tracee) initTailCall(mapName string, mapIndexes []uint32, progName stri
 	return nil
 }
 
+// initDerivationTable initializes tracee's events.DerivationTable.
+// we declare for each Event (represented through it's ID) to which other
+// events it can be derived and the corresponding function to derive into that Event.
+func (t *Tracee) initDerivationTable() error {
+	// sanity check for containers dependency
+	if t.containers == nil {
+		return fmt.Errorf("nil tracee containers")
+	}
+
+	pathResolver := containers.InitPathResolver(&t.pidsInMntns)
+	soLoader := sharedobjs.InitContainersSymbolsLoader(&pathResolver, 1024)
+
+	t.eventDerivations = derive.Table{
+		events.CgroupMkdir: {
+			events.ContainerCreate: {
+				Enabled:        t.events[events.ContainerCreate].submit,
+				DeriveFunction: derive.ContainerCreate(t.containers),
+			},
+		},
+		events.CgroupRmdir: {
+			events.ContainerRemove: {
+				Enabled:        t.events[events.ContainerRemove].submit,
+				DeriveFunction: derive.ContainerRemove(t.containers),
+			},
+		},
+		events.PrintSyscallTable: {
+			events.HookedSyscalls: {
+				Enabled:        t.events[events.PrintSyscallTable].submit,
+				DeriveFunction: derive.DetectHookedSyscall(t.kernelSymbols),
+			},
+		},
+		events.DnsRequest: {
+			events.NetPacket: {
+				Enabled:        t.events[events.NetPacket].submit,
+				DeriveFunction: derive.NetPacket(),
+			},
+		},
+		events.DnsResponse: {
+			events.NetPacket: {
+				Enabled:        t.events[events.NetPacket].submit,
+				DeriveFunction: derive.NetPacket(),
+			},
+		},
+		events.PrintNetSeqOps: {
+			events.HookedSeqOps: {
+				Enabled:        t.events[events.HookedSeqOps].submit,
+				DeriveFunction: derive.HookedSeqOps(t.kernelSymbols),
+			},
+		},
+		events.SharedObjectLoaded: {
+			events.SymbolsLoaded: {
+				Enabled: t.events[events.SymbolsLoaded].submit,
+				DeriveFunction: derive.SymbolsLoaded(
+					soLoader,
+					t.config.Filter.ArgFilter.Filters[events.SymbolsLoaded]["symbols"].Equal,
+					t.config.Filter.ArgFilter.Filters[events.SymbolsLoaded]["library_path"].NotEqual,
+					t.config.Debug,
+				),
+			},
+		},
+	}
+
+	return nil
+}
+
 // options config should match defined values in ebpf code
 const (
 	optDetectOrigSyscall uint32 = 1 << iota