From eff31b3a825c13e328ecda3c3ffd88c0c974516b Mon Sep 17 00:00:00 2001 From: Luna Date: Wed, 22 Jul 2026 03:10:26 -0700 Subject: [PATCH] feat(go): enrich snapshots asynchronously --- docs/GO_PORT_HANDOFF.md | 22 ++-- docs/PACKAGING.md | 7 +- go-agent/README.md | 6 +- go-agent/cmd/enodia-sentinel-go/main.go | 2 + go-agent/internal/enrich/enrich.go | 58 +++++++++ go-agent/internal/snapshot/store.go | 142 +++++++++++++++++++++++ go-agent/internal/snapshot/store_test.go | 47 ++++++++ 7 files changed, 270 insertions(+), 14 deletions(-) diff --git a/docs/GO_PORT_HANDOFF.md b/docs/GO_PORT_HANDOFF.md index 71ad581..e4a6e1e 100644 --- a/docs/GO_PORT_HANDOFF.md +++ b/docs/GO_PORT_HANDOFF.md @@ -1,8 +1,8 @@ # Go Port Handoff -Saved: 2026-07-22T02:10:00-07:00 +Saved: 2026-07-22T02:22:00-07:00 Branch: `main` -Base commit: `fd2bbba` (`feat(go): add bounded hot-path enrichment`) +Base commit: `1ab4add` (`feat(go): chain retained snapshot artifacts`) Status: implemented and green, but uncommitted ## Worktree warning @@ -10,9 +10,9 @@ Status: implemented and green, but uncommitted The checkout is intentionally dirty and contains work from multiple related continuations. Do not reset, clean, or broadly restage it. -- The validation-sidecar, incident-persistence, and bounded-enrichment tranches - are signed commits `f85c2e8`, `538d1d9`, and `fd2bbba`. The local-assurance - slice is uncommitted. +- The validation-sidecar, incident-persistence, bounded-enrichment, and local- + assurance tranches are signed commits `f85c2e8`, `538d1d9`, `fd2bbba`, and + `1ab4add`. The asynchronous-enrichment slice is uncommitted. - At this checkpoint, `git status --short` has 20 entries with untracked directories collapsed. - The Python GUI files and tests are separate pre-existing work. Preserve them @@ -99,8 +99,11 @@ static binary. `enodia.hash_chain.v1` records. Artifact SHA-256 is streamed, previous-hash lookup reads a bounded chain tail, and append completion is synced before the alert event is emitted. -- Package ownership, executable enrichment hashes, file metadata, richer - integrity anchors, and notification fan-out are not yet ported. +- A single 16-job asynchronous worker now streams SHA-256 only for regular + executables at most 8 MiB and adds lstat file metadata. Slow I/O occurs + outside the snapshot lock; a full queue simply leaves optional fields unknown. +- Package ownership, richer integrity anchors, and notification fan-out are not + yet ported. - No live system service was installed or enabled during development. ## Last green verification @@ -145,9 +148,8 @@ the suite still exits successfully. ## Resume here -1. Move slow enrichment collectors (ownership, bounded executable hashing, file - metadata, integrity anchors) behind a bounded asynchronous worker without - destructive response behavior. +1. Add bounded package-ownership and integrity-anchor collectors to the existing + asynchronous worker without destructive response behavior. 2. Connect the isolated Go snapshot/event state to explicit read-only management consumers while keeping Python authoritative. 3. Continue broader Phase 3 rule metadata/state parity before considering any diff --git a/docs/PACKAGING.md b/docs/PACKAGING.md index 6bed7a5..3b59598 100644 --- a/docs/PACKAGING.md +++ b/docs/PACKAGING.md @@ -95,8 +95,11 @@ sidecar also writes a private atomic `incidents.json` with the required `enodia.incident.v1` fields, process-lineage/time-window grouping, and additive correlation evidence. Hot-path Go enrichment is bounded and no-I/O: process and parent context, lineage, remote classification, and candidate paths come from -already-captured records. Python remains authoritative for ownership/hashes, -file and richer integrity metadata, notifications, and management consumers. +already-captured records. A bounded asynchronous worker subsequently hashes +regular executables no larger than 8 MiB and records file metadata; a full queue +skips those optional fields without delaying an alert. Python remains +authoritative for package ownership, richer integrity metadata, notifications, +and management consumers. Each snapshot text/JSON revision is SHA-256 hashed into a synchronized private `hash-chain.jsonl` using the stable `enodia.hash_chain.v1` record schema. diff --git a/go-agent/README.md b/go-agent/README.md index 665623b..efdcb46 100644 --- a/go-agent/README.md +++ b/go-agent/README.md @@ -115,8 +115,10 @@ the configured `incident_tracking`, `incident_window`, and `incident_lineage_depth` settings; same-second alert batches update one incident and one snapshot name idempotently. Hot-path enrichment derives bounded process/parent context, lineage, remote-IP classification, and candidate paths -from already-captured data without new I/O. Package ownership, hashes, file -metadata, richer integrity anchors, and notifications remain future parity +from already-captured data without new I/O. A 16-job asynchronous worker then +adds streamed SHA-256 values for regular executables up to 8 MiB and file +metadata; saturation leaves fields unknown rather than delaying alerts. Package +ownership, richer integrity anchors, and notifications remain future parity work. Snapshot text/JSON revisions append synchronized `enodia.hash_chain.v1` records to a private local `hash-chain.jsonl`. diff --git a/go-agent/cmd/enodia-sentinel-go/main.go b/go-agent/cmd/enodia-sentinel-go/main.go index cee64df..ef319d1 100644 --- a/go-agent/cmd/enodia-sentinel-go/main.go +++ b/go-agent/cmd/enodia-sentinel-go/main.go @@ -246,6 +246,8 @@ func run() error { ); err != nil { return err } + snapshotStore.EnableSlowEnrichment() + defer snapshotStore.Close() } notifier := sdnotify.FromEnvironment() if err := notifier.Notify("READY=1\nSTATUS=initialization complete; poll loop ready"); err != nil { diff --git a/go-agent/internal/enrich/enrich.go b/go-agent/internal/enrich/enrich.go index b612108..cb8cac8 100644 --- a/go-agent/internal/enrich/enrich.go +++ b/go-agent/internal/enrich/enrich.go @@ -6,9 +6,14 @@ package enrich import ( + "crypto/sha256" + "encoding/hex" + "io" + "os" "regexp" "sort" "strings" + "syscall" "codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/model" "codeberg.org/anassaeneroi/enodia-sentinal/go-agent/internal/netutil" @@ -18,6 +23,7 @@ const ( maxAlerts = 128 maxProcesses = 64 maxIndicators = 128 + MaxHashBytes = int64(8 << 20) ) var ( @@ -33,6 +39,58 @@ type Process struct { PPIDComm string } +// SlowResult contains bounded filesystem enrichment collected away from the +// alert-emission path. A file larger than MaxHashBytes is described but not +// hashed, avoiding unbounded worker time or memory use. +type SlowResult struct { + Processes map[int]map[string]any + Files map[string]map[string]any +} + +func CollectSlow(processes []Process, paths []string) SlowResult { + result := SlowResult{Processes: map[int]map[string]any{}, Files: map[string]map[string]any{}} + for _, process := range processes { + exe := cleanPath(process.Exe) + if exe == "" { + continue + } + result.Processes[process.PID] = map[string]any{"exe_sha256": boundedHash(exe)} + } + for _, path := range paths { + result.Files[path] = fileMetadata(path) + } + return result +} + +func boundedHash(path string) any { + info, err := os.Stat(path) + if err != nil || !info.Mode().IsRegular() || info.Size() > MaxHashBytes { + return nil + } + file, err := os.Open(path) + if err != nil { + return nil + } + defer file.Close() + hash := sha256.New() + if _, err := io.Copy(hash, io.LimitReader(file, MaxHashBytes+1)); err != nil { + return nil + } + return hex.EncodeToString(hash.Sum(nil)) +} + +func fileMetadata(path string) map[string]any { + info, err := os.Lstat(path) + if err != nil { + return map[string]any{"exists": false, "mode": nil, "uid": nil, "gid": nil} + } + mode, uid, gid := int(info.Mode().Perm()), -1, -1 + if stat, ok := info.Sys().(*syscall.Stat_t); ok { + uid, gid = int(stat.Uid), int(stat.Gid) + } + return map[string]any{"exists": true, "mode": mode, "uid": uid, "gid": gid} +} + // Build returns the stable Python enrichment keys without invoking any slow // collectors. Input and indicator caps keep attacker-controlled details from // causing unbounded snapshot work or output. diff --git a/go-agent/internal/snapshot/store.go b/go-agent/internal/snapshot/store.go index 92f3241..6beb549 100644 --- a/go-agent/internal/snapshot/store.go +++ b/go-agent/internal/snapshot/store.go @@ -67,9 +67,18 @@ type Store struct { LineageDepth int Assurance *assurance.Chain + slowJobs chan slowJob + slowClosed bool + slowWG sync.WaitGroup + mu sync.Mutex } +type slowJob struct { + jsonPath string + logPath string +} + // New preflights the isolated snapshot directory before service readiness. func New(dir, procRoot string, maxCount, maxAgeDays int) (*Store, error) { if dir == "" { @@ -126,6 +135,36 @@ func (s *Store) ConfigureIncidents(enabled bool, window, lineageDepth int) error return nil } +// EnableSlowEnrichment starts one bounded worker. It deliberately never runs +// package-manager commands: queue saturation or collector failure only leaves +// fields unknown and can never delay or reject an alert event. +func (s *Store) EnableSlowEnrichment() { + s.mu.Lock() + defer s.mu.Unlock() + if s.slowJobs != nil || s.slowClosed { + return + } + s.slowJobs = make(chan slowJob, 16) + go s.runSlowEnrichment(s.slowJobs) +} + +// Close drains accepted slow-enrichment work during a clean service shutdown. +func (s *Store) Close() { + s.mu.Lock() + if s.slowJobs == nil || s.slowClosed { + s.mu.Unlock() + return + } + close(s.slowJobs) + s.slowClosed = true + s.mu.Unlock() + s.slowWG.Wait() +} + +// FlushSlow waits for work accepted before the call. It is used by clean +// shutdown/tests; alert emission never waits for this collector. +func (s *Store) FlushSlow() { s.slowWG.Wait() } + // CaptureEvent merges one alert envelope into its second-granularity snapshot. // Python filenames use second precision, so merging avoids overwriting sibling // alerts emitted by one poll sweep or concurrent kernel sources. @@ -200,9 +239,112 @@ func (s *Store) CaptureEvent(event map[string]any) (string, error) { if err := s.pruneLocked(); err != nil { return "", err } + s.enqueueSlow(slowJob{jsonPath: jsonPath, logPath: base + ".log"}) return filepath.Base(base + ".log"), nil } +func (s *Store) enqueueSlow(job slowJob) { + if s.slowJobs == nil || s.slowClosed { + return + } + s.slowWG.Add(1) + select { + case s.slowJobs <- job: + default: + s.slowWG.Done() + } +} + +func (s *Store) runSlowEnrichment(jobs <-chan slowJob) { + for job := range jobs { + s.applySlowEnrichment(job) + s.slowWG.Done() + } +} + +func (s *Store) applySlowEnrichment(job slowJob) { + raw, err := os.ReadFile(job.jsonPath) + if err != nil { + return + } + var report Report + if json.Unmarshal(raw, &report) != nil || report.Schema != Schema { + return + } + processes := enrichmentProcesses(report.Processes) + paths := enrichmentPaths(report.Enrichment) + updates := enrich.CollectSlow(processes, paths) + + // All slow I/O has completed. Take the lock only to merge into the newest + // same-second report, so a concurrent alert capture never loses evidence. + s.mu.Lock() + defer s.mu.Unlock() + raw, err = os.ReadFile(job.jsonPath) + if err != nil || json.Unmarshal(raw, &report) != nil || report.Schema != Schema { + return + } + mergeSlow(&report, updates) + if err := writeJSONAtomic(job.jsonPath, report); err != nil { + return + } + if err := writeAtomic(job.logPath, []byte(formatText(report))); err != nil { + return + } + if s.Assurance != nil { + captured, err := time.Parse(time.RFC3339Nano, report.Time) + if err != nil { + return + } + _, _ = s.Assurance.Append("snapshot-log", job.logPath, captured) + _, _ = s.Assurance.Append("snapshot-json", job.jsonPath, captured) + } +} + +func enrichmentPaths(value map[string]any) []string { + rows, _ := value["files"].([]any) + paths := make([]string, 0, len(rows)) + for _, row := range rows { + if item, ok := row.(map[string]any); ok { + if path, ok := item["path"].(string); ok && path != "" { + paths = append(paths, path) + } + } + } + return paths +} + +func mergeSlow(report *Report, updates enrich.SlowResult) { + if report.Enrichment == nil { + return + } + if rows, ok := report.Enrichment["processes"].([]any); ok { + for _, row := range rows { + item, ok := row.(map[string]any) + if !ok { + continue + } + pid, _ := item["pid"].(float64) + if update := updates.Processes[int(pid)]; update != nil { + item["exe_sha256"] = update["exe_sha256"] + } + } + } + if rows, ok := report.Enrichment["files"].([]any); ok { + for _, row := range rows { + item, ok := row.(map[string]any) + if !ok { + continue + } + path, _ := item["path"].(string) + if update := updates.Files[path]; update != nil { + for key, value := range update { + item[key] = value + } + } + } + } +} + func (s *Store) collectProcesses(alerts []model.Alert) []ProcessDetail { pids := alertPIDs(alerts) result := make([]ProcessDetail, 0, len(pids)) diff --git a/go-agent/internal/snapshot/store_test.go b/go-agent/internal/snapshot/store_test.go index 905905e..40d3ab9 100644 --- a/go-agent/internal/snapshot/store_test.go +++ b/go-agent/internal/snapshot/store_test.go @@ -130,6 +130,53 @@ func TestCaptureEventLinksMergedSnapshotToOneIncident(t *testing.T) { } } +func TestSlowEnrichmentAddsBoundedHashAndFileMetadata(t *testing.T) { + store, err := New(t.TempDir(), "/unused", 10, 30) + if err != nil { + t.Fatal(err) + } + store.EnableSlowEnrichment() + defer store.Close() + exe := filepath.Join(store.Dir, "payload") + watch := filepath.Join(store.Dir, "persistence.service") + if err := os.WriteFile(exe, []byte("stage-one"), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(watch, []byte("[Service]"), 0o600); err != nil { + t.Fatal(err) + } + store.Collect = func(pid int) ProcessDetail { + return ProcessDetail{PID: pid, Comm: "payload", Exe: exe, FDs: map[string]string{}} + } + event := alertEvent("2026-07-22T10:11:12Z", model.Alert{SID: 1, Severity: "HIGH", Signature: "test", Classtype: "test", Key: "file", Detail: "modified " + watch, PIDs: []int{42}}) + if _, err := store.CaptureEvent(event); err != nil { + t.Fatal(err) + } + store.FlushSlow() + raw, err := os.ReadFile(filepath.Join(store.Dir, "alert-20260722-101112.json")) + if err != nil { + t.Fatal(err) + } + var report Report + if err := json.Unmarshal(raw, &report); err != nil { + t.Fatal(err) + } + processes := report.Enrichment["processes"].([]any) + process := processes[0].(map[string]any) + if digest, ok := process["exe_sha256"].(string); !ok || len(digest) != 64 { + t.Fatalf("process enrichment=%#v", process) + } + files := report.Enrichment["files"].([]any) + file := files[0].(map[string]any) + if file["exists"] != true || file["mode"] != float64(0o600) { + t.Fatalf("file enrichment=%#v", file) + } + logRaw, err := os.ReadFile(filepath.Join(store.Dir, "alert-20260722-101112.log")) + if err != nil || !strings.Contains(string(logRaw), "ENODIA SENTINEL ALERT") { + t.Fatalf("slow log err=%v text=%q", err, logRaw) + } +} + func TestRetentionPrunesCountAndAgeAsPairs(t *testing.T) { store, err := New(t.TempDir(), "/unused", 2, 1) if err != nil {