Fix unsound regex fast path, output path collisions, and drop-sink validation

services/cli/internal/app/app.go

@@ -170,9 +170,14 @@ func runPipeline(ctx context.Context, args []string, s Streams) (err error) {
 			usage(s.Err)
 			return fmt.Errorf("run: --input must be text or jsonl, got %q", *input)
 		}
-		for _, path := range inputs {
-			if path == *reviewPath {
-				return fmt.Errorf("run: --review path %q is also an input file", *reviewPath)
+		if *reviewPath != "" {
+			if samePath(*reviewPath, *rulesPath) {
+				return fmt.Errorf("run: --review path %q is also the rules file", *reviewPath)
+			}
+			for _, path := range inputs {
+				if samePath(*reviewPath, path) {
+					return fmt.Errorf("run: --review path %q is also an input file", *reviewPath)
+				}
 			}
 		}
 		ruleStage, lerr := rules.Load(*rulesPath)
@@ -316,13 +321,10 @@ func buildFromConfig(path string, s Streams, inputs []string, cleanup *[]func()
 // checkOutputPaths rejects a config whose file sinks collide with an input file
 // or with each other, before anything is opened and truncated.
 func checkOutputPaths(cfg *config.Config, configPath, base string, inputs []string) error {
-	protected := map[string]bool{normPath(configPath): true}
-	for _, in := range inputs {
-		protected[normPath(in)] = true
-	}
+	protected := append([]string{configPath}, inputs...)
 	for _, sp := range cfg.Stages {
 		if sp.Path != "" {
-			protected[normPath(resolve(base, sp.Path))] = true
+			protected = append(protected, resolve(base, sp.Path))
 		}
 	}
 	specs := []config.SinkSpec{cfg.Sink}
@@ -341,23 +343,39 @@ func checkOutputPaths(cfg *config.Config, configPath, base string, inputs []stri
 	if stdouts > 1 {
 		return errors.New("run: two sinks write to stdout")
 	}
-	outs := make(map[string]bool)
+	var outs []string
 	for _, sp := range specs {
 		if sp.Type != "jsonl" || sp.Path == "" {
 			continue
 		}
-		key := normPath(resolve(base, sp.Path))
-		if protected[key] {
-			return fmt.Errorf("run: config output %q collides with an input, the config, or a stage declaration file", sp.Path)
+		p := resolve(base, sp.Path)
+		for _, pr := range protected {
+			if samePath(p, pr) {
+				return fmt.Errorf("run: config output %q collides with an input, the config, or a stage declaration file", sp.Path)
+			}
 		}
-		if outs[key] {
-			return fmt.Errorf("run: config writes %q more than once", sp.Path)
+		for _, o := range outs {
+			if samePath(p, o) {
+				return fmt.Errorf("run: config writes %q more than once", sp.Path)
+			}
 		}
-		outs[key] = true
+		outs = append(outs, p)
 	}
 	return nil
 }
 
+// samePath reports whether a and b name the same file: by filesystem identity
+// when both exist (catches hard links and case aliases), by normalized path
+// otherwise.
+func samePath(a, b string) bool {
+	ia, errA := os.Stat(a)
+	ib, errB := os.Stat(b)
+	if errA == nil && errB == nil {
+		return os.SameFile(ia, ib)
+	}
+	return normPath(a) == normPath(b)
+}
+
 // normPath resolves p to an absolute, symlink-resolved path so two spellings of
 // the same file compare equal.
 func normPath(p string) string {

services/cli/internal/app/app_test.go

@@ -192,6 +192,60 @@ func TestRunRejectsReviewEqualToInput(t *testing.T) {
 	}
 }
 
+func TestRunRejectsReviewAliasingInput(t *testing.T) {
+	const content = "GET /healthz 200\npayment failed for user 42\n"
+	cases := []struct {
+		name   string
+		review func(dir, input string) string // maps the real input path to an aliased review spelling
+	}{
+		{"dot-slash prefix", func(_, input string) string { return "." + string(filepath.Separator) + filepath.Base(input) }},
+		{"absolute spelling", func(_, input string) string { return input }},
+		{"symlink to input", func(dir, input string) string {
+			link := filepath.Join(dir, "link.log")
+			if err := os.Symlink(input, link); err != nil {
+				t.Skipf("symlink unsupported: %v", err)
+			}
+			return link
+		}},
+		{"hard link to input", func(dir, input string) string {
+			link := filepath.Join(dir, "hard.log")
+			if err := os.Link(input, link); err != nil {
+				t.Skipf("hard link unsupported: %v", err)
+			}
+			return link
+		}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			dir := t.TempDir()
+			rulesPath := writeFile(t, dir, "rules.yml", testRules)
+			input := writeFile(t, dir, "in.log", content)
+			review := tc.review(dir, input)
+
+			// Run from the input's directory so a bare-basename input path resolves there.
+			cwd, err := os.Getwd()
+			if err != nil {
+				t.Fatalf("getwd: %v", err)
+			}
+			if err := os.Chdir(dir); err != nil {
+				t.Fatalf("chdir: %v", err)
+			}
+			t.Cleanup(func() { _ = os.Chdir(cwd) })
+
+			var out, errOut bytes.Buffer
+			runErr := Run(context.Background(),
+				[]string{"run", "--rules", rulesPath, "--review", review, filepath.Base(input)},
+				Streams{In: strings.NewReader(""), Out: &out, Err: &errOut})
+			if runErr == nil || !strings.Contains(runErr.Error(), "also an input") {
+				t.Fatalf("Run() error = %v, want a review/input conflict error", runErr)
+			}
+			if data, _ := os.ReadFile(input); string(data) != content {
+				t.Fatalf("input file was truncated to %q; the aliased review path must not zero the input", data)
+			}
+		})
+	}
+}
+
 func TestRunRequiresRulesFlag(t *testing.T) {
 	var out, errOut bytes.Buffer
 	err := Run(context.Background(), []string{"run"},
@@ -242,3 +296,59 @@ func TestRunRejectsInvalidMinConfidence(t *testing.T) {
 		t.Fatalf("Run() error = %v, want min confidence validation error", err)
 	}
 }
+
+func TestRunRejectsReviewAliasingRulesFile(t *testing.T) {
+	dir := t.TempDir()
+	rulesPath := writeFile(t, dir, "rules.yml", testRules)
+	input := writeFile(t, dir, "in.log", "GET /healthz 200\n")
+	rulesContent, err := os.ReadFile(rulesPath)
+	if err != nil {
+		t.Fatalf("read rules: %v", err)
+	}
+
+	link := filepath.Join(dir, "rules-link.yml")
+	if err := os.Link(rulesPath, link); err != nil {
+		if err = os.Symlink(rulesPath, link); err != nil {
+			t.Skipf("hard link and symlink unsupported: %v", err)
+		}
+	}
+	for _, review := range []string{rulesPath, link} {
+		var out, errOut bytes.Buffer
+		runErr := Run(context.Background(),
+			[]string{"run", "--rules", rulesPath, "--review", review, input},
+			Streams{In: strings.NewReader(""), Out: &out, Err: &errOut})
+		if runErr == nil || !strings.Contains(runErr.Error(), "also the rules file") {
+			t.Fatalf("Run(review=%q) error = %v, want a review/rules conflict error", review, runErr)
+		}
+		if data, _ := os.ReadFile(rulesPath); string(data) != string(rulesContent) {
+			t.Fatalf("rules file was truncated to %q", data)
+		}
+	}
+}
+
+func TestConfigRejectsOutputHardLinkedToInput(t *testing.T) {
+	dir := t.TempDir()
+	writeFile(t, dir, "rules.yml", testRules)
+	input := writeFile(t, dir, "in.log", "GET /healthz 200\n")
+	link := filepath.Join(dir, "out.jsonl")
+	if err := os.Link(input, link); err != nil {
+		t.Skipf("hard link unsupported: %v", err)
+	}
+	cfgPath := writeFile(t, dir, "cfg.yaml", `version: 1
+input: { type: text }
+stages:
+  - { id: rules, type: rules, path: rules.yml, gate: 1.0 }
+sink: { type: jsonl, path: out.jsonl }
+`)
+
+	var out, errOut bytes.Buffer
+	runErr := Run(context.Background(),
+		[]string{"run", "--config", cfgPath, input},
+		Streams{In: strings.NewReader(""), Out: &out, Err: &errOut})
+	if runErr == nil || !strings.Contains(runErr.Error(), "collides") {
+		t.Fatalf("Run() error = %v, want an output collision error", runErr)
+	}
+	if data, _ := os.ReadFile(input); string(data) != "GET /healthz 200\n" {
+		t.Fatalf("input file was truncated to %q", data)
+	}
+}

shared/pkg/config/config.go

@@ -153,6 +153,9 @@ func (c *Config) Validate() error {
 	if err := c.Sink.validate("sink"); err != nil {
 		return err
 	}
+	if c.Sink.Type == "drop" {
+		return errors.New("config: the default sink cannot be drop")
+	}
 	for category, sk := range c.Routes {
 		if category == "" {
 			return errors.New("config: a route category must not be empty")

shared/pkg/config/config_test.go

@@ -44,6 +44,7 @@ func TestValidate(t *testing.T) {
 		}, "either path or stream"},
 		{"bad stream", func(c *config.Config) { c.Sink = config.SinkSpec{Type: "jsonl", Stream: "socket"} }, "stream"},
 		{"drop sink with target", func(c *config.Config) { c.Sink = config.SinkSpec{Type: "drop", Path: "x"} }, "takes no path"},
+		{"drop default sink", func(c *config.Config) { c.Sink = config.SinkSpec{Type: "drop"} }, "default sink cannot be drop"},
 		{"empty route category", func(c *config.Config) {
 			c.Routes = map[string]config.SinkSpec{"": {Type: "drop"}}
 		}, "route category must not be empty"},

shared/pkg/stage/rules/rules.go

@@ -15,6 +15,7 @@ import (
 	"os"
 	"reflect"
 	"regexp"
+	"regexp/syntax"
 	"strconv"
 	"strings"
 
@@ -297,7 +298,7 @@ func regexCheck(pattern, label string) (check, error) {
 	desc := fmt.Sprintf("regex %q", pattern)
 	lit, complete := re.LiteralPrefix()
 	switch {
-	case lit != "" && complete:
+	case lit != "" && complete && !hasZeroWidthAssertion(pattern):
 		b := []byte(lit)
 		return check{
 			desc:  desc,
@@ -317,6 +318,30 @@ func regexCheck(pattern, label string) (check, error) {
 	}
 }
 
+// hasZeroWidthAssertion reports whether pattern contains ^ $ \A \z \b or \B.
+// LiteralPrefix ignores these, so the substring-only fast path is unsound
+// whenever one is present.
+func hasZeroWidthAssertion(pattern string) bool {
+	re, err := syntax.Parse(pattern, syntax.Perl)
+	if err != nil {
+		return true
+	}
+	return containsAssertion(re)
+}
+
+func containsAssertion(re *syntax.Regexp) bool {
+	switch re.Op {
+	case syntax.OpBeginLine, syntax.OpEndLine, syntax.OpBeginText, syntax.OpEndText, syntax.OpWordBoundary, syntax.OpNoWordBoundary:
+		return true
+	}
+	for _, sub := range re.Sub {
+		if containsAssertion(sub) {
+			return true
+		}
+	}
+	return false
+}
+
 func groupCheck(m Matcher, label string) (check, error) {
 	set := 0
 	for _, on := range []bool{m.Contains != nil, m.Regex != nil, m.Field != nil} {
@@ -384,7 +409,7 @@ func fieldCheck(fm FieldMatcher, label string) (check, error) {
 		desc = fmt.Sprintf("field %s matches %q", path, re.String())
 		lit, complete := re.LiteralPrefix()
 		switch {
-		case lit != "" && complete:
+		case lit != "" && complete && !hasZeroWidthAssertion(*fm.Regex):
 			pred = func(fields map[string]any) bool {
 				s, ok := lookupString(fields, segments)
 				return ok && strings.Contains(s, lit)

shared/pkg/stage/rules/rules_test.go

@@ -6,6 +6,8 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -74,6 +76,66 @@ func rec(fields map[string]any) domain.Record {
 	return domain.Record{Kind: domain.KindJSON, Fields: fields}
 }
 
+// TestClassifyAnchoredRegexDoesNotOverMatch guards the literal-prefilter fast
+// path: a fully-anchored literal pattern (^X$ / \AX\z) must match only the exact
+// string, never an arbitrary substring. re.LiteralPrefix reports complete=true
+// for these patterns yet drops the anchors, so the fast path must still confirm
+// the anchors with the compiled regex.
+func TestClassifyAnchoredRegexDoesNotOverMatch(t *testing.T) {
+	bodyCases := []struct {
+		name    string
+		pattern string
+		payload string
+		want    error
+	}{
+		{"caret-dollar exact matches", "^PING$", "PING", nil},
+		{"caret-dollar rejects substring", "^PING$", "ERROR received unexpected PING flood from 10.0.0.1", stage.ErrUnclassified},
+		{"escaped anchors exact matches", `\APING\z`, "PING", nil},
+		{"escaped anchors reject substring", `\APING\z`, "unexpected PING flood", stage.ErrUnclassified},
+		{"literal with space exact matches", "^GET /health$", "GET /health", nil},
+		{"literal with space rejects substring", "^GET /health$", "GET /health 200", stage.ErrUnclassified},
+		{"word boundary matches whole word", `\bPING\b`, "a PING b", nil},
+		{"word boundary rejects embedded word", `\bPING\b`, "aPINGb", stage.ErrUnclassified},
+		{"negated word boundary matches embedded", `\BING\B`, "aPINGb", nil},
+		{"negated word boundary rejects word edge", `\BPING\B`, "a PING b", stage.ErrUnclassified},
+	}
+	for _, tc := range bodyCases {
+		t.Run("body/"+tc.name, func(t *testing.T) {
+			yml := "rules:\n  - category: heartbeat\n    regex: [" + strconv.Quote(tc.pattern) + "]\n"
+			s := mustParse(t, yml)
+			c, err := s.Classify(context.Background(), domain.Record{Data: []byte(tc.payload)})
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("Classify(%q) err = %v, want %v", tc.payload, err, tc.want)
+			}
+			if tc.want == nil && c.Category != "heartbeat" {
+				t.Fatalf("Classify(%q) = %+v, want heartbeat", tc.payload, c)
+			}
+		})
+	}
+
+	fieldCases := []struct {
+		name  string
+		level string
+		want  error
+	}{
+		{"exact value matches", "error", nil},
+		{"substring rejected", "non-error state cleared", stage.ErrUnclassified},
+	}
+	for _, tc := range fieldCases {
+		t.Run("field/"+tc.name, func(t *testing.T) {
+			yml := "rules:\n  - category: err\n    fields:\n      - path: level\n        regex: \"^error$\"\n"
+			s := mustParse(t, yml)
+			c, err := s.Classify(context.Background(), rec(map[string]any{"level": tc.level}))
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("Classify(level=%q) err = %v, want %v", tc.level, err, tc.want)
+			}
+			if tc.want == nil && c.Category != "err" {
+				t.Fatalf("Classify(level=%q) = %+v, want err", tc.level, c)
+			}
+		})
+	}
+}
+
 func TestClassifyFieldMatchers(t *testing.T) {
 	cases := []struct {
 		name   string
@@ -441,3 +503,43 @@ func TestLookupNumberKinds(t *testing.T) {
 		t.Fatal("lookupNumber() ok = true for a missing path, want false")
 	}
 }
+
+// FuzzRegexRuleParity pins the regex fast paths (prefilter, complete-literal)
+// to plain regexp behavior over arbitrary payloads.
+func FuzzRegexRuleParity(f *testing.F) {
+	patterns := []string{
+		"^PING$", `\APING\z`, `\bPING\b`, `\BING\B`, "PING", "^PING", "PING$",
+		"^GET /health$", "p[io]ng", "(warn|error) disk", "payment (failed|declined)",
+	}
+	stages := make([]stage.Stage, len(patterns))
+	regexps := make([]*regexp.Regexp, len(patterns))
+	for i, pat := range patterns {
+		yml := "rules:\n  - category: hit\n    regex: [" + strconv.Quote(pat) + "]\n"
+		st, err := Parse([]byte(yml))
+		if err != nil {
+			f.Fatalf("Parse(%q): %v", pat, err)
+		}
+		stages[i] = st
+		regexps[i] = regexp.MustCompile(pat)
+	}
+	f.Add("PING")
+	f.Add("a PING b")
+	f.Add("aPINGb")
+	f.Add("GET /health 200")
+	f.Add("payment declined for order 7")
+	f.Add("")
+	f.Fuzz(func(t *testing.T, data string) {
+		r := domain.Record{ID: "f", Data: []byte(data)}
+		for i, st := range stages {
+			_, err := st.Classify(context.Background(), r)
+			got := err == nil
+			want := regexps[i].MatchString(data)
+			if got != want {
+				t.Fatalf("pattern %q on %q: rule matched=%v, regexp matched=%v", patterns[i], data, got, want)
+			}
+			if err != nil && !errors.Is(err, stage.ErrUnclassified) {
+				t.Fatalf("pattern %q on %q: unexpected error %v", patterns[i], data, err)
+			}
+		}
+	})
+}