refactor(eventsource): Unified security telemetry logger session

Unified security telemetry session serves as a container
for the events published by all ETW providers except the
core NT Kernel Logger provider. By enabling all providers
inside the same session, we can preserve event ordering and
save extra resources allocated for the ETW session buffers.

Author: rabbitstack

internal/etw/source.go

@@ -32,7 +32,6 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/sys/etw"
 	"github.com/rabbitstack/fibratus/pkg/util/multierror"
 	log "github.com/sirupsen/logrus"
-	"golang.org/x/sys/windows"
 	"golang.org/x/sys/windows/registry"
 	"time"
 )
@@ -162,16 +161,34 @@ func (e *EventSource) Open(config *config.Config) error {
 		}
 	}
 
-	e.addTrace(etw.KernelLoggerSession, etw.KernelTraceControlGUID)
+	// add the core NT Kernel Logger trace
+	e.addTrace(NewKernelTrace(config))
+
+	// security telemetry trace hosts remaining ETW providers
+	trace := NewTrace(etw.SecurityTelemetrySession, config)
 
 	if config.EventSource.EnableDNSEvents {
-		e.addTrace(etw.DNSClientSession, etw.DNSClientGUID)
+		trace.AddProvider(etw.DNSClientGUID, etw.NoKeywords, false)
 	}
+
 	if config.EventSource.EnableAuditAPIEvents {
-		e.addTrace(etw.KernelAuditAPICallsSession, etw.KernelAuditAPICallsGUID)
+		trace.AddProvider(etw.KernelAuditAPICallsGUID, etw.NoKeywords, config.EventSource.StackEnrichment)
 	}
+
 	if config.EventSource.EnableThreadpoolEvents {
-		e.addTrace(etw.ThreadpoolSession, etw.ThreadpoolGUID)
+		// thread pool provider must be configured with
+		// stack extensions to activate stack walks events
+		var stackexts *StackExtensions
+		if e.config.EventSource.StackEnrichment {
+			stackexts = NewStackExtensions(config.EventSource)
+			stackexts.EnableThreadpoolCallstack()
+		}
+		trace.AddProvider(etw.ThreadpoolGUID, etw.NoKeywords, config.EventSource.StackEnrichment, WithStackExts(stackexts))
+	}
+
+	if trace.HasProviders() {
+		// add security telemetry trace
+		e.addTrace(trace)
 	}
 
 	for _, trace := range e.traces {
@@ -305,6 +322,6 @@ func (e *EventSource) RegisterEventListener(lis event.Listener) {
 	e.listeners = append(e.listeners, lis)
 }
 
-func (e *EventSource) addTrace(name string, guid windows.GUID) {
-	e.traces = append(e.traces, NewTrace(name, guid, 0x0, e.config))
+func (e *EventSource) addTrace(trace *Trace) {
+	e.traces = append(e.traces, trace)
 }

internal/etw/source_test.go

@@ -101,7 +101,7 @@ func TestEventSourceStartTraces(t *testing.T) {
 			1,
 			[]etw.EventTraceFlags{0x6018203, 0},
 		},
-		{"start kernel logger and audit api sessions",
+		{"start kernel and security telemetry logger sessions",
 			&config.Config{
 				EventSource: config.EventSourceConfig{
 					EnableThreadEvents:   true,

internal/etw/stackext.go

@@ -54,6 +54,9 @@ func (s *StackExtensions) AddStackTracingWith(guid windows.GUID, hookID uint16)
 // EventIds returns all event types eligible for stack tracing.
 func (s *StackExtensions) EventIds() []etw.ClassicEventID { return s.ids }
 
+// Empty determines if this stack extensions has registered event identifiers.
+func (s *StackExtensions) Empty() bool { return len(s.ids) == 0 }
+
 // EnableProcessCallstack populates the stack identifiers
 // with event types eligible for emitting stack walk events
 // related to process telemetry, such as creating a process,

internal/etw/trace.go

@@ -72,16 +72,10 @@ func initEventTraceProps(c config.EventSourceConfig) etw.EventTraceProperties {
 	}
 }
 
-// Trace is the essential building block for controlling
-// trace sessions and configuring event consumers. Such
-// operations include starting, stopping, and flushing
-// trace sessions, and opening the trace for processing
-// and event consumption.
-type Trace struct {
-	// Name represents the unique tracing session name.
-	Name string
+// ProviderInfo describes ETW provider metadata.
+type ProviderInfo struct {
 	// GUID is the globally unique identifier for the
-	// ETW provider.
+	// ETW provider for which the session is started.
 	GUID windows.GUID
 	// Keywords is the bitmask of keywords that determine
 	// the categories of events for the provider to emit.
@@ -91,11 +85,42 @@ type Trace struct {
 	// for providers that are enabled via etw.EnableProvider
 	// API.
 	Keywords uint64
-
+	// EnableStacks indicates if callstacks are enabled for
+	// this provider.
+	EnableStacks bool
 	// stackExtensions manager stack tracing enablement.
 	// For each event present in the stack identifiers,
 	// the StackWalk event is published by the provider.
 	stackExtensions *StackExtensions
+}
+
+func (p *ProviderInfo) HasStackExtensions() bool {
+	return p.stackExtensions != nil && !p.stackExtensions.Empty()
+}
+
+// Trace is the essential building block for controlling
+// trace sessions and configuring event consumers. Such
+// operations include starting, stopping, and flushing
+// trace sessions, and opening the trace for processing
+// and event consumption. Trace can be configured to
+// operate a single ETW provider, or it can act as a
+// container for multiple provider sessions.
+type Trace struct {
+	// Name represents the unique tracing session name.
+	Name string
+	// GUID is the globally unique identifier for the
+	// ETW provider for which the session is started.
+	GUID windows.GUID
+
+	// Providers is the list of providers to be run inside
+	// the tracing session. For each provider, the GUID,
+	// keywords and other parameters can be specified.
+	Providers []ProviderInfo
+
+	// stackExtensions manages stack tracing enablement.
+	// For each event present in the stack identifiers,
+	// the StackWalk event is published by the provider.
+	stackExtensions *StackExtensions
 
 	// startHandle is the session handle returned by the
 	// etw.StartTrace function. This handle is
@@ -120,13 +145,59 @@ type Trace struct {
 	errs chan error
 }
 
-// NewTrace creates a new trace with specified name, provider GUID, and keywords.
-func NewTrace(name string, guid windows.GUID, keywords uint64, config *config.Config) *Trace {
-	t := &Trace{Name: name, GUID: guid, Keywords: keywords, stackExtensions: NewStackExtensions(config.EventSource), config: config}
+type opts struct {
+	stackexts *StackExtensions
+}
+
+// Option represents the option for the trace.
+type Option func(o *opts)
+
+// WithStackExts sets the stack extensions.
+func WithStackExts(stackexts *StackExtensions) Option {
+	return func(o *opts) {
+		o.stackexts = stackexts
+	}
+}
+
+// NewKernelTrace creates a new NT Kernel Logger trace.
+func NewKernelTrace(config *config.Config) *Trace {
+	t := &Trace{Name: etw.KernelLoggerSession, GUID: etw.KernelTraceControlGUID, stackExtensions: NewStackExtensions(config.EventSource), config: config}
 	t.enableCallstacks()
 	return t
 }
 
+// NewTrace creates a new trace that can host various ETW provider sessions.
+// The providers to be run inside the session can be given in the last argument
+// or added by the AddProvider method.
+func NewTrace(name string, config *config.Config, providers ...ProviderInfo) *Trace {
+	t := &Trace{Name: name, config: config, Providers: make([]ProviderInfo, 0)}
+	t.Providers = providers
+	return t
+}
+
+// AddProvider adds a new provider to the multi trace session
+// with optional parameters that influence the provider.
+func (t *Trace) AddProvider(guid windows.GUID, keywords uint64, enableStacks bool, options ...Option) {
+	var opts opts
+
+	for _, opt := range options {
+		opt(&opts)
+	}
+
+	t.Providers = append(t.Providers, ProviderInfo{GUID: guid, Keywords: keywords, EnableStacks: enableStacks, stackExtensions: opts.stackexts})
+}
+
+// HasProviders determines if this trace contains providers.
+func (t *Trace) HasProviders() bool { return len(t.Providers) > 0 }
+
+// IsGUIDEmpty determines if the provider GUID is empty.
+func (t *Trace) IsGUIDEmpty() bool {
+	return t.GUID.Data1 == 0 &&
+		t.GUID.Data2 == 0 &&
+		t.GUID.Data3 == 0 &&
+		t.GUID.Data4 == [8]byte{}
+}
+
 func (t *Trace) enableCallstacks() {
 	if t.IsKernelTrace() {
 		t.stackExtensions.EnableProcessCallstack()
@@ -137,10 +208,6 @@ func (t *Trace) enableCallstacks() {
 
 		t.stackExtensions.EnableMemoryCallstack()
 	}
-
-	if t.IsThreadpoolTrace() {
-		t.stackExtensions.EnableThreadpoolCallstack()
-	}
 }
 
 // Start registers and starts an event tracing session.
@@ -151,6 +218,11 @@ func (t *Trace) Start() error {
 	if len(t.Name) > maxLoggerNameSize {
 		return fmt.Errorf("trace name [%s] is too long", t.Name)
 	}
+
+	if !t.IsGUIDEmpty() && t.HasProviders() {
+		return fmt.Errorf("%s trace has the root GUID set but providers are not empty", t.Name)
+	}
+
 	cfg := t.config.EventSource
 	props := initEventTraceProps(cfg)
 	flags := t.enableFlagsDynamically(cfg)
@@ -212,21 +284,30 @@ func (t *Trace) Start() error {
 		return etw.SetTraceSystemFlags(handle, sysTraceFlags)
 	}
 
-	// if we're starting a trace for non-system logger, the call
-	// to etw.EnableTrace is needed to configure how an ETW provider
-	// publishes events to the trace session. For instance, if stack
-	// enrichment is enabled, it is necessary to instruct the provider
-	// to emit stack addresses in the extended data item section when
-	// writing events to the session buffers
-	if cfg.StackEnrichment && !t.IsThreadpoolTrace() {
-		return etw.EnableTraceWithOpts(t.GUID, t.startHandle, t.Keywords, etw.EnableTraceOpts{WithStacktrace: true})
-	} else if cfg.StackEnrichment && len(t.stackExtensions.EventIds()) > 0 {
-		if err := etw.EnableStackTracing(t.startHandle, t.stackExtensions.EventIds()); err != nil {
-			return fmt.Errorf("fail to enable system events callstack tracing: %v", err)
+	// For each provider in multi trace, the call to etw.EnableTrace is
+	// needed to configure how an ETW provider publishes events to the
+	// trace session.
+	// For instance, if stack enrichment is enabled, it is necessary to
+	// instruct the provider to emit stack addresses in the extended
+	// data item section when writing events to the session buffers
+	for _, provider := range t.Providers {
+		switch {
+		case provider.EnableStacks && provider.HasStackExtensions():
+			if err := etw.EnableStackTracing(t.startHandle, provider.stackExtensions.EventIds()); err != nil {
+				return fmt.Errorf("fail to enable provider callstack tracing: %v", err)
+			}
+			if err := etw.EnableTrace(provider.GUID, t.startHandle, provider.Keywords); err != nil {
+				return err
+			}
+		case provider.EnableStacks:
+			opts := etw.EnableTraceOpts{WithStacktrace: true}
+			if err := etw.EnableTraceWithOpts(provider.GUID, t.startHandle, provider.Keywords, opts); err != nil {
+				return err
+			}
 		}
 	}
 
-	return etw.EnableTrace(t.GUID, t.startHandle, t.Keywords)
+	return nil
 }
 
 // IsStarted indicates if the trace is started successfully.
@@ -317,9 +398,6 @@ func (t *Trace) Close() error {
 // IsKernelTrace determines if this is the system logger trace.
 func (t *Trace) IsKernelTrace() bool { return t.GUID == etw.KernelTraceControlGUID }
 
-// IsThreadpoolTrace determines if this is the thread pool logger trace.
-func (t *Trace) IsThreadpoolTrace() bool { return t.GUID == etw.ThreadpoolGUID }
-
 // enableFlagsDynamically crafts the system logger event mask
 // depending on the compiled rules result or the config state.
 // System logger flags is a bitmask that indicates which kernel events

internal/etw/trace_test.go

@@ -20,7 +20,6 @@ package etw
 
 import (
 	"github.com/rabbitstack/fibratus/pkg/config"
-	"github.com/rabbitstack/fibratus/pkg/sys/etw"
 	"github.com/stretchr/testify/require"
 	"testing"
 	"time"
@@ -37,7 +36,7 @@ func TestStartTrace(t *testing.T) {
 		},
 	}
 
-	trace := NewTrace(etw.KernelLoggerSession, etw.KernelTraceControlGUID, 0, cfg)
+	trace := NewKernelTrace(cfg)
 	require.NoError(t, trace.Start())
 	require.True(t, trace.IsStarted())
 	defer trace.Stop()

pkg/event/types_windows.go

@@ -35,12 +35,10 @@ type Source uint8
 const (
 	// SystemLogger event is emitted by the system provider
 	SystemLogger Source = iota
-	// AuditAPICallsLogger event is emitted by Audit API calls provider
-	AuditAPICallsLogger
-	// DNSLogger event is emitted by DNS provider
-	DNSLogger
-	// ThreadpoolLogger event is emitted by thread pool provider
-	ThreadpoolLogger
+	// SecurityTelemetryLogger event is emitted by the combination of multiple providers.
+	// Most notably, DNS, thread pool, and kernel audit API providers are in charge of
+	// publishing the events.
+	SecurityTelemetryLogger
 )
 
 // Type identifies an event type. It comprises the event GUID + hook ID to uniquely identify the event
@@ -578,27 +576,15 @@ func (t *Type) HookID() uint16 {
 // Source designates the provenance of this event type.
 func (t Type) Source() Source {
 	switch t {
-	case OpenProcess, OpenThread, SetThreadContext, CreateSymbolicLinkObject:
-		return AuditAPICallsLogger
-	case QueryDNS, ReplyDNS:
-		return DNSLogger
-	case SubmitThreadpoolWork, SubmitThreadpoolCallback, SetThreadpoolTimer:
-		return ThreadpoolLogger
+	case OpenProcess, OpenThread, SetThreadContext, CreateSymbolicLinkObject,
+		QueryDNS, ReplyDNS, SubmitThreadpoolWork, SubmitThreadpoolCallback,
+		SetThreadpoolTimer:
+		return SecurityTelemetryLogger
 	default:
 		return SystemLogger
 	}
 }
 
-// CanArriveOutOfOrder indicates if the event can be
-// emitted by the provider in out-of-order fashion, i.e.
-// its timestamp is perfectly aligned in relation to other
-// events, but it appears first on the consumer callback
-// before other events published before it.
-func (t Type) CanArriveOutOfOrder() bool {
-	return t.Category() == Threadpool || t.Subcategory() == DNS ||
-		t == OpenProcess || t == OpenThread || t == SetThreadContext || t == CreateSymbolicLinkObject
-}
-
 // TypeFromParts builds the event type from provider GUID and hook ID.
 func TypeFromParts(g windows.GUID, id uint16) Type { return pack(g, id) }
 

pkg/event/types_windows_test.go

@@ -128,9 +128,3 @@ func TestGUIDAndHookIDFromEventType(t *testing.T) {
 		})
 	}
 }
-
-func TestCanArriveOutOfOrder(t *testing.T) {
-	assert.False(t, RegSetValue.CanArriveOutOfOrder())
-	assert.False(t, VirtualAlloc.CanArriveOutOfOrder())
-	assert.True(t, OpenProcess.CanArriveOutOfOrder())
-}

pkg/filter/ql/literal.go

@@ -21,7 +21,6 @@ package ql
 import (
 	"github.com/rabbitstack/fibratus/pkg/event"
 	"github.com/rabbitstack/fibratus/pkg/filter/fields"
-	"golang.org/x/sys/windows"
 	"net"
 	"reflect"
 	"strconv"
@@ -379,24 +378,19 @@ func (s Sequence) IsConstrained() bool {
 }
 
 func (s *Sequence) init() {
-	// determine if the sequence references
-	// an event type that can arrive out-of-order.
-	// The edge case is for unordered events emitted
-	// by the same provider where the temporal order
-	// is guaranteed
-	guids := make(map[windows.GUID]bool)
+	// determine if the sequence references an event type
+	// that can arrive out-of-order. This happens if the
+	// expressions in the sequence reference event types
+	// from different event sources
+	sources := make(map[event.Source]bool)
+
 	for _, expr := range s.Expressions {
 		for _, etype := range expr.types {
-			if etype.CanArriveOutOfOrder() {
-				s.IsUnordered = true
-			}
-			guids[etype.GUID()] = true
+			sources[etype.Source()] = true
 		}
 	}
 
-	if s.IsUnordered && len(guids) == 1 {
-		s.IsUnordered = false
-	}
+	s.IsUnordered = len(sources) > 1
 }
 
 func (s Sequence) impairBy() bool {