x-pack/libbeat/reader/etw: New reader to collect ETW logs (#36914)

Add support for collecting Microsoft ETW events in Libbeat.

.github/CODEOWNERS

@@ -173,6 +173,8 @@ CHANGELOG*
 /x-pack/filebeat/modules.d/zoom.yml.disabled @elastic/security-service-integrations
 /x-pack/filebeat/processors/decode_cef/ @elastic/sec-deployment-and-devices
 /x-pack/heartbeat/ @elastic/obs-ds-hosted-services
+/x-pack/libbeat/reader/parquet/ @elastic/security-service-integrations
+/x-pack/libbeat/reader/etw/ @elastic/sec-windows-platform
 /x-pack/metricbeat/ @elastic/elastic-agent-data-plane
 /x-pack/metricbeat/docs/ # Listed without an owner to avoid maintaining doc ownership for each input and module.
 /x-pack/metricbeat/module/activemq @elastic/obs-infraobs-integrations
@@ -219,4 +221,3 @@ CHANGELOG*
 /x-pack/osquerybeat/ @elastic/sec-deployment-and-devices
 /x-pack/packetbeat/ @elastic/sec-linux-platform
 /x-pack/winlogbeat/ @elastic/sec-windows-platform
-/x-pack/libbeat/reader/parquet/ @elastic/security-service-integrations

CHANGELOG.next.asciidoc

@@ -181,6 +181,8 @@ Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will d
 *Libbeat*
 - Add watcher that can be used to monitor Linux kernel events. {pull}37833[37833]
 
+- Added support for ETW reader. {pull}36914[36914]
+
 *Heartbeat*
 - Added status to monitor run log report.
 - Upgrade github.com/elastic/go-elasticsearch/v8 to v8.12.0. {pull}37673[37673]

x-pack/libbeat/Jenkinsfile.yml

@@ -27,6 +27,43 @@ stages:
             branches: true     ## for all the branches
             tags: true         ## for all the tags
         stage: extended
+    ## For now Windows CI tests for Libbeat are only enabled for ETW
+    ## It only contains Go tests
+    windows-2022:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2022"
+        stage: mandatory
+    windows-2019:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2019"
+        stage: extended_win
+    windows-2016:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2016"
+        stage: mandatory
+    windows-2012:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2012-r2"
+        stage: extended_win
+    windows-11:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-11"
+        stage: extended_win
+    windows-10:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-10"
+        stage: extended_win
+    windows-8:
+        mage: "mage -w reader/etw build goUnitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-8"
+        stage: extended_win
     unitTest:
         mage: "mage build unitTest"
         stage: mandatory

x-pack/libbeat/reader/etw/config.go

@@ -0,0 +1,16 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package etw
+
+type Config struct {
+	Logfile         string // Path to the logfile
+	ProviderGUID    string // GUID of the ETW provider
+	ProviderName    string // Name of the ETW provider
+	SessionName     string // Name for new ETW session
+	TraceLevel      string // Level of tracing (e.g., "verbose")
+	MatchAnyKeyword uint64 // Filter for any matching keywords (bitmask)
+	MatchAllKeyword uint64 // Filter for all matching keywords (bitmask)
+	Session         string // Existing session to attach
+}

x-pack/libbeat/reader/etw/controller.go

@@ -0,0 +1,121 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+//go:build windows
+
+package etw
+
+import (
+	"errors"
+	"fmt"
+	"syscall"
+)
+
+// AttachToExistingSession queries the status of an existing ETW session.
+// On success, it updates the Session's handler with the queried information.
+func (s *Session) AttachToExistingSession() error {
+	// Convert the session name to UTF16 for Windows API compatibility.
+	sessionNamePtr, err := syscall.UTF16PtrFromString(s.Name)
+	if err != nil {
+		return fmt.Errorf("failed to convert session name: %w", err)
+	}
+
+	// Query the current state of the ETW session.
+	err = s.controlTrace(0, sessionNamePtr, s.properties, EVENT_TRACE_CONTROL_QUERY)
+	switch {
+	case err == nil:
+		// Get the session handler from the properties struct.
+		s.handler = uintptr(s.properties.Wnode.Union1)
+
+		return nil
+
+	// Handle specific errors related to the query operation.
+	case errors.Is(err, ERROR_BAD_LENGTH):
+		return fmt.Errorf("bad length when querying handler: %w", err)
+	case errors.Is(err, ERROR_INVALID_PARAMETER):
+		return fmt.Errorf("invalid parameters when querying handler: %w", err)
+	case errors.Is(err, ERROR_WMI_INSTANCE_NOT_FOUND):
+		return fmt.Errorf("session is not running: %w", err)
+	default:
+		return fmt.Errorf("failed to get handler: %w", err)
+	}
+}
+
+// CreateRealtimeSession initializes and starts a new real-time ETW session.
+func (s *Session) CreateRealtimeSession() error {
+	// Convert the session name to UTF16 format for Windows API compatibility.
+	sessionPtr, err := syscall.UTF16PtrFromString(s.Name)
+	if err != nil {
+		return fmt.Errorf("failed to convert session name: %w", err)
+	}
+
+	// Start the ETW trace session.
+	err = s.startTrace(&s.handler, sessionPtr, s.properties)
+	switch {
+	case err == nil:
+
+	// Handle specific errors related to starting the trace session.
+	case errors.Is(err, ERROR_ALREADY_EXISTS):
+		return fmt.Errorf("session already exists: %w", err)
+	case errors.Is(err, ERROR_INVALID_PARAMETER):
+		return fmt.Errorf("invalid parameters when starting session trace: %w", err)
+	default:
+		return fmt.Errorf("failed to start trace: %w", err)
+	}
+
+	// Set additional parameters for trace enabling.
+	// See https://learn.microsoft.com/en-us/windows/win32/api/evntrace/ns-evntrace-enable_trace_parameters#members
+	params := EnableTraceParameters{
+		Version: 2, // ENABLE_TRACE_PARAMETERS_VERSION_2
+	}
+
+	// Zero timeout means asynchronous enablement
+	const timeout = 0
+
+	// Enable the trace session with extended options.
+	err = s.enableTrace(s.handler, &s.GUID, EVENT_CONTROL_CODE_ENABLE_PROVIDER, s.traceLevel, s.matchAnyKeyword, s.matchAllKeyword, timeout, &params)
+	switch {
+	case err == nil:
+		return nil
+	// Handle specific errors related to enabling the trace session.
+	case errors.Is(err, ERROR_INVALID_PARAMETER):
+		return fmt.Errorf("invalid parameters when enabling session trace: %w", err)
+	case errors.Is(err, ERROR_TIMEOUT):
+		return fmt.Errorf("timeout value expired before the enable callback completed: %w", err)
+	case errors.Is(err, ERROR_NO_SYSTEM_RESOURCES):
+		return fmt.Errorf("exceeded the number of trace sessions that can enable the provider: %w", err)
+	default:
+		return fmt.Errorf("failed to enable trace: %w", err)
+	}
+}
+
+// StopSession closes the ETW session and associated handles if they were created.
+func (s *Session) StopSession() error {
+	if !s.Realtime {
+		return nil
+	}
+
+	if isValidHandler(s.traceHandler) {
+		// Attempt to close the trace and handle potential errors.
+		if err := s.closeTrace(s.traceHandler); err != nil && !errors.Is(err, ERROR_CTX_CLOSE_PENDING) {
+			return fmt.Errorf("failed to close trace: %w", err)
+		}
+	}
+
+	if s.NewSession {
+		// If we created the session, send a control command to stop it.
+		return s.controlTrace(
+			s.handler,
+			nil,
+			s.properties,
+			EVENT_TRACE_CONTROL_STOP,
+		)
+	}
+
+	return nil
+}
+
+func isValidHandler(handler uint64) bool {
+	return handler != 0 && handler != INVALID_PROCESSTRACE_HANDLE
+}

x-pack/libbeat/reader/etw/controller_test.go

@@ -0,0 +1,190 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+//go:build windows
+
+package etw
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/sys/windows"
+)
+
+func TestAttachToExistingSession_Error(t *testing.T) {
+	// Mock implementation of controlTrace
+	controlTrace := func(traceHandle uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties,
+		controlCode uint32) error {
+		return ERROR_WMI_INSTANCE_NOT_FOUND
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:         "TestSession",
+		properties:   &EventTraceProperties{},
+		controlTrace: controlTrace,
+	}
+
+	err := session.AttachToExistingSession()
+	assert.EqualError(t, err, "session is not running: The instance name passed was not recognized as valid by a WMI data provider.")
+}
+
+func TestAttachToExistingSession_Success(t *testing.T) {
+	// Mock implementation of controlTrace
+	controlTrace := func(traceHandle uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties,
+		controlCode uint32) error {
+		// Set a mock handler value
+		properties.Wnode.Union1 = 12345
+		return nil
+	}
+
+	// Create a Session instance with initialized Properties
+	session := &Session{
+		Name:         "TestSession",
+		properties:   &EventTraceProperties{},
+		controlTrace: controlTrace,
+	}
+
+	err := session.AttachToExistingSession()
+
+	assert.NoError(t, err)
+	assert.Equal(t, uintptr(12345), session.handler, "Handler should be set to the mock value")
+}
+
+func TestCreateRealtimeSession_StartTraceError(t *testing.T) {
+	// Mock implementation of startTrace
+	startTrace := func(traceHandle *uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties) error {
+		return ERROR_ALREADY_EXISTS
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:       "TestSession",
+		properties: &EventTraceProperties{},
+		startTrace: startTrace,
+	}
+
+	err := session.CreateRealtimeSession()
+	assert.EqualError(t, err, "session already exists: Cannot create a file when that file already exists.")
+}
+
+func TestCreateRealtimeSession_EnableTraceError(t *testing.T) {
+	// Mock implementations
+	startTrace := func(traceHandle *uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties) error {
+		*traceHandle = 12345 // Mock handler value
+		return nil
+	}
+
+	enableTrace := func(traceHandle uintptr,
+		providerId *windows.GUID,
+		isEnabled uint32,
+		level uint8,
+		matchAnyKeyword uint64,
+		matchAllKeyword uint64,
+		enableProperty uint32,
+		enableParameters *EnableTraceParameters) error {
+		return ERROR_INVALID_PARAMETER
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:        "TestSession",
+		properties:  &EventTraceProperties{},
+		startTrace:  startTrace,
+		enableTrace: enableTrace,
+	}
+
+	err := session.CreateRealtimeSession()
+	assert.EqualError(t, err, "invalid parameters when enabling session trace: The parameter is incorrect.")
+}
+
+func TestCreateRealtimeSession_Success(t *testing.T) {
+	// Mock implementations
+	startTrace := func(traceHandle *uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties) error {
+		*traceHandle = 12345 // Mock handler value
+		return nil
+	}
+
+	enableTrace := func(traceHandle uintptr,
+		providerId *windows.GUID,
+		isEnabled uint32,
+		level uint8,
+		matchAnyKeyword uint64,
+		matchAllKeyword uint64,
+		enableProperty uint32,
+		enableParameters *EnableTraceParameters) error {
+		return nil
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:        "TestSession",
+		properties:  &EventTraceProperties{},
+		startTrace:  startTrace,
+		enableTrace: enableTrace,
+	}
+
+	err := session.CreateRealtimeSession()
+
+	assert.NoError(t, err)
+	assert.Equal(t, uintptr(12345), session.handler, "Handler should be set to the mock value")
+}
+
+func TestStopSession_Error(t *testing.T) {
+	// Mock implementation of closeTrace
+	closeTrace := func(traceHandle uint64) error {
+		return ERROR_INVALID_PARAMETER
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Realtime:     true,
+		NewSession:   true,
+		traceHandler: 12345, // Example handler value
+		properties:   &EventTraceProperties{},
+		closeTrace:   closeTrace,
+	}
+
+	err := session.StopSession()
+	assert.EqualError(t, err, "failed to close trace: The parameter is incorrect.")
+}
+
+func TestStopSession_Success(t *testing.T) {
+	// Mock implementations
+	closeTrace := func(traceHandle uint64) error {
+		return nil
+	}
+
+	controlTrace := func(traceHandle uintptr,
+		instanceName *uint16,
+		properties *EventTraceProperties,
+		controlCode uint32) error {
+		// Set a mock handler value
+		return nil
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Realtime:     true,
+		NewSession:   true,
+		traceHandler: 12345, // Example handler value
+		properties:   &EventTraceProperties{},
+		closeTrace:   closeTrace,
+		controlTrace: controlTrace,
+	}
+
+	err := session.StopSession()
+	assert.NoError(t, err)
+}