enodia-sentinal/go-agent/internal/agent/agent.go

// SPDX-License-Identifier: GPL-3.0-or-later

// Package agent owns the parallel Go sweep loop. It emits JSON-compatible
// records but does not write Python daemon state or replace the production
// service during migration.
package agent

import (
	"context"
	"fmt"
	"os"
	"sync"
	"time"

	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/baseline"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/config"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/detectors"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/fim"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/model"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/pkgdb"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/rootcheck"
	"codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/schema"
)

const Version = "0.7.0"

// CaptureFunc returns one injectable SystemState-equivalent sweep.
type CaptureFunc func() (model.State, error)

// EmitFunc receives one enodia.event.v1-compatible record.
type EmitFunc func(map[string]any) error

// Agent is the migration sidecar sweep loop.
type Agent struct {
	Config  config.Config
	Capture CaptureFunc
	Host    func() (string, error)
	Now     func() time.Time
	// Lifecycle is nil for deterministic fixtures. Live sidecars attach a
	// baseline manager so startup grace and durable first-seen state match the
	// Python oracle without contaminating parity inputs.
	Lifecycle *baseline.Manager
	// FIM is optional even when the sidecar is stateful. It owns slow hashing
	// and package verification and hands completed alerts back to Sweep, which
	// keeps the existing cooldown and snapshot pipeline authoritative.
	FIM *fim.Manager
	// PackageDB watches the mutable package checksum database itself. It is kept
	// separate from FIM because a package-manager database edit can otherwise
	// make pacman-based file verification appear clean.
	PackageDB *pkgdb.Manager
	// Rootcheck is likewise a sidecar-only asynchronous integrity engine. It
	// uses the captured socket state but collects independent proc/sys/tool views
	// outside this sweep's detector and event-output locks.
	Rootcheck *rootcheck.Manager

	cooldownMu    sync.Mutex
	cooldowns     map[string]time.Time
	probeMu       sync.RWMutex
	ebpf          string
	ebpfExec      string
	ebpfSyscall   string
	initializeMu  sync.Mutex
	initialized   bool
	initializeErr error
}

// New builds an agent with production clock and hostname providers.
func New(cfg config.Config, capture CaptureFunc) *Agent {
	return &Agent{
		Config:      cfg,
		Capture:     capture,
		Host:        os.Hostname,
		Now:         time.Now,
		ebpf:        "unknown",
		ebpfExec:    "unknown",
		ebpfSyscall: "unknown",
	}
}

// Sweep captures state once and returns alert events followed by one status
// event. The executable replaces the zero-value retention fields after its
// optional snapshot sink has durably processed the preceding alert records.
func (a *Agent) Sweep() ([]map[string]any, error) {
	state, err := a.Capture()
	if err != nil {
		return nil, err
	}
	now := a.Now()
	if a.Lifecycle != nil {
		if err := a.Lifecycle.Prepare(&state, now); err != nil {
			return nil, err
		}
	}
	if a.FIM != nil {
		// Scheduling is deliberately before detector evaluation, but FIM itself
		// runs asynchronously so a large hash tree never delays this sweep.
		a.FIM.MaybeStart(now)
	}
	if a.PackageDB != nil {
		// Like FIM, DB fingerprinting happens away from detector evaluation. A
		// completed tamper result still returns through the shared alert pipeline.
		a.PackageDB.MaybeStart(now)
	}
	if a.Rootcheck != nil {
		// Capture provides the socket view rootcheck compares with procfs and ps;
		// pass this immutable sweep snapshot to avoid a second full capture.
		a.Rootcheck.MaybeStart(now, state)
	}
	host, err := a.Host()
	if err != nil {
		return nil, err
	}
	timestamp := now.Local().Format(time.RFC3339Nano)

	alerts := make([]model.Alert, 0)
	// Preserve the relative order of Python detectors.REGISTRY. Snapshot and
	// parity consumers rely on stable alert ordering even while unported slots
	// are temporarily absent from the Go implementation.
	if a.Config.Enabled("reverse_shell") {
		alerts = append(alerts, detectors.ReverseShell(state, a.Config)...)
	}
	if a.Config.Enabled("ld_preload") {
		alerts = append(alerts, detectors.LDPreload(state)...)
	}
	if a.Config.Enabled("deleted_exe") {
		alerts = append(alerts, detectors.DeletedExe(state)...)
	}
	if a.Config.Enabled("input_snooper") {
		alerts = append(alerts, detectors.InputSnooper(state, a.Config)...)
	}
	if a.Config.Enabled("credential_access") {
		alerts = append(alerts, detectors.CredentialAccess(state, a.Config)...)
	}
	if a.Config.Enabled("stealth_network") {
		alerts = append(alerts, detectors.StealthNetwork(state, a.Config)...)
	}
	if a.Config.Enabled("memory_obfuscation") {
		alerts = append(alerts, detectors.MemoryObfuscation(state, a.Config)...)
	}
	if a.Config.Enabled("egress") {
		alerts = append(alerts, detectors.Egress(state, a.Config)...)
	}
	// Python's registry uses listener_baseline as one shared arming gate for
	// all four baseline-diff detectors, including first_seen and persistence.
	if state.ListenerBaseline != nil {
		if a.Config.Enabled("first_seen") {
			alerts = append(alerts, detectors.FirstSeen(state, a.Config)...)
		}
		if a.Config.Enabled("new_listener") {
			alerts = append(alerts, detectors.NewListener(state, a.Config)...)
		}
		if a.Config.Enabled("persistence") {
			alerts = append(alerts, detectors.Persistence(state, a.Config)...)
		}
		if a.Config.Enabled("new_suid") {
			alerts = append(alerts, detectors.NewSUID(state, a.Config)...)
		}
	}
	if a.Lifecycle != nil {
		if err := a.Lifecycle.Commit(state, now); err != nil {
			return nil, err
		}
	}
	if a.FIM != nil {
		// Background engines return results here instead of emitting directly, so
		// they use this Agent's shared cooldown, snapshot, and event ordering.
		alerts = append(alerts, a.FIM.Drain()...)
	}
	if a.PackageDB != nil {
		alerts = append(alerts, a.PackageDB.Drain()...)
	}
	if a.Rootcheck != nil {
		alerts = append(alerts, a.Rootcheck.Drain()...)
	}
	events, err := a.alertEvents(alerts, now, host, timestamp)
	if err != nil {
		return nil, err
	}
	ebpfStatus, ebpfExecStatus, ebpfSyscallStatus := a.probeStatus()
	status := schema.Status{
		Schema:      schema.StatusV1,
		Version:     Version,
		Running:     true,
		TotalAlerts: 0,
		Counts:      map[string]int{},
		LastAlert:   nil,
		EBPF:        ebpfStatus,
		EBPFExec:    ebpfExecStatus,
		EBPFSyscall: ebpfSyscallStatus,
		Host:        host,
	}
	record, err := schema.Build("status", status, host, timestamp)
	if err != nil {
		return nil, err
	}
	return append(events, record), nil
}

// SetExecProbeStatus updates both the compatibility eBPF field and the
// specific exec-probe field. It is safe to call from a source goroutine after
// an asynchronous reader failure.
func (a *Agent) SetExecProbeStatus(status string) {
	a.probeMu.Lock()
	defer a.probeMu.Unlock()
	a.ebpf = status
	a.ebpfExec = status
}

// SetSyscallProbeStatus updates the syscall-specific probe field. The legacy
// eBPF field continues to mirror exec status, matching the Python contract.
func (a *Agent) SetSyscallProbeStatus(status string) {
	a.probeMu.Lock()
	defer a.probeMu.Unlock()
	a.ebpfSyscall = status
}

func (a *Agent) probeStatus() (string, string, string) {
	a.probeMu.RLock()
	defer a.probeMu.RUnlock()
	return a.ebpf, a.ebpfExec, a.ebpfSyscall
}

// AlertEvents applies the shared cooldown gate and wraps asynchronous detector
// results in the same schema used by sweep alerts. Live kernel sources call
// this method instead of maintaining a second emission path.
func (a *Agent) AlertEvents(alerts []model.Alert) ([]map[string]any, error) {
	now := a.Now()
	host, err := a.Host()
	if err != nil {
		return nil, err
	}
	return a.alertEvents(alerts, now, host, now.Local().Format(time.RFC3339Nano))
}

func (a *Agent) alertEvents(alerts []model.Alert, now time.Time, host, timestamp string) ([]map[string]any, error) {
	alerts = a.freshAlerts(alerts, now)
	records := make([]map[string]any, 0, len(alerts))
	for _, alert := range alerts {
		record, err := schema.Build("alert", alert, host, timestamp)
		if err != nil {
			return nil, err
		}
		records = append(records, record)
	}
	return records, nil
}

// Initialize establishes live baselines before asynchronous event sources are
// started. It is idempotent so an executable can enforce startup ordering and
// Run can still safely own initialization for library callers.
func (a *Agent) Initialize() error {
	a.initializeMu.Lock()
	defer a.initializeMu.Unlock()
	if a.initialized {
		return a.initializeErr
	}
	a.initialized = true
	if a.Lifecycle != nil {
		if a.Capture == nil {
			a.initializeErr = fmt.Errorf("capture function is required for baseline initialization")
			return a.initializeErr
		}
		// Use the same injectable clock for lifecycle gates and event timestamps;
		// otherwise fixed-clock tests could arm against wall time by accident.
		a.Lifecycle.Now = a.Now
		a.initializeErr = a.Lifecycle.Initialize(baseline.CaptureFunc(a.Capture))
		if a.initializeErr != nil {
			return a.initializeErr
		}
	}
	if a.FIM != nil {
		// FIM must establish its immutable baseline before its first scheduled
		// scan. The manager writes only the caller-selected sidecar LogDir.
		a.initializeErr = a.FIM.Initialize()
		if a.initializeErr != nil {
			return a.initializeErr
		}
	}
	if a.PackageDB != nil {
		// Establishing the anchor precedes monitoring, just as the Python daemon
		// does. Later unexplained changes deliberately do not refresh it.
		a.initializeErr = a.PackageDB.Initialize()
	}
	return a.initializeErr
}

// freshAlerts mirrors Python's per-key cooldown gate. The mutex matters once
// live event sources join polling: asynchronous kernel alerts and sweep alerts
// must consume the same cooldown slots instead of racing into duplicates.
func (a *Agent) freshAlerts(alerts []model.Alert, now time.Time) []model.Alert {
	a.cooldownMu.Lock()
	defer a.cooldownMu.Unlock()
	if a.cooldowns == nil {
		a.cooldowns = make(map[string]time.Time)
	}
	fresh := make([]model.Alert, 0, len(alerts))
	for _, alert := range alerts {
		previous, seen := a.cooldowns[alert.Key]
		if seen && now.Sub(previous) < a.Config.Cooldown {
			continue
		}
		a.cooldowns[alert.Key] = now
		fresh = append(fresh, alert)
	}
	return fresh
}

// Run emits one sweep immediately and then waits for each configured interval.
// once is used by parity checks and operator-visible smoke tests.
func (a *Agent) Run(ctx context.Context, once bool, emit EmitFunc) error {
	if a.Capture == nil || emit == nil {
		return fmt.Errorf("capture and emit functions are required")
	}
	if err := a.Initialize(); err != nil {
		return err
	}
	for {
		events, err := a.Sweep()
		if err != nil {
			return err
		}
		for _, event := range events {
			if err := emit(event); err != nil {
				return err
			}
		}
		if once {
			return nil
		}
		timer := time.NewTimer(a.Config.SampleInterval)
		select {
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return nil
		case <-timer.C:
		}
	}
}