feat(agent): add native auditd collector for Linux

- Implement native auditd collector using go-libaudit v2 with netlink multicast
- Add enterprise-ready auditd configuration (50-utmstack.rules)
- Respect existing customer audit rules (additive approach)
- Add cleanup on agent uninstall (removes UTMStack rules only)
- Support automatic auditd installation on Debian/Ubuntu/RHEL/Fedora
- Handle migration path for existing auditd installations
- Add distro detection for package manager selection
- Remove legacy beats/filebeat commented code

Author: yllada

agent/cmd/uninstall.go

@@ -47,6 +47,15 @@ var uninstallCmd = &cobra.Command{
 		if err = pb.DeleteAgent(cnf); err != nil {
 			utils.Logger.ErrorF("error deleting agent: %v", err)
 		}
+
+		// Uninstall dependencies (cleanup auditd rules, etc.)
+		fmt.Print("Cleaning up dependencies... ")
+		if err = dependency.UninstallAll(); err != nil {
+			fmt.Printf("Warning: %v\n", err)
+		} else {
+			fmt.Println("[OK]")
+		}
+
 		if err = collector.UninstallAll(); err != nil {
 			fmt.Printf("error uninstalling collectors: %v\n", err)
 			os.Exit(1)

agent/collector/auditd/auditd.go

@@ -0,0 +1,23 @@
+// Package auditd provides a native collector for Linux Audit Framework events.
+// It uses go-libaudit to receive events via netlink multicast and reassembles
+// them before sending to the log queue.
+package auditd
+
+import "time"
+
+const (
+	// auditdRestartDelay is the initial delay between reconnection attempts
+	auditdRestartDelay = 5 * time.Second
+
+	// auditdMaxRestartDelay is the maximum backoff delay for reconnection
+	auditdMaxRestartDelay = 5 * time.Minute
+
+	// reassemblerMaxInFlight is the maximum number of events held for reassembly
+	reassemblerMaxInFlight = 50
+
+	// reassemblerTimeout is how long to wait for related messages before flushing
+	reassemblerTimeout = 2 * time.Second
+
+	// maintainInterval is how often to run Reassembler.Maintain() to flush stale events
+	maintainInterval = 500 * time.Millisecond
+)

agent/collector/auditd/auditd_linux.go

@@ -0,0 +1,193 @@
+//go:build linux
+// +build linux
+
+package auditd
+
+import (
+	"context"
+	"os"
+	"sync"
+	"time"
+
+	libaudit "github.com/elastic/go-libaudit/v2"
+	"github.com/elastic/go-libaudit/v2/auparse"
+	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/utmstack/UTMStack/agent/utils"
+)
+
+// AuditdCollector collects Linux Audit events via netlink multicast
+type AuditdCollector struct {
+	client      auditReceiver
+	reassembler *libaudit.Reassembler
+	cancel      context.CancelFunc
+	mu          sync.Mutex
+}
+
+// New creates a new AuditdCollector
+func New() *AuditdCollector {
+	return &AuditdCollector{}
+}
+
+// Name returns the collector name
+func (a *AuditdCollector) Name() string {
+	return "auditd"
+}
+
+// Start begins collecting audit events and sending them to the queue
+func (a *AuditdCollector) Start(ctx context.Context, queue chan *plugins.Log) {
+	// Preflight check for audit capability
+	if err := checkAuditCapability(); err != nil {
+		utils.Logger.ErrorF("auditd: preflight check failed: %v", err)
+		return
+	}
+
+	host, err := os.Hostname()
+	if err != nil {
+		utils.Logger.ErrorF("auditd: error getting hostname: %v", err)
+		host = "unknown"
+	}
+
+	restartDelay := auditdRestartDelay
+
+	for {
+		select {
+		case <-ctx.Done():
+			utils.Logger.Info("auditd collector stopping due to context cancellation")
+			return
+		default:
+		}
+
+		exitCode := a.runAuditClient(ctx, host, queue)
+
+		if exitCode == 0 {
+			utils.Logger.Info("auditd client exited normally")
+		} else {
+			utils.Logger.ErrorF("auditd client exited with code %d, restarting in %v", exitCode, restartDelay)
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(restartDelay):
+		}
+
+		// Exponential backoff
+		restartDelay *= 2
+		if restartDelay > auditdMaxRestartDelay {
+			restartDelay = auditdMaxRestartDelay
+		}
+	}
+}
+
+// runAuditClient creates the audit client and runs the receive loop
+func (a *AuditdCollector) runAuditClient(ctx context.Context, host string, queue chan *plugins.Log) int {
+	a.mu.Lock()
+	clientCtx, cancel := context.WithCancel(ctx)
+	a.cancel = cancel
+
+	// Create multicast audit client
+	client, err := newAuditClient()
+	if err != nil {
+		a.mu.Unlock()
+		utils.Logger.ErrorF("auditd: error creating audit client: %v", err)
+		return -1
+	}
+	a.client = client
+
+	// Create event stream for reassembled events
+	stream := newEventStream(queue, host)
+
+	// Create reassembler
+	reassembler, err := libaudit.NewReassembler(reassemblerMaxInFlight, reassemblerTimeout, stream)
+	if err != nil {
+		client.Close()
+		a.mu.Unlock()
+		utils.Logger.ErrorF("auditd: error creating reassembler: %v", err)
+		return -1
+	}
+	a.reassembler = reassembler
+	a.mu.Unlock()
+
+	utils.Logger.Info("auditd collector started (netlink multicast)")
+
+	// Start maintenance goroutine for reassembler
+	go a.runMaintenance(clientCtx)
+
+	// Main receive loop
+	for {
+		select {
+		case <-clientCtx.Done():
+			a.cleanup()
+			return 0
+		default:
+		}
+
+		// Receive with non-blocking to allow checking context
+		msg, err := client.Receive(false)
+		if err != nil {
+			utils.Logger.ErrorF("auditd: error receiving message: %v", err)
+			a.cleanup()
+			return -1
+		}
+
+		if msg == nil {
+			// No message available, brief sleep to avoid busy loop
+			time.Sleep(10 * time.Millisecond)
+			continue
+		}
+
+		// Parse message type from raw data
+		msgType := auparse.AuditMessageType(msg.Type)
+
+		// Push to reassembler for event grouping
+		if err := reassembler.Push(msgType, msg.Data); err != nil {
+			utils.Logger.ErrorF("auditd: error pushing to reassembler: %v", err)
+		}
+	}
+}
+
+// runMaintenance periodically calls Maintain() to flush stale events
+func (a *AuditdCollector) runMaintenance(ctx context.Context) {
+	ticker := time.NewTicker(maintainInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			a.mu.Lock()
+			if a.reassembler != nil {
+				if err := a.reassembler.Maintain(); err != nil {
+					utils.Logger.ErrorF("auditd: error in reassembler maintenance: %v", err)
+				}
+			}
+			a.mu.Unlock()
+		}
+	}
+}
+
+// cleanup closes the client and reassembler
+func (a *AuditdCollector) cleanup() {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if a.reassembler != nil {
+		a.reassembler.Close()
+		a.reassembler = nil
+	}
+	if a.client != nil {
+		a.client.Close()
+		a.client = nil
+	}
+}
+
+// Stop stops the collector
+func (a *AuditdCollector) Stop() {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if a.cancel != nil {
+		a.cancel()
+	}
+}

agent/collector/auditd/auditd_other.go

@@ -0,0 +1,32 @@
+//go:build !linux
+// +build !linux
+
+package auditd
+
+import (
+	"context"
+
+	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/utmstack/UTMStack/agent/utils"
+)
+
+// AuditdCollector is a no-op stub for non-Linux platforms
+type AuditdCollector struct{}
+
+// New creates a new AuditdCollector (no-op on non-Linux)
+func New() *AuditdCollector {
+	return &AuditdCollector{}
+}
+
+// Name returns the collector name
+func (a *AuditdCollector) Name() string {
+	return "auditd"
+}
+
+// Start is a no-op on non-Linux platforms
+func (a *AuditdCollector) Start(ctx context.Context, queue chan *plugins.Log) {
+	utils.Logger.Info("auditd collector not supported on this platform, skipping")
+}
+
+// Stop is a no-op on non-Linux platforms
+func (a *AuditdCollector) Stop() {}

agent/collector/auditd/capabilities.go

@@ -0,0 +1,40 @@
+//go:build linux
+// +build linux
+
+package auditd
+
+import (
+	"os/exec"
+	"strings"
+
+	"github.com/utmstack/UTMStack/agent/utils"
+)
+
+// checkAuditCapability checks if the audit system is available and enabled.
+// Uses auditctl -s to verify audit status since /proc/sys/kernel/auditing
+// doesn't exist on all kernel versions.
+func checkAuditCapability() error {
+	// Check if auditctl exists
+	auditctlPath, err := exec.LookPath("auditctl")
+	if err != nil {
+		utils.Logger.ErrorF("auditd: auditctl not found in PATH: %v", err)
+		return err
+	}
+
+	// Run auditctl -s to check audit status
+	cmd := exec.Command(auditctlPath, "-s")
+	output, err := cmd.Output()
+	if err != nil {
+		utils.Logger.ErrorF("auditd: failed to run auditctl -s: %v", err)
+		return err
+	}
+
+	// Check if enabled=1 in output
+	if !strings.Contains(string(output), "enabled 1") && !strings.Contains(string(output), "enabled=1") {
+		utils.Logger.Info("auditd: kernel auditing is disabled (enabled != 1), collector will not start")
+		return nil
+	}
+
+	utils.Logger.Info("auditd: audit system is enabled and ready")
+	return nil
+}

agent/collector/auditd/client.go

@@ -0,0 +1,38 @@
+//go:build linux
+// +build linux
+
+package auditd
+
+import (
+	libaudit "github.com/elastic/go-libaudit/v2"
+)
+
+// auditReceiver interface wraps the go-libaudit client for testability
+type auditReceiver interface {
+	Receive(nonBlocking bool) (*libaudit.RawAuditMessage, error)
+	Close() error
+}
+
+// auditClientWrapper wraps the go-libaudit AuditClient to implement auditReceiver
+type auditClientWrapper struct {
+	client *libaudit.AuditClient
+}
+
+// Receive receives a raw audit message from the netlink socket
+func (w *auditClientWrapper) Receive(nonBlocking bool) (*libaudit.RawAuditMessage, error) {
+	return w.client.Receive(nonBlocking)
+}
+
+// Close closes the underlying audit client
+func (w *auditClientWrapper) Close() error {
+	return w.client.Close()
+}
+
+// newAuditClient creates a new multicast audit client wrapped in the auditReceiver interface
+func newAuditClient() (auditReceiver, error) {
+	client, err := libaudit.NewMulticastAuditClient(nil)
+	if err != nil {
+		return nil, err
+	}
+	return &auditClientWrapper{client: client}, nil
+}