feat(etw): Windows Kernel Registry provider for captured value data enrichment

internal/etw/consumer.go

@@ -23,7 +23,6 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/config"
 	"github.com/rabbitstack/fibratus/pkg/event"
 	"github.com/rabbitstack/fibratus/pkg/filter"
-	"github.com/rabbitstack/fibratus/pkg/handle"
 	"github.com/rabbitstack/fibratus/pkg/ps"
 	"github.com/rabbitstack/fibratus/pkg/sys/etw"
 )
@@ -47,15 +46,15 @@ type Consumer struct {
 // NewConsumer builds a new event consumer.
 func NewConsumer(
 	psnap ps.Snapshotter,
-	hsnap handle.Snapshotter,
 	config *config.Config,
 	sequencer *event.Sequencer,
 	evts chan *event.Event,
+	processors processors.Chain,
 ) *Consumer {
 	return &Consumer{
 		q:          event.NewQueueWithChannel(evts, config.EventSource.StackEnrichment, config.ForwardMode || config.IsCaptureSet()),
 		sequencer:  sequencer,
-		processors: processors.NewChain(psnap, hsnap, config),
+		processors: processors,
 		psnap:      psnap,
 		config:     config,
 	}

internal/etw/processors/registry_windows.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -41,11 +42,20 @@ var (
 		return fmt.Errorf("unable to read value %s : %v", filepath.Join(key, subkey), err)
 	}
 
+	// valueTTL specifies the maximum allowed period for RegSetValueInternal events
+	// to remain in the queue
+	valueTTL = time.Minute * 2
+	// valuePurgerInterval specifies the purge interval for stale values
+	valuePurgerInterval = time.Minute
+
 	// kcbCount counts the total KCBs found during the duration of the kernel session
 	kcbCount      = expvar.NewInt("registry.kcb.count")
 	kcbMissCount  = expvar.NewInt("registry.kcb.misses")
 	keyHandleHits = expvar.NewInt("registry.key.handle.hits")
 
+	readValueOps     = expvar.NewInt("registry.read.value.ops")
+	capturedDataHits = expvar.NewInt("registry.data.hits")
+
 	handleThrottleCount uint32
 )
 
@@ -57,6 +67,13 @@ type registryProcessor struct {
 	// keys stores the mapping between the KCB (Key Control Block) and the key name.
 	keys  map[uint64]string
 	hsnap handle.Snapshotter
+
+	values map[uint32][]*event.Event
+	mu     sync.Mutex
+
+	purger *time.Ticker
+
+	quit chan struct{}
 }
 
 func newRegistryProcessor(hsnap handle.Snapshotter) Processor {
@@ -68,10 +85,18 @@ func newRegistryProcessor(hsnap handle.Snapshotter) Processor {
 			atomic.StoreUint32(&handleThrottleCount, 0)
 		}
 	}()
-	return &registryProcessor{
-		keys:  make(map[uint64]string),
-		hsnap: hsnap,
+
+	r := &registryProcessor{
+		keys:   make(map[uint64]string),
+		hsnap:  hsnap,
+		values: make(map[uint32][]*event.Event),
+		purger: time.NewTicker(valuePurgerInterval),
+		quit:   make(chan struct{}, 1),
 	}
+
+	go r.housekeep()
+
+	return r
 }
 
 func (r *registryProcessor) ProcessEvent(e *event.Event) (*event.Event, bool, error) {
@@ -93,6 +118,12 @@ func (r *registryProcessor) processEvent(e *event.Event) (*event.Event, error) {
 		delete(r.keys, khandle)
 		kcbCount.Add(-1)
 	default:
+		if e.IsRegSetValueInternal() {
+			// store the event in temporary queue
+			r.pushSetValue(e)
+			return e, nil
+		}
+
 		khandle := e.Params.MustGetUint64(params.RegKeyHandle)
 		// we have to obey a straightforward algorithm to connect relative
 		// key names to their root keys. If key handle is equal to zero we
@@ -116,7 +147,32 @@ func (r *registryProcessor) processEvent(e *event.Event) (*event.Event, error) {
 			}
 		}
 
-		// get the type/value of the registry key and append to parameters
+		if e.IsRegSetValue() {
+			// previously stored RegSetValueInternal event
+			// is popped from the queue. RegSetValue can
+			// be enriched with registry value type/data
+			v := r.popSetValue(e)
+			if v == nil {
+				// try to read captured data from userspace
+				goto readValue
+			}
+
+			capturedDataHits.Add(1)
+
+			// enrich the event with value data/type parameters
+			typ, err := v.Params.GetUint32(params.RegValueType)
+			if err == nil {
+				e.AppendEnum(params.RegValueType, typ, key.RegistryValueTypes)
+			}
+			data, err := v.Params.Get(params.RegData)
+			if err == nil {
+				e.AppendParam(params.RegData, data.Type, data.Value)
+			}
+
+			return e, nil
+		}
+
+	readValue:
 		if !e.IsRegSetValue() || !e.IsSuccess() {
 			return e, nil
 		}
@@ -126,36 +182,42 @@ func (r *registryProcessor) processEvent(e *event.Event) (*event.Event, error) {
 			return e, nil
 		}
 
+		// get the type/value of the registry key and append to parameters
 		rootkey, subkey := key.Format(keyName)
-		if rootkey != key.Invalid {
-			typ, val, err := rootkey.ReadValue(subkey)
-			if err != nil {
-				errno, ok := err.(windows.Errno)
-				if ok && (errno.Is(os.ErrNotExist) || err == windows.ERROR_ACCESS_DENIED) {
-					return e, nil
-				}
-				return e, ErrReadValue(rootkey.String(), keyName, err)
-			}
-			e.AppendEnum(params.RegValueType, typ, key.RegistryValueTypes)
-			switch typ {
-			case registry.SZ, registry.EXPAND_SZ:
-				e.AppendParam(params.RegValue, params.UnicodeString, val)
-			case registry.MULTI_SZ:
-				e.AppendParam(params.RegValue, params.Slice, val)
-			case registry.BINARY:
-				e.AppendParam(params.RegValue, params.Binary, val)
-			case registry.QWORD:
-				e.AppendParam(params.RegValue, params.Uint64, val)
-			case registry.DWORD:
-				e.AppendParam(params.RegValue, params.Uint32, uint32(val.(uint64)))
+		if rootkey == key.Invalid {
+			return e, nil
+		}
+
+		readValueOps.Add(1)
+		typ, val, err := rootkey.ReadValue(subkey)
+		if err != nil {
+			errno, ok := err.(windows.Errno)
+			if ok && (errno.Is(os.ErrNotExist) || err == windows.ERROR_ACCESS_DENIED) {
+				return e, nil
 			}
+			return e, ErrReadValue(rootkey.String(), keyName, err)
+		}
+		e.AppendEnum(params.RegValueType, typ, key.RegistryValueTypes)
+
+		switch typ {
+		case registry.SZ, registry.EXPAND_SZ:
+			e.AppendParam(params.RegData, params.UnicodeString, val)
+		case registry.MULTI_SZ:
+			e.AppendParam(params.RegData, params.Slice, val)
+		case registry.BINARY:
+			e.AppendParam(params.RegData, params.Binary, val)
+		case registry.QWORD:
+			e.AppendParam(params.RegData, params.Uint64, val)
+		case registry.DWORD:
+			e.AppendParam(params.RegData, params.Uint32, uint32(val.(uint64)))
 		}
 	}
+
 	return e, nil
 }
 
-func (registryProcessor) Name() ProcessorType { return Registry }
-func (registryProcessor) Close()              {}
+func (*registryProcessor) Name() ProcessorType { return Registry }
+func (r *registryProcessor) Close()            { r.quit <- struct{}{} }
 
 func (r *registryProcessor) findMatchingKey(pid uint32, relativeKeyName string) string {
 	// we want to prevent too frequent queries on the process' handles
@@ -166,10 +228,12 @@ func (r *registryProcessor) findMatchingKey(pid uint32, relativeKeyName string)
 	if atomic.LoadUint32(&handleThrottleCount) > maxHandleQueries {
 		return relativeKeyName
 	}
+
 	handles, err := r.hsnap.FindHandles(pid)
 	if err != nil {
 		return relativeKeyName
 	}
+
 	for _, h := range handles {
 		if h.Type != handle.Key {
 			continue
@@ -179,5 +243,71 @@ func (r *registryProcessor) findMatchingKey(pid uint32, relativeKeyName string)
 			return h.Name
 		}
 	}
+
 	return relativeKeyName
 }
+
+// pushSetValue stores the internal RegSetValue event
+// into per process identifier queue.
+func (r *registryProcessor) pushSetValue(e *event.Event) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	vals, ok := r.values[e.PID]
+	if !ok {
+		r.values[e.PID] = []*event.Event{e}
+	} else {
+		r.values[e.PID] = append(vals, e)
+	}
+}
+
+// popSetValue traverses the internal RegSetValue queue
+// and pops the event if the suffixes match.
+func (r *registryProcessor) popSetValue(e *event.Event) *event.Event {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	vals, ok := r.values[e.PID]
+	if !ok {
+		return nil
+	}
+
+	var v *event.Event
+	for i := len(vals) - 1; i >= 0; i-- {
+		val := vals[i]
+		if strings.HasSuffix(e.GetParamAsString(params.RegPath), val.GetParamAsString(params.RegPath)) {
+			v = val
+			r.values[e.PID] = append(vals[:i], vals[i+1:]...)
+			break
+		}
+	}
+
+	return v
+}
+
+func (r *registryProcessor) valuesSize(pid uint32) int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.values[pid])
+}
+
+func (r *registryProcessor) housekeep() {
+	for {
+		select {
+		case <-r.purger.C:
+			r.mu.Lock()
+			for pid, vals := range r.values {
+				for i, val := range vals {
+					if time.Since(val.Timestamp) < valueTTL {
+						continue
+					}
+					r.values[pid] = append(vals[:i], vals[i+1:]...)
+				}
+				if len(vals) == 0 {
+					delete(r.values, pid)
+				}
+			}
+			r.mu.Unlock()
+		case <-r.quit:
+			return
+		}
+	}
+}

internal/etw/processors/registry_windows_test.go

@@ -23,11 +23,18 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/event/params"
 	"github.com/rabbitstack/fibratus/pkg/handle"
 	htypes "github.com/rabbitstack/fibratus/pkg/handle/types"
+	"github.com/rabbitstack/fibratus/pkg/util/key"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"testing"
+	"time"
 )
 
+func init() {
+	valueTTL = time.Millisecond * 150
+	valuePurgerInterval = time.Millisecond * 300
+}
+
 func TestRegistryProcessor(t *testing.T) {
 	var tests = []struct {
 		name           string
@@ -161,7 +168,53 @@ func TestRegistryProcessor(t *testing.T) {
 			func(e *event.Event, t *testing.T, hsnap *handle.SnapshotterMock, p Processor) {
 				assert.Equal(t, `HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Windows\Directory`, e.GetParamAsString(params.RegPath))
 				assert.Equal(t, `REG_EXPAND_SZ`, e.GetParamAsString(params.RegValueType))
-				assert.Equal(t, `%SystemRoot%`, e.GetParamAsString(params.RegValue))
+				assert.Equal(t, `%SystemRoot%`, e.GetParamAsString(params.RegData))
+			},
+		},
+		{
+			"process registry set value from internal event",
+			&event.Event{
+				Type:     event.RegSetValue,
+				Category: event.Registry,
+				PID:      23234,
+				Params: event.Params{
+					params.RegPath:      {Name: params.RegPath, Type: params.Key, Value: `\REGISTRY\MACHINE\SYSTEM\CurrentControlSet\Control\Windows\Directory`},
+					params.RegKeyHandle: {Name: params.RegKeyHandle, Type: params.Uint64, Value: uint64(0)},
+				},
+			},
+			func(p Processor) {
+				p.(*registryProcessor).values[23234] = []*event.Event{
+					{
+						Type:      event.RegSetValueInternal,
+						Timestamp: time.Now(),
+						Params: event.Params{
+							params.RegPath:      {Name: params.RegPath, Type: params.Key, Value: `\SessionId`},
+							params.RegData:      {Name: params.RegData, Type: params.UnicodeString, Value: "{ABD9EA10-87F6-11EB-9ED5-645D86501328}"},
+							params.RegValueType: {Name: params.RegValueType, Type: params.Enum, Value: uint32(1), Enum: key.RegistryValueTypes},
+							params.RegKeyHandle: {Name: params.RegKeyHandle, Type: params.Uint64, Value: uint64(0)}},
+					},
+					{
+						Type:      event.RegSetValueInternal,
+						Timestamp: time.Now(),
+						Params: event.Params{
+							params.RegPath:      {Name: params.RegPath, Type: params.Key, Value: `\Directory`},
+							params.RegData:      {Name: params.RegData, Type: params.UnicodeString, Value: "%SYSTEMROOT%"},
+							params.RegValueType: {Name: params.RegValueType, Type: params.Enum, Value: uint32(2), Enum: key.RegistryValueTypes},
+							params.RegKeyHandle: {Name: params.RegKeyHandle, Type: params.Uint64, Value: uint64(0)}},
+					},
+				}
+			},
+			func() *handle.SnapshotterMock {
+				hsnap := new(handle.SnapshotterMock)
+				return hsnap
+			},
+			func(e *event.Event, t *testing.T, hsnap *handle.SnapshotterMock, p Processor) {
+				assert.Equal(t, `HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Windows\Directory`, e.GetParamAsString(params.RegPath))
+				assert.Equal(t, `REG_EXPAND_SZ`, e.GetParamAsString(params.RegValueType))
+				assert.Equal(t, `%SYSTEMROOT%`, e.GetParamAsString(params.RegData))
+				assert.Equal(t, p.(*registryProcessor).valuesSize(23234), 1)
+				time.Sleep(time.Millisecond * 500)
+				assert.Equal(t, p.(*registryProcessor).valuesSize(23234), 0)
 			},
 		},
 	}