Evidence-fidelity gating, wider Sigma coverage, collector/canary reliability

CHANGELOG.md

@@ -1,6 +1,24 @@
 # Changelog
 All notable changes to this project follow Keep a Changelog and Semantic Versioning.
 
+## [Unreleased]
+### Added
+- Evidence fidelity: normalizer tags each event's source; `process_creation` rules now
+  only accept genuine process-creation telemetry (Sysmon/EventChannel eventdata, auditd
+  execve). Low-fidelity `full_log`/decoder scrapes can no longer produce a false `DETECTED`.
+- Sigma matcher coverage: `re` (regex), numeric `lt`/`lte`/`gt`/`gte`, and keyword
+  (full-text search) identifiers.
+- Wazuh collector date pre-filter: queries read only the day(s) a window spans instead of
+  the whole archive; scanner buffer enlarged so long archive events are no longer truncated.
+- Unit tests for the previously untested `canary` and `report` packages.
+### Changed
+- Canary now executes once and polls telemetry until a deadline (configurable via `Checker`)
+  instead of re-firing on fixed-interval retries.
+- Techniques whose collected events are all low-fidelity report `NO_TELEMETRY` (collection
+  gap) rather than `MISSED` (proven detection miss).
+- Dry-run / synthetic pipeline prints an unmistakable banner so its output cannot be mistaken
+  for real telemetry.
+
 ## [1.2.0] — 2026-07-04
 ### Added
 - Pipeline canary (positive control): per-run marker, gating logic, `make canary`

README.md

@@ -139,7 +139,17 @@ go test ./internal/evaluator/ -v -run Regression
 ## Supported Sigma subset
 
 The native Go matcher supports the Sigma specification subset needed for process-creation rules:
-field modifiers (`contains`, `startswith`, `endswith`, `|all`), condition grammar (`and`/`or`/`not`/parens/`1 of them`/`all of them`), and case-insensitive matching. Not yet supported: regex, aggregation expressions, `near`, correlated rules.
+field modifiers (`contains`, `startswith`, `endswith`, `|all`, `re`, numeric `lt`/`lte`/`gt`/`gte`),
+`*` wildcards in values, keyword (full-text) search identifiers, condition grammar
+(`and`/`or`/`not`/parens/`1 of them`/`all of them`), and case-insensitive matching. Not yet
+supported: aggregation expressions, `near`, correlated rules, `base64`/`cidr` modifiers.
+
+**Evidence fidelity.** For `process_creation` rules the evaluator only accepts genuine
+process-creation telemetry (Sysmon/EventChannel `eventdata`, auditd `execve`). Command-output and
+metadata scrapes (`full_log`, decoder name) are tagged low-fidelity and can never satisfy a
+process-creation rule — so a log line that merely mentions a binary cannot produce a false
+`DETECTED`. When a technique collects only low-fidelity events, the verdict is `NO_TELEMETRY`
+(a collection gap), not `MISSED`.
 
 ## Limitations
 

cmd/purpleloop/main.go

@@ -71,6 +71,15 @@ func main() {
 
 	ctx := context.Background()
 
+	// Warn loudly whenever any pipeline stage is synthetic, so a dry-run report
+	// can never be mistaken for real detection evidence.
+	if *dryRun || *victim == "" || *manager == "" {
+		fmt.Fprintln(os.Stderr, "┌─────────────────────────────────────────────────────────────┐")
+		fmt.Fprintln(os.Stderr, "│  DRY-RUN / SYNTHETIC PIPELINE — results are NOT real telemetry │")
+		fmt.Fprintln(os.Stderr, "│  Missing --victim-container or --manager-container.            │")
+		fmt.Fprintln(os.Stderr, "└─────────────────────────────────────────────────────────────┘")
+	}
+
 	switch {
 	case *arbiterFile != "":
 		if err := runArbiter(ctx, *arbiterFile, *output, *dryRun, *victim, *manager); err != nil {

internal/canary/canary.go

@@ -32,62 +32,91 @@ func NewMarker() string {
 	return "purpleloop-canary-" + hex.EncodeToString(b)
 }
 
-// Check executes the canary on one platform and verifies the marker was detected.
-func Check(ctx context.Context, marker string, exec model.Executor,
-	coll model.Collector, platform string, target model.Target, dryRun bool) Result {
+// Checker verifies the canary marker on one platform. Zero-value fields fall
+// back to production defaults; tests override the timing and rule paths.
+type Checker struct {
+	RulePath     string        // canary Sigma rule (default: rulePath const)
+	RulesDir     string        // rules root passed to the evaluator (default: "detections")
+	PollInterval time.Duration // gap between telemetry polls (default: 5s)
+	Timeout      time.Duration // total time to wait for the marker (default: 90s)
+	WindowPad    time.Duration // padding around the run window (default: 2m)
+}
 
-	r := Result{Platform: platform, Marker: marker}
+func (ck Checker) withDefaults() Checker {
+	if ck.RulePath == "" {
+		ck.RulePath = rulePath
+	}
+	if ck.RulesDir == "" {
+		ck.RulesDir = "detections"
+	}
+	if ck.PollInterval == 0 {
+		ck.PollInterval = 5 * time.Second
+	}
+	if ck.Timeout == 0 {
+		ck.Timeout = 90 * time.Second
+	}
+	if ck.WindowPad == 0 {
+		ck.WindowPad = 2 * time.Minute
+	}
+	return ck
+}
 
-	// Build the canary atomic
-	atomic := atomicFor(platform, marker)
+// Run executes the canary once, then polls telemetry until the marker is
+// detected or the timeout elapses. Executing a single time (rather than
+// re-firing on every retry) keeps the run window tight and the marker unique.
+func (ck Checker) Run(ctx context.Context, marker string, exec model.Executor,
+	coll model.Collector, platform string, target model.Target) Result {
 
-	// Execute with 3-try bounded retry for ingestion lag
-	for attempt := 0; attempt < 3; attempt++ {
-		if attempt > 0 {
-			select {
-			case <-ctx.Done():
-				r.Err = ctx.Err()
-				return r
-			case <-time.After(time.Duration(attempt+1) * 5 * time.Second):
-			}
-		}
+	ck = ck.withDefaults()
+	r := Result{Platform: platform, Marker: marker}
+	atomic := atomicFor(platform, marker)
 
-		run, err := exec.Run(ctx, atomic, target)
-		if err != nil {
-			r.Err = fmt.Errorf("canary execute: %w", err)
-			continue
-		}
-		_ = exec.Cleanup(ctx, atomic, target)
+	run, err := exec.Run(ctx, atomic, target)
+	if err != nil {
+		r.Err = fmt.Errorf("canary execute: %w", err)
+		return r
+	}
+	defer func() { _ = exec.Cleanup(ctx, atomic, target) }()
 
-		time.Sleep(10 * time.Second) // ingest delay
+	eval := evaluator.RuleMatcherEvaluator{RulesDir: ck.RulesDir}
+	rule := model.SigmaRule{Path: ck.RulePath, Title: "Pipeline Canary"}
+	window := run.Window(ck.WindowPad)
 
-		events, err := coll.Query(ctx, run.Window(2*time.Minute), target.Host)
+	deadline := time.Now().Add(ck.Timeout)
+	for {
+		events, err := coll.Query(ctx, window, target.Host)
 		if err != nil {
 			r.Err = fmt.Errorf("canary collect: %w", err)
-			continue
-		}
-		if len(events) == 0 {
-			continue
+		} else if len(events) > 0 {
+			verdict, evidence, err := eval.Evaluate(rule, events)
+			switch {
+			case err != nil:
+				r.Err = fmt.Errorf("canary evaluate: %w", err)
+			case verdict == model.Detected && evidenceContainsMarker(evidence, marker):
+				r.Healthy = true
+				r.Evidence = evidence
+				r.Err = nil
+				return r
+			}
 		}
 
-		// Evaluate canary rule
-		eval := evaluator.RuleMatcherEvaluator{RulesDir: "detections"}
-		rule := model.SigmaRule{Path: rulePath, Title: "Pipeline Canary"}
-		verdict, evidence, err := eval.Evaluate(rule, events)
-		if err != nil {
-			r.Err = fmt.Errorf("canary evaluate: %w", err)
-			continue
+		if time.Now().After(deadline) {
+			return r
 		}
-
-		// Must be DETECTED AND contain exact marker
-		if verdict == model.Detected && evidenceContainsMarker(evidence, marker) {
-			r.Healthy = true
-			r.Evidence = evidence
+		select {
+		case <-ctx.Done():
+			r.Err = ctx.Err()
 			return r
+		case <-time.After(ck.PollInterval):
 		}
 	}
+}
 
-	return r
+// Check executes the canary on one platform and verifies the marker was
+// detected, using production defaults.
+func Check(ctx context.Context, marker string, exec model.Executor,
+	coll model.Collector, platform string, target model.Target, dryRun bool) Result {
+	return Checker{}.Run(ctx, marker, exec, coll, platform, target)
 }
 
 func atomicFor(platform, marker string) model.AtomicTest {

internal/canary/canary_test.go

@@ -0,0 +1,86 @@
+package canary
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/jayelbotvibe-web/purple-loop/internal/model"
+)
+
+type fakeExecutor struct{ ran bool }
+
+func (f *fakeExecutor) Run(context.Context, model.AtomicTest, model.Target) (model.RunResult, error) {
+	f.ran = true
+	now := time.Now().UTC()
+	return model.RunResult{StartedAt: now, FinishedAt: now}, nil
+}
+func (f *fakeExecutor) Cleanup(context.Context, model.AtomicTest, model.Target) error { return nil }
+
+// fakeCollector returns whatever events it is configured with.
+type fakeCollector struct{ events []model.Event }
+
+func (f fakeCollector) Query(context.Context, model.TimeWindow, string) ([]model.Event, error) {
+	return f.events, nil
+}
+
+func testChecker() Checker {
+	return Checker{
+		RulePath:     "../../detections/canary/pipeline_canary.yml",
+		RulesDir:     "../../detections",
+		PollInterval: time.Millisecond,
+		Timeout:      20 * time.Millisecond,
+	}
+}
+
+func TestChecker_HealthyOnProcessTelemetry(t *testing.T) {
+	marker := NewMarker()
+	raw, _ := json.Marshal(map[string]string{
+		"Image":       "C:\\Windows\\System32\\cmd.exe",
+		"CommandLine": "cmd.exe /c echo " + marker,
+	})
+	coll := fakeCollector{events: []model.Event{{ID: "e1", Raw: raw}}}
+
+	r := testChecker().Run(context.Background(), marker, &fakeExecutor{}, coll, "windows", model.Target{Host: "win"})
+	if !r.Healthy {
+		t.Fatalf("expected healthy canary, got err=%v", r.Err)
+	}
+	if len(r.Evidence) == 0 {
+		t.Error("healthy canary should carry evidence")
+	}
+}
+
+func TestChecker_NotHealthyWithoutTelemetry(t *testing.T) {
+	marker := NewMarker()
+	coll := fakeCollector{events: nil}
+
+	r := testChecker().Run(context.Background(), marker, &fakeExecutor{}, coll, "linux", model.Target{Host: "victim"})
+	if r.Healthy {
+		t.Fatal("no telemetry must not report healthy")
+	}
+}
+
+// A command-output log echoing the marker is NOT process-creation evidence,
+// so the canary must not report the pipeline healthy on it.
+func TestChecker_LowFidelityLogIsNotHealthy(t *testing.T) {
+	marker := NewMarker()
+	raw := json.RawMessage(`{"full_log":"ossec: output: 'echo ` + marker + `': ` + marker + `"}`)
+	coll := fakeCollector{events: []model.Event{{ID: "log1", Raw: raw}}}
+
+	r := testChecker().Run(context.Background(), marker, &fakeExecutor{}, coll, "linux", model.Target{Host: "victim"})
+	if r.Healthy {
+		t.Fatal("low-fidelity command-output must not report the pipeline healthy")
+	}
+}
+
+func TestNewMarker_Unique(t *testing.T) {
+	a, b := NewMarker(), NewMarker()
+	if a == b {
+		t.Error("markers should be unique")
+	}
+	if !strings.HasPrefix(a, "purpleloop-canary-") {
+		t.Errorf("unexpected marker format: %q", a)
+	}
+}

internal/collector/prefilter_test.go

@@ -0,0 +1,44 @@
+package collector
+
+import (
+	"testing"
+	"time"
+
+	"github.com/jayelbotvibe-web/purple-loop/internal/model"
+)
+
+func TestDateAlternation(t *testing.T) {
+	mk := func(s, e string) model.TimeWindow {
+		start, _ := time.Parse(time.RFC3339, s)
+		end, _ := time.Parse(time.RFC3339, e)
+		return model.TimeWindow{Start: start, End: end}
+	}
+
+	cases := []struct {
+		name string
+		w    model.TimeWindow
+		want string
+	}{
+		{"same day", mk("2026-07-04T09:00:00Z", "2026-07-04T09:05:00Z"), "2026-07-04"},
+		{"spans midnight", mk("2026-07-04T23:59:00Z", "2026-07-05T00:02:00Z"), "2026-07-04|2026-07-05"},
+		{"zero window", model.TimeWindow{}, ""},
+		{"reversed", mk("2026-07-05T00:00:00Z", "2026-07-04T00:00:00Z"), ""},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := dateAlternation(c.w); got != c.want {
+				t.Errorf("dateAlternation = %q, want %q", got, c.want)
+			}
+		})
+	}
+}
+
+func TestShQuote(t *testing.T) {
+	if got := shQuote("victim01"); got != "'victim01'" {
+		t.Errorf("shQuote = %q", got)
+	}
+	// An embedded single quote must be escaped so it cannot break out.
+	if got := shQuote("a'b"); got != `'a'\''b'` {
+		t.Errorf("shQuote escaping = %q", got)
+	}
+}