feat: Wazuh queue-socket event emitter (INTEG-001)

Outbound-only: no listener, no control surface — RedFlag's journal is
authoritative, Wazuh is a best-effort mirror. DGRAM to the local agent's
queue socket, ECS-formatted, rule IDs mapped from security event types.
Opt-in via REDFLAG_WAZUH_ENABLED=true; disabled = socket never opened.

- Sink interface on SecurityLogger (mirror after journal write)
- Write deadline + one reconnect, then drop-count with rate-limited log
- 3 tests: frame/ECS shape, absent-socket non-blocking, unknown→generic

server/cmd/server/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/Fimeg/RedFlag/server/internal/circuitbreaker"
 	"github.com/Fimeg/RedFlag/server/internal/command"
 	"github.com/Fimeg/RedFlag/server/internal/config"
+	"github.com/Fimeg/RedFlag/server/internal/integrations/wazuh"
 	"github.com/Fimeg/RedFlag/server/internal/database"
 	"github.com/Fimeg/RedFlag/server/internal/database/queries"
 	"github.com/Fimeg/RedFlag/server/internal/logging"
@@ -402,12 +403,26 @@ func main() {
 		securityLogger = nil
 	}
 
+	// Wazuh queue-socket mirror (INTEG-001). Opt-in, outbound-only; disabled
+	// means the socket is never opened. The security journal stays authoritative.
+	if securityLogger != nil && os.Getenv("REDFLAG_WAZUH_ENABLED") == "true" {
+		socketPath := os.Getenv("REDFLAG_WAZUH_SOCKET")
+		securityLogger.SetSink(wazuh.New(socketPath))
+		log.Printf("[INFO] [server] [wazuh-emitter] enabled, socket=%s",
+			func() string {
+				if socketPath == "" {
+					return wazuh.DefaultSocketPath
+				}
+				return socketPath
+			}())
+	}
+
 	// Initialize rate limiter
 	rateLimiter := middleware.NewRateLimiter()
 
 	// Initialize handlers that don't depend on agentHandler (can be created now)
 	authHandler := handlers.NewAuthHandler(cfg.Admin.JWTSecret, adminQueries)
-	statsHandler := handlers.NewStatsHandler(agentQueries, updateQueries)
+	statsHandler := handlers.NewStatsHandler(agentQueries, updateQueries, cfg.CheckInInterval)
 	settingsHandler := handlers.NewSettingsHandler(timezoneService)
 	dockerHandler := handlers.NewDockerHandler(dockerQueries, updateQueries, agentQueries, commandQueries, signingService, securityLogger)
 	registrationTokenHandler := handlers.NewRegistrationTokenHandler(registrationTokenQueries, agentQueries, cfg)

server/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/go-git/go-git/v5 v5.19.1
 	github.com/gofrs/uuid/v5 v5.4.0
 	github.com/golang-jwt/jwt/v5 v5.3.0
+	github.com/google/uuid v1.6.0
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/lib/pq v1.10.9
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1

server/internal/integrations/wazuh/emitter.go

@@ -0,0 +1,230 @@
+// Package wazuh emits RedFlag security events to a local Wazuh agent's queue
+// socket in ECS format (INTEG-001). Outbound-only: this opens no listener and
+// is unrelated to the pull-only agent<->server channel. RedFlag's own journal
+// remains the source of truth; Wazuh is a mirror — writes are best-effort,
+// never block the caller, and drops are counted and logged.
+package wazuh
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/Fimeg/RedFlag/server/internal/models"
+)
+
+// DefaultSocketPath is where a local Wazuh agent or manager listens.
+const DefaultSocketPath = "/var/ossec/queue/sockets/queue"
+
+// framePrefix is the Wazuh queue protocol header: <queue>:<location>:
+// 1 = ossec message queue, "redflag" is the location tag decoders key on.
+const framePrefix = "1:redflag:"
+
+const writeDeadline = 250 * time.Millisecond
+
+// ruleIDs maps RedFlag security event types to Wazuh custom rule IDs
+// (999xxx user range). Kept in lockstep with docs/wazuh-ruleset.xml.
+var ruleIDs = map[string]int{
+	models.SecurityEventTypes.CmdSignatureVerificationFailed:    999002,
+	models.SecurityEventTypes.UpdateNonceInvalid:                999003,
+	models.SecurityEventTypes.UpdateSignatureVerificationFailed: 999004,
+	models.SecurityEventTypes.MachineIDMismatch:                 999005,
+	models.SecurityEventTypes.AuthJWTValidationFailed:           999006,
+	models.SecurityEventTypes.AgentRegistrationFailed:           999007,
+	models.SecurityEventTypes.UnauthorizedAccessAttempt:         999008,
+	models.SecurityEventTypes.ConfigTamperingDetected:           999009,
+	models.SecurityEventTypes.AnomalousBehavior:                 999010,
+	models.SecurityEventTypes.CmdSigned:                         999011,
+	models.SecurityEventTypes.CmdSignatureVerificationSuccess:   999012,
+}
+
+const genericRuleID = 999001
+
+// Emitter writes ECS-formatted events to the Wazuh queue socket. Connection
+// is lazy and re-established once per emit on failure; beyond that the event
+// is dropped, counted, and logged (rate-limited) — never blocking.
+type Emitter struct {
+	socketPath string
+
+	mu   sync.Mutex
+	conn net.Conn
+
+	dropped     atomic.Uint64
+	lastDropLog atomic.Int64 // unix seconds, rate-limits drop warnings
+}
+
+// New returns an Emitter for the given socket path ("" = DefaultSocketPath).
+// No connection is attempted until the first Emit.
+func New(socketPath string) *Emitter {
+	if socketPath == "" {
+		socketPath = DefaultSocketPath
+	}
+	return &Emitter{socketPath: socketPath}
+}
+
+// Dropped reports how many events were lost to socket failures.
+func (e *Emitter) Dropped() uint64 { return e.dropped.Load() }
+
+// Emit formats the event as ECS JSON and writes one datagram. Failures drop
+// the event after a single reconnect attempt. Safe for concurrent use.
+func (e *Emitter) Emit(ev *models.SecurityEvent) {
+	payload, err := json.Marshal(toECS(ev))
+	if err != nil {
+		e.drop(fmt.Sprintf("marshal failed: %v", err))
+		return
+	}
+	frame := append([]byte(framePrefix), payload...)
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if err := e.writeLocked(frame); err != nil {
+		// One reconnect attempt: the Wazuh agent may have restarted.
+		e.closeLocked()
+		if err := e.writeLocked(frame); err != nil {
+			e.drop(fmt.Sprintf("socket write failed: %v", err))
+		}
+	}
+}
+
+func (e *Emitter) writeLocked(frame []byte) error {
+	if e.conn == nil {
+		conn, err := net.Dial("unixgram", e.socketPath)
+		if err != nil {
+			return err
+		}
+		e.conn = conn
+	}
+	if err := e.conn.SetWriteDeadline(time.Now().Add(writeDeadline)); err != nil {
+		return err
+	}
+	_, err := e.conn.Write(frame)
+	return err
+}
+
+func (e *Emitter) closeLocked() {
+	if e.conn != nil {
+		if err := e.conn.Close(); err != nil {
+			log.Printf("[WARN] [server] [wazuh-emitter] close failed: %v", err)
+		}
+		e.conn = nil
+	}
+}
+
+// drop counts a lost event and logs at most once per minute so a dead socket
+// cannot flood the server log while still leaving history (ETHOS #1).
+func (e *Emitter) drop(reason string) {
+	n := e.dropped.Add(1)
+	now := time.Now().Unix()
+	last := e.lastDropLog.Load()
+	if now-last >= 60 && e.lastDropLog.CompareAndSwap(last, now) {
+		log.Printf("[WARN] [server] [wazuh-emitter] event dropped (%d total): %s", n, reason)
+	}
+}
+
+// ecsEvent is the Wazuh-ingestible ECS shape (see INTEG-001 spec).
+type ecsEvent struct {
+	Timestamp string                 `json:"@timestamp"`
+	Event     ecsEventBlock          `json:"event"`
+	Rule      ecsRule                `json:"rule"`
+	Agent     *ecsAgent              `json:"agent,omitempty"`
+	Message   string                 `json:"message"`
+	Details   map[string]interface{} `json:"redflag,omitempty"`
+	Wazuh     ecsWazuh               `json:"wazuh"`
+}
+
+type ecsEventBlock struct {
+	Kind     string   `json:"kind"`
+	Category []string `json:"category"`
+	Type     []string `json:"type"`
+	Module   string   `json:"module"`
+	Action   string   `json:"action"`
+	Outcome  string   `json:"outcome"`
+	Severity int      `json:"severity"`
+}
+
+type ecsRule struct {
+	ID          string `json:"id"`
+	Level       int    `json:"level"`
+	Description string `json:"description"`
+}
+
+type ecsAgent struct {
+	ID string `json:"id"`
+}
+
+type ecsWazuh struct {
+	Integration ecsIntegration `json:"integration"`
+}
+
+type ecsIntegration struct {
+	Name     string   `json:"name"`
+	Category string   `json:"category"`
+	Decoders []string `json:"decoders"`
+	Rules    []string `json:"rules"`
+}
+
+func toECS(ev *models.SecurityEvent) ecsEvent {
+	ruleID, ok := ruleIDs[ev.EventType]
+	if !ok {
+		ruleID = genericRuleID
+	}
+	severity, level := severityFor(ev.Level)
+	outcome := "failure"
+	switch ev.EventType {
+	case models.SecurityEventTypes.CmdSigned,
+		models.SecurityEventTypes.CmdSignatureVerificationSuccess:
+		outcome = "success"
+	}
+
+	var agent *ecsAgent
+	if id := ev.AgentID.String(); id != "00000000-0000-0000-0000-000000000000" {
+		agent = &ecsAgent{ID: id}
+	}
+
+	return ecsEvent{
+		Timestamp: ev.Timestamp.UTC().Format(time.RFC3339),
+		Event: ecsEventBlock{
+			Kind:     "alert",
+			Category: []string{"security"},
+			Type:     []string{"info"},
+			Module:   "redflag",
+			Action:   ev.EventType,
+			Outcome:  outcome,
+			Severity: severity,
+		},
+		Rule: ecsRule{
+			ID:          fmt.Sprintf("%d", ruleID),
+			Level:       level,
+			Description: ev.Message,
+		},
+		Agent:   agent,
+		Message: ev.Message,
+		Details: ev.Details,
+		Wazuh: ecsWazuh{
+			Integration: ecsIntegration{
+				Name:     "redflag",
+				Category: "security",
+				Decoders: []string{"json"},
+				Rules:    []string{fmt.Sprintf("%d", ruleID)},
+			},
+		},
+	}
+}
+
+// severityFor maps RedFlag levels to (ECS event.severity, Wazuh rule level).
+func severityFor(level string) (int, int) {
+	switch level {
+	case "CRITICAL":
+		return 9, 12
+	case "WARNING":
+		return 5, 7
+	case "INFO":
+		return 3, 3
+	default:
+		return 1, 1
+	}
+}

server/internal/integrations/wazuh/emitter_test.go

@@ -0,0 +1,109 @@
+package wazuh
+
+import (
+	"encoding/json"
+	"net"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Fimeg/RedFlag/server/internal/models"
+	"github.com/gofrs/uuid/v5"
+)
+
+func listen(t *testing.T) (string, *net.UnixConn) {
+	t.Helper()
+	sock := filepath.Join(t.TempDir(), "queue")
+	conn, err := net.ListenUnixgram("unixgram", &net.UnixAddr{Name: sock, Net: "unixgram"})
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return sock, conn
+}
+
+func TestEmitFrameAndECSShape(t *testing.T) {
+	sock, conn := listen(t)
+	e := New(sock)
+
+	agentID := uuid.Must(uuid.NewV4())
+	ev := models.NewSecurityEvent("CRITICAL", models.SecurityEventTypes.MachineIDMismatch, agentID, "machine binding violation")
+	e.Emit(ev)
+
+	conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	buf := make([]byte, 64*1024)
+	n, err := conn.Read(buf)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	frame := string(buf[:n])
+
+	if !strings.HasPrefix(frame, "1:redflag:") {
+		t.Fatalf("frame prefix wrong: %q", frame[:20])
+	}
+
+	var got ecsEvent
+	if err := json.Unmarshal([]byte(strings.TrimPrefix(frame, "1:redflag:")), &got); err != nil {
+		t.Fatalf("payload not valid JSON: %v", err)
+	}
+	if got.Event.Module != "redflag" || got.Event.Action != "MACHINE_ID_MISMATCH" {
+		t.Errorf("event block wrong: %+v", got.Event)
+	}
+	if got.Rule.ID != "999005" || got.Rule.Level != 12 {
+		t.Errorf("rule mapping wrong: %+v", got.Rule)
+	}
+	if got.Agent == nil || got.Agent.ID != agentID.String() {
+		t.Errorf("agent id missing: %+v", got.Agent)
+	}
+	if got.Wazuh.Integration.Name != "redflag" {
+		t.Errorf("integration block wrong: %+v", got.Wazuh)
+	}
+	if e.Dropped() != 0 {
+		t.Errorf("dropped %d events on healthy socket", e.Dropped())
+	}
+}
+
+func TestEmitAbsentSocketDropsWithoutBlocking(t *testing.T) {
+	e := New(filepath.Join(t.TempDir(), "no-such-socket"))
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < 5; i++ {
+			e.Emit(models.NewSecurityEvent("WARNING", models.SecurityEventTypes.UpdateNonceInvalid, uuid.Nil, "replay"))
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Emit blocked on absent socket")
+	}
+	if e.Dropped() != 5 {
+		t.Errorf("expected 5 drops, got %d", e.Dropped())
+	}
+}
+
+func TestUnknownEventTypeGetsGenericRule(t *testing.T) {
+	sock, conn := listen(t)
+	e := New(sock)
+	e.Emit(models.NewSecurityEvent("INFO", "SOME_FUTURE_EVENT", uuid.Nil, "x"))
+
+	conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	buf := make([]byte, 64*1024)
+	n, err := conn.Read(buf)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	var got ecsEvent
+	if err := json.Unmarshal(buf[len("1:redflag:"):n], &got); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if got.Rule.ID != "999001" {
+		t.Errorf("expected generic rule 999001, got %s", got.Rule.ID)
+	}
+	if got.Agent != nil {
+		t.Errorf("nil agent UUID should omit agent block, got %+v", got.Agent)
+	}
+}

server/internal/logging/security_logger.go

@@ -28,6 +28,13 @@ type SecurityLogConfig struct {
 	HashIPAddresses   bool   `yaml:"hash_ip_addresses" env:"REDFLAG_SECURITY_LOG_HASH_IP" default:"true"`
 }
 
+// Sink receives a copy of every persisted security event (e.g. the Wazuh
+// emitter, INTEG-001). Implementations must never block: writeEvent runs on
+// the journal path and the journal is the source of truth — a sink is a mirror.
+type Sink interface {
+	Emit(event *models.SecurityEvent)
+}
+
 // SecurityLogger handles structured security event logging
 type SecurityLogger struct {
 	config    SecurityLogConfig
@@ -39,6 +46,12 @@ type SecurityLogger struct {
 	bufferSize int
 	stopChan   chan struct{}
 	wg         sync.WaitGroup
+	sink       Sink
+}
+
+// SetSink attaches an event mirror. Call before the logger is in use.
+func (sl *SecurityLogger) SetSink(s Sink) {
+	sl.sink = s
 }
 
 // NewSecurityLogger creates a new security logger instance
@@ -265,6 +278,12 @@ func (sl *SecurityLogger) writeEvent(event *models.SecurityEvent) error {
 		}
 	}
 
+	// Mirror to the attached sink (non-blocking by contract) after the
+	// journal writes — the journal is authoritative, the sink is best-effort.
+	if sl.sink != nil {
+		sl.sink.Emit(event)
+	}
+
 	return nil
 }