x-pack/libbeat/reader/etw: New reader to collect ETW logs

.github/CODEOWNERS

@@ -200,3 +200,4 @@ CHANGELOG*
 /x-pack/packetbeat/ @elastic/security-external-integrations
 /x-pack/winlogbeat/ @elastic/security-external-integrations
 /x-pack/libbeat/reader/parquet/ @elastic/security-external-integrations
+/x-pack/libbeat/reader/etw/ @elastic/security-external-integrations

CHANGELOG.next.asciidoc

@@ -230,6 +230,8 @@ Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will d
 
 *Libbeat*
 
+- Added support for ETW reader. {pull}36914[36914]
+
 *Heartbeat*
 - Added status to monitor run log report.
 - Capture and log the individual connection metrics for all the lightweight monitors

x-pack/libbeat/Jenkinsfile.yml

@@ -27,6 +27,41 @@ stages:
             branches: true     ## for all the branches
             tags: true         ## for all the tags
         stage: extended
+    windows-2022:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2022"
+        stage: mandatory
+    windows-2019:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2019"
+        stage: extended_win
+    windows-2016:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2016"
+        stage: mandatory
+    windows-2012:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-2012-r2"
+        stage: extended_win
+    windows-11:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-11"
+        stage: extended_win
+    windows-10:
+        mage: "mage build unitTest"
+        platforms:             ## override default labels in this specific stage.
+            - "windows-10"
+        stage: extended_win
+    windows-8:
+        mage: "mage build unitTest"
+        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,18 @@
+// 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
+
+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/consumer.go

@@ -0,0 +1,60 @@
+// 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"
+)
+
+// StartConsumer initializes and starts the ETW event tracing session.
+func (s *Session) StartConsumer() error {
+	var elf EventTraceLogfile
+	var err error
+
+	// Configure EventTraceLogfile based on the session type (realtime or not).
+	if !s.Realtime {
+		elf.LogFileMode = PROCESS_TRACE_MODE_EVENT_RECORD
+		logfilePtr, err := syscall.UTF16PtrFromString(s.Name)
+		if err != nil {
+			return fmt.Errorf("failed to convert logfile name")
+		}
+		elf.LogFileName = logfilePtr
+	} else {
+		elf.LogFileMode = PROCESS_TRACE_MODE_EVENT_RECORD | PROCESS_TRACE_MODE_REAL_TIME
+		sessionPtr, err := syscall.UTF16PtrFromString(s.Name)
+		if err != nil {
+			return fmt.Errorf("failed to convert session name")
+		}
+		elf.LoggerName = sessionPtr
+	}
+
+	// Set callbacks and context for the session.
+	elf.BufferCallback = s.BufferCallback
+	elf.Callback = s.Callback
+	elf.Context = 0
+
+	// Open an ETW trace processing handle for consuming events
+	// from an ETW real-time trace session or an ETW log file.
+	s.TraceHandler, err = OpenTraceFunc(&elf)
+	if err != nil {
+		// Handle specific errors for trace opening.
+		if errors.Is(err, ERROR_BAD_PATHNAME) {
+			return fmt.Errorf("invalid log source when opening trace: %w", err)
+		} else if errors.Is(err, ERROR_ACCESS_DENIED) {
+			return fmt.Errorf("access denied when opening trace: %w", err)
+		}
+		return fmt.Errorf("failed to open trace: %w", err)
+	}
+	// Process the trace. This function blocks until processing ends.
+	if err := ProcessTraceFunc(&s.TraceHandler, 1, nil, nil); err != nil {
+		return fmt.Errorf("failed to process trace: %w", err)
+	}
+
+	return nil
+}

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

@@ -0,0 +1,107 @@
+// 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"
+)
+
+func TestStartConsumer_OpenTraceError(t *testing.T) {
+	// Backup original functions and defer restoration
+	originalOpenTraceFunc := OpenTraceFunc
+	defer func() {
+		OpenTraceFunc = originalOpenTraceFunc
+	}()
+
+	// Mock implementations
+	OpenTraceFunc = func(elf *EventTraceLogfile) (uint64, error) {
+		return 0, ERROR_ACCESS_DENIED // Mock a valid session handler
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:           "TestSession",
+		Realtime:       false,
+		BufferCallback: uintptr(0),
+		Callback:       uintptr(0),
+	}
+
+	// Test StartConsumer
+	err := session.StartConsumer()
+
+	// Assertions
+	assert.EqualError(t, err, "access denied when opening trace: Access is denied.")
+}
+
+func TestStartConsumer_ProcessTraceError(t *testing.T) {
+	// Backup original functions and defer restoration
+	originalOpenTraceFunc := OpenTraceFunc
+	originalProcessTraceFunc := ProcessTraceFunc
+	defer func() {
+		OpenTraceFunc = originalOpenTraceFunc
+		ProcessTraceFunc = originalProcessTraceFunc
+	}()
+
+	// Mock implementations
+	OpenTraceFunc = func(elf *EventTraceLogfile) (uint64, error) {
+		return 12345, nil // Mock a valid session handler
+	}
+
+	ProcessTraceFunc = func(handleArray *uint64, handleCount uint32, startTime *FileTime, endTime *FileTime) error {
+		return ERROR_INVALID_PARAMETER
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:           "TestSession",
+		Realtime:       true,
+		BufferCallback: uintptr(0),
+		Callback:       uintptr(0),
+	}
+
+	// Test StartConsumer
+	err := session.StartConsumer()
+
+	// Assertions
+	assert.EqualError(t, err, "failed to process trace: The parameter is incorrect.")
+}
+
+func TestStartConsumer_Success(t *testing.T) {
+	// Backup original functions and defer restoration
+	originalOpenTraceFunc := OpenTraceFunc
+	originalProcessTraceFunc := ProcessTraceFunc
+	defer func() {
+		OpenTraceFunc = originalOpenTraceFunc
+		ProcessTraceFunc = originalProcessTraceFunc
+	}()
+
+	// Mock implementations
+	OpenTraceFunc = func(elf *EventTraceLogfile) (uint64, error) {
+		return 12345, nil // Mock a valid session handler
+	}
+
+	ProcessTraceFunc = func(handleArray *uint64, handleCount uint32, startTime *FileTime, endTime *FileTime) error {
+		return nil
+	}
+
+	// Create a Session instance
+	session := &Session{
+		Name:           "TestSession",
+		Realtime:       true,
+		BufferCallback: uintptr(0),
+		Callback:       uintptr(0),
+	}
+
+	// Test StartConsumer
+	err := session.StartConsumer()
+
+	// Assertions
+	assert.NoError(t, err)
+	assert.Equal(t, uint64(12345), session.TraceHandler, "TraceHandler should be set to the mock value")
+}

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

@@ -0,0 +1,132 @@
+// 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"
+	"unsafe"
+)
+
+func isValidHandler(handler uint64) bool {
+	if handler == 0 || handler == INVALID_PROCESSTRACE_HANDLE {
+		return false
+	}
+	return true
+}
+
+// GetHandler queries the status of an existing ETW session to get its handler and properties.
+func (s *Session) GetHandler() 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")
+	}
+
+	// Query the current state of the ETW session.
+	if err = ControlTraceFunc(
+		0,
+		sessionNamePtr,
+		s.Properties,
+		EVENT_TRACE_CONTROL_QUERY,
+	); err != nil {
+		// Handle specific errors related to the query operation.
+		if errors.Is(err, ERROR_BAD_LENGTH) {
+			return fmt.Errorf("bad length when querying handler: %w", err)
+		} else if errors.Is(err, ERROR_INVALID_PARAMETER) {
+			return fmt.Errorf("invalid parameters when querying handler: %w", err)
+		} else if errors.Is(err, ERROR_WMI_INSTANCE_NOT_FOUND) {
+			return fmt.Errorf("session is not running")
+		}
+		return fmt.Errorf("failed to get handler: %w", err)
+	}
+
+	// Get the session handler from the properties struct.
+	s.Handler = uintptr(s.Properties.Wnode.Union1)
+
+	return nil
+}
+
+// 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")
+	}
+
+	// Start the ETW trace session.
+	err = StartTraceFunc(
+		&s.Handler,
+		sessionPtr,
+		s.Properties,
+	)
+	if err != nil {
+		// Handle specific errors related to starting the trace session.
+		if errors.Is(err, ERROR_ALREADY_EXISTS) {
+			return fmt.Errorf("session already exists: %w", err)
+		} else if errors.Is(err, ERROR_INVALID_PARAMETER) {
+			return fmt.Errorf("invalid parameters when starting session trace: %w", err)
+		}
+		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
+	}
+
+	// Enable the trace session with extended options.
+	if err := EnableTraceFunc(
+		s.Handler,
+		(*GUID)(unsafe.Pointer(&s.GUID)),
+		EVENT_CONTROL_CODE_ENABLE_PROVIDER,
+		s.TraceLevel,
+		s.MatchAnyKeyword,
+		s.MatchAllKeyword,
+		0,       // Asynchronous enablement with zero timeout
+		&params, // Additional parameters
+	); err != nil {
+		// Handle specific errors related to enabling the trace session.
+		if errors.Is(err, ERROR_INVALID_PARAMETER) {
+			return fmt.Errorf("invalid parameters when enabling session trace: %w", err)
+		} else if errors.Is(err, ERROR_TIMEOUT) {
+			return fmt.Errorf("timeout value expired before the enable callback completed: %w", err)
+		} else if errors.Is(err, ERROR_NO_SYSTEM_RESOURCES) {
+			return fmt.Errorf("exceeded the number of trace sessions that can enable the provider: %w", err)
+		}
+		return fmt.Errorf("failed to enable trace: %w", err)
+	}
+
+	return nil
+}
+
+// StopSession closes the ETW session and associated handles if they were created.
+func (s *Session) StopSession() error {
+	if s.Realtime {
+		if isValidHandler(s.TraceHandler) {
+			// Attempt to close the trace and handle potential errors.
+			if err := CloseTraceFunc(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 ControlTraceFunc(
+				s.Handler,
+				nil,
+				s.Properties,
+				EVENT_TRACE_CONTROL_STOP,
+			)
+		}
+	}
+
+	return nil
+}