Subscribe arm agent to new windows channels

Author: Kbayero

.github/workflows/v11-principal-alpha-deployment.yml

@@ -4,10 +4,10 @@ on:
   workflow_dispatch:
     inputs:
       version_tag:
-        description: "Version to deploy."
+        description: "Version to deploy.(e.g., v1.0.0-alpha.1)"
         required: true
       event_processor_tag:
-        description: "Event processor version to use for this deployment."
+        description: "Event processor version to use for this deployment.(e.g., 1.0.0-beta)"
         required: true
 
 jobs:

.github/workflows/v11-used-build.yml

@@ -63,8 +63,6 @@ jobs:
           $env:GOOS = "linux"
           $env:GOARCH = "amd64"
           go build -o utmstack_agent_service -v .
-          # $env:GOARCH = "arm64"
-          # go build -o utmstack_agent_service_arm64 -v .
       
           $env:GOOS = "windows"
           $env:GOARCH = "amd64"

agent/collectors/windows_amd64.go

@@ -73,7 +73,6 @@ func (w Windows) Install() error {
 }
 
 func (w Windows) SendLogs() {
-	host, _ := os.Hostname()
 	logLinesChan := make(chan string)
 	path := utils.GetMyPath()
 	winbLogPath := filepath.Join(path, "beats", "winlogbeat", "logs")

agent/collectors/windows_arm64.go

@@ -12,7 +12,9 @@ import (
 	"os/signal"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
+	"time"
 	"unsafe"
 
 	"github.com/threatwinds/go-sdk/plugins"
@@ -75,8 +77,10 @@ type EventSubscription struct {
 	Channel      string
 	Query        string
 	Errors       chan error
-	Callback     func(event *Event)
 	winAPIHandle windows.Handle
+
+	mu      sync.Mutex
+	running bool
 }
 
 const (
@@ -91,9 +95,13 @@ var (
 	procEvtSubscribe = modwevtapi.NewProc("EvtSubscribe")
 	procEvtRender    = modwevtapi.NewProc("EvtRender")
 	procEvtClose     = modwevtapi.NewProc("EvtClose")
+	incomingEvents   = make(chan string, 1024)
 )
 
 func (evtSub *EventSubscription) Create() error {
+	evtSub.mu.Lock()
+	defer evtSub.mu.Unlock()
+
 	if evtSub.winAPIHandle != 0 {
 		return fmt.Errorf("windows_events: subscription has already been created")
 	}
@@ -132,6 +140,9 @@ func (evtSub *EventSubscription) Create() error {
 }
 
 func (evtSub *EventSubscription) Close() error {
+	evtSub.mu.Lock()
+	defer evtSub.mu.Unlock()
+
 	if evtSub.winAPIHandle == 0 {
 		return fmt.Errorf("windows_events: no active subscription to close")
 	}
@@ -146,47 +157,70 @@ func (evtSub *EventSubscription) Close() error {
 func (evtSub *EventSubscription) winAPICallback(action, userContext, event uintptr) uintptr {
 	switch action {
 	case evtSubscribeActionError:
-		evtSub.Errors <- fmt.Errorf("windows_events: error in callback, code: %x", uint16(event))
-	case evtSubscribeActionDeliver:
-		bufferSize := uint32(4096)
-		for {
-			renderSpace := make([]uint16, bufferSize/2)
-			bufferUsed := uint32(0)
-			propertyCount := uint32(0)
-			ret, _, err := procEvtRender.Call(
-				0,
-				event,
-				evtRenderEventXML,
-				uintptr(bufferSize),
-				uintptr(unsafe.Pointer(&renderSpace[0])),
-				uintptr(unsafe.Pointer(&bufferUsed)),
-				uintptr(unsafe.Pointer(&propertyCount)),
-			)
-			if ret == 0 {
-				if err == windows.ERROR_INSUFFICIENT_BUFFER {
-					bufferSize *= 2
-					continue
+		err := fmt.Errorf("windows_events: error in callback, code: %x", uint16(event))
+		evtSub.Errors <- err
+
+		go func(channel string) {
+			utils.Logger.LogF(100, "Attempting to resubscribe to channel: %s after error: %v", channel, err)
+			evtSub.mu.Lock()
+			defer evtSub.mu.Unlock()
+
+			_ = evtSub.Close()
+
+			for {
+				time.Sleep(5 * time.Second)
+				if err := evtSub.Create(); err != nil {
+					utils.Logger.ErrorF("Retry failed for channel %s: %s", channel, err)
+				} else {
+					utils.Logger.LogF(100, "Resubscribed to channel: %s", channel)
+					break
 				}
-				evtSub.Errors <- fmt.Errorf("windows_events: failed to render event: %w", err)
-				return 0
-			}
-			xmlStr := windows.UTF16ToString(renderSpace)
-			xmlStr = cleanXML(xmlStr)
-
-			dataParsed := new(Event)
-			if err := xml.Unmarshal([]byte(xmlStr), dataParsed); err != nil {
-				evtSub.Errors <- fmt.Errorf("windows_events: failed to parse XML: %s", err)
-			} else {
-				evtSub.Callback(dataParsed)
 			}
+		}(evtSub.Channel)
+
+	case evtSubscribeActionDeliver:
+		utils.Logger.LogF(100, "Received event from channel: %s", evtSub.Channel)
+		xmlStr, err := quickRenderXML(event)
+		if err != nil {
+			evtSub.Errors <- fmt.Errorf("render in callback: %v", err)
 			break
 		}
+		select {
+		case incomingEvents <- xmlStr:
+		default:
+			utils.Logger.ErrorF("incomingEvents lleno: evento descartado")
+		}
 	default:
 		evtSub.Errors <- fmt.Errorf("windows_events: unsupported action in callback: %x", uint16(action))
 	}
 	return 0
 }
 
+func quickRenderXML(h uintptr) (string, error) {
+	bufSize := uint32(4096)
+	for {
+		space := make([]uint16, bufSize/2)
+		used := uint32(0)
+		prop := uint32(0)
+
+		ret, _, err := procEvtRender.Call(
+			0, h, evtRenderEventXML,
+			uintptr(bufSize),
+			uintptr(unsafe.Pointer(&space[0])),
+			uintptr(unsafe.Pointer(&used)),
+			uintptr(unsafe.Pointer(&prop)),
+		)
+		if ret == 0 {
+			if err == windows.ERROR_INSUFFICIENT_BUFFER {
+				bufSize *= 2
+				continue
+			}
+			return "", err
+		}
+		return cleanXML(windows.UTF16ToString(space)), nil
+	}
+}
+
 func cleanXML(xmlStr string) string {
 	xmlStr = strings.TrimSpace(xmlStr)
 	if idx := strings.Index(xmlStr, "<?xml"); idx > 0 {
@@ -211,39 +245,21 @@ func getCollectorsInstances() []Collector {
 
 func (w Windows) SendLogs() {
 	errorsChan := make(chan error, 10)
+	go eventWorker()
 
-	callback := func(event *Event) {
-		eventJSON, err := convertEventToJSON(event)
-		if err != nil {
-			utils.Logger.ErrorF("error converting event to JSON: %v", err)
-			return
-		}
-		host, err := os.Hostname()
-		if err != nil {
-			utils.Logger.ErrorF("error getting hostname: %v", err)
-			host = "unknown"
-		}
-		validatedLog, _, err := validations.ValidateString(eventJSON, false)
-		if err != nil {
-			utils.Logger.LogF(100, "error validating log: %s: %v", eventJSON, err)
-			return
-		}
-		logservice.LogQueue <- &plugins.Log{
-			DataType:   string(config.DataTypeWindowsAgent),
-			DataSource: host,
-			Raw:        validatedLog,
-		}
+	channels := []string{
+		"Security", "Application", "System", "Windows Powershell", "Microsoft-Windows-Powershell/Operational", "ForwardedEvents",
+		"Microsoft-Windows-WinLogon/Operational", "Microsoft-Windows-Windows Firewall With Advanced Security/Firewall",
+		"Microsoft-Windows-Windows Defender/Operational",
 	}
 
-	channels := []string{"Security", "Application", "System"}
 	var subscriptions []*EventSubscription
 
 	for _, channel := range channels {
 		sub := &EventSubscription{
-			Channel:  channel,
-			Query:    "*",
-			Errors:   errorsChan,
-			Callback: callback,
+			Channel: channel,
+			Query:   "*",
+			Errors:  errorsChan,
 		}
 		if err := sub.Create(); err != nil {
 			utils.Logger.ErrorF("Error subscribing to channel %s: %s", channel, err)
@@ -271,6 +287,44 @@ func (w Windows) SendLogs() {
 	utils.Logger.LogF(100, "Agent finished successfully.")
 }
 
+func eventWorker() {
+	host, err := os.Hostname()
+	if err != nil {
+		utils.Logger.ErrorF("error getting hostname: %v", err)
+		host = "unknown"
+	}
+
+	for xmlStr := range incomingEvents {
+		ev := new(Event)
+		if err := xml.Unmarshal([]byte(xmlStr), ev); err != nil {
+			utils.Logger.ErrorF("unmarshal error: %v", err)
+			continue
+		}
+
+		eventJSON, err := convertEventToJSON(ev)
+		if err != nil {
+			utils.Logger.ErrorF("toJSON error: %v", err)
+			continue
+		}
+
+		validatedLog, _, err := validations.ValidateString(eventJSON, false)
+		if err != nil {
+			utils.Logger.LogF(100, "validation error: %s: %v", eventJSON, err)
+			continue
+		}
+
+		select {
+		case logservice.LogQueue <- &plugins.Log{
+			DataSource: host,
+			DataType:   string(config.DataTypeWindowsAgent),
+			Raw:        validatedLog,
+		}:
+		default:
+			utils.Logger.LogF(100, "LogQueue full: event discarded")
+		}
+	}
+}
+
 func convertEventToJSON(event *Event) (string, error) {
 	eventMap := map[string]interface{}{
 		"timestamp":     event.System.TimeCreated.SystemTime,