From cad389f62f7d2413d11fdbe8e9b3e34bb0d12b0a Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 15:23:06 +0000 Subject: [PATCH] feat(ingestion): symlink tripwire for the Syncthing-shared brain dir (#157) (#159) Co-authored-by: Mathias --- ingestion/cmd/server/main.go | 10 +- ingestion/cmd/server/symlink_check.go | 39 ++++ ingestion/cmd/server/symlink_check_test.go | 65 ++++++ .../internal/symlinkguard/symlinkguard.go | 153 ++++++++++++++ .../symlinkguard/symlinkguard_test.go | 189 ++++++++++++++++++ 5 files changed, 455 insertions(+), 1 deletion(-) create mode 100644 ingestion/cmd/server/symlink_check.go create mode 100644 ingestion/cmd/server/symlink_check_test.go create mode 100644 ingestion/internal/symlinkguard/symlinkguard.go create mode 100644 ingestion/internal/symlinkguard/symlinkguard_test.go diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index 59def19..f0fc95d 100644 --- a/ingestion/cmd/server/main.go +++ b/ingestion/cmd/server/main.go @@ -33,6 +33,7 @@ import ( "github.com/mathiasbq/hyperguild/ingestion/internal/reranker" "github.com/mathiasbq/hyperguild/ingestion/internal/search" "github.com/mathiasbq/hyperguild/ingestion/internal/startupretry" + "github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard" "github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore" "github.com/mathiasbq/hyperguild/ingestion/internal/watcher" "github.com/mathiasbq/hyperguild/ingestion/internal/webhook" @@ -173,7 +174,8 @@ func alertingWarnings(getenv func(string) string) []string { var out []string if getenv("BRAIN_NTFY_URL") == "" { out = append(out, "alerting DISABLED: BRAIN_NTFY_URL unset — "+ - "brain_edit_root governance rewrites and capture audit degradation will not notify") + "brain_edit_root governance rewrites, capture audit degradation and the "+ + "brain symlink tripwire (hyperguild#157) will not notify") } return out } @@ -239,6 +241,10 @@ func main() { } brainDir := envOr("INGEST_BRAIN_DIR", "../brain") + if err := startupSymlinkCheck(brainDir); err != nil { + logger.Error("brain symlink tripwire", "err", err) + os.Exit(1) + } port := envOr("INGEST_PORT", "3300") llmURL := envOr("INGEST_LLM_URL", "http://iguana:4000/v1") @@ -360,6 +366,8 @@ func main() { } ctx := context.Background() + // hyperguild#157: report a symlink that appears in the brain while running. + go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier()) // Built before the watcher, the graph backfill and session ingestion, and // retried rather than fatal (#83). Two separate reasons: diff --git a/ingestion/cmd/server/symlink_check.go b/ingestion/cmd/server/symlink_check.go new file mode 100644 index 0000000..f1b987e --- /dev/null +++ b/ingestion/cmd/server/symlink_check.go @@ -0,0 +1,39 @@ +package main + +import ( + "errors" + "fmt" + "os" + "strconv" + "time" + + "github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard" +) + +// startupSymlinkCheck is the hyperguild#157 tripwire at boot: a symlink in the +// brain dir would let brain readers serve a file from outside the brain, so +// the server does not start until it is removed. A directory the scan cannot +// read could hide one, so it refuses on that too. A missing brain dir (first +// boot) is the only case it lets pass. +func startupSymlinkCheck(brainDir string) error { + r, err := symlinkguard.Scan(brainDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("refusing to start: cannot scan the brain dir %s (hyperguild#157): %w", brainDir, err) + } + if !r.Clean() { + return fmt.Errorf("refusing to start: brain dir %s has %s (hyperguild#157); fix them first", brainDir, r) + } + return nil +} + +// symlinkScanInterval reads BRAIN_SYMLINK_SCAN_INTERVAL (seconds), falling +// back to the default on an unset, non-numeric or non-positive value. +func symlinkScanInterval(getenv func(string) string) time.Duration { + if s, err := strconv.Atoi(getenv("BRAIN_SYMLINK_SCAN_INTERVAL")); err == nil && s > 0 { + return time.Duration(s) * time.Second + } + return symlinkguard.DefaultInterval +} diff --git a/ingestion/cmd/server/symlink_check_test.go b/ingestion/cmd/server/symlink_check_test.go new file mode 100644 index 0000000..0f6a357 --- /dev/null +++ b/ingestion/cmd/server/symlink_check_test.go @@ -0,0 +1,65 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// hyperguild#157: the server refuses to start while a symlink exists in the +// brain dir, or while any part of it cannot be scanned (which could hide +// one); a missing brain dir is the only thing it lets pass. +func TestStartupSymlinkCheckRefusesABrainWithASymlink(t *testing.T) { + brain := t.TempDir() + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "planted"))) + err := startupSymlinkCheck(brain) + require.Error(t, err) + assert.Contains(t, err.Error(), "planted") +} + +func TestStartupSymlinkCheckRefusesABrainWithAnUnreadableDir(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root reads mode-000 directories") + } + brain := t.TempDir() + locked := filepath.Join(brain, "aaa") + require.NoError(t, os.MkdirAll(locked, 0o755)) + require.NoError(t, os.Chmod(locked, 0o000)) + t.Cleanup(func() { _ = os.Chmod(locked, 0o755) }) + err := startupSymlinkCheck(brain) + require.Error(t, err) + assert.Contains(t, err.Error(), "aaa") +} + +func TestStartupSymlinkCheckPassesACleanOrMissingBrain(t *testing.T) { + brain := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(brain, "n.md"), []byte("x"), 0o644)) + assert.NoError(t, startupSymlinkCheck(brain)) + assert.NoError(t, startupSymlinkCheck(filepath.Join(brain, "missing"))) +} + +func TestSymlinkScanIntervalFallsBackOnNonPositive(t *testing.T) { + for _, v := range []string{"0", "-5", "", "junk"} { + got := symlinkScanInterval(func(string) string { return v }) + assert.Positive(t, got, "env %q", v) + } +} + +func TestAlertingWarningsNamesTheSymlinkTripwire(t *testing.T) { + w := strings.Join(alertingWarnings(func(string) string { return "" }), " ") + assert.Contains(t, w, "symlink tripwire") +} + +func TestStartupSymlinkCheckRefusesAnUnlistableBrainRoot(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root lists mode-0100 directories") + } + brain := t.TempDir() + require.NoError(t, os.Chmod(brain, 0o100)) + t.Cleanup(func() { _ = os.Chmod(brain, 0o755) }) + assert.Error(t, startupSymlinkCheck(brain)) +} diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go new file mode 100644 index 0000000..2dc11f0 --- /dev/null +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -0,0 +1,153 @@ +// Package symlinkguard is the hyperguild#157 tripwire. The brain directory is +// a Syncthing sendreceive folder and Syncthing syncs symlinks; several brain +// readers (search, vectorstore, context, graph) follow a symlink out of the +// brain and would index or return whatever it points to. Until every reader +// reads through an os.Root, a symlink inside the brain, or a directory the +// scan cannot read (which could hide one), is an incident: the server refuses +// to start on one (main), and Watch reports one that appears while it runs. +package symlinkguard + +import ( + "context" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" +) + +// DefaultInterval is used when Watch is given a non-positive interval. +const DefaultInterval = 5 * time.Minute + +// Notifier is the out-of-band alert sink (audit.Notifier satisfies it). +type Notifier interface { + Notify(ctx context.Context, msg string) error +} + +// Result is what a scan found: symlinks, and directories it could not read +// (each of which could hide a symlink). Paths are brain-relative. +type Result struct { + Symlinks []string + Unreadable []string +} + +// Clean reports whether the scan found nothing to act on. +func (r Result) Clean() bool { return len(r.Symlinks) == 0 && len(r.Unreadable) == 0 } + +// String describes the findings for a log line or alert. Paths are quoted, so +// a name containing a newline cannot blur two findings together. +func (r Result) String() string { + q := func(xs []string) string { + out := make([]string, len(xs)) + for i, x := range xs { + out[i] = strconv.Quote(x) + } + return strings.Join(out, ", ") + } + var parts []string + if len(r.Symlinks) > 0 { + parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks))) + } + if len(r.Unreadable) > 0 { + parts = append(parts, fmt.Sprintf("%d unreadable dir(s): %s", len(r.Unreadable), q(r.Unreadable))) + } + return strings.Join(parts, "; ") +} + +// Scan walks the brain dir and reports every symlink in it and every +// directory it could not read, continuing past both. Only the ROOT .git is +// skipped (it is brain-sync's, and no reader walks it); a nested .git is +// scanned like any other directory. brainDir itself may be a symlink (a +// volume mount); it is resolved first. The error is non-nil when the root +// cannot be resolved (wrapping os.ErrNotExist when it is missing) or cannot be +// listed. +func Scan(brainDir string) (Result, error) { + root, err := filepath.EvalSymlinks(brainDir) + if err != nil { + return Result{}, fmt.Errorf("resolve brain dir: %w", err) + } + var r Result + rootGit := filepath.Join(root, ".git") + walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { + rel, _ := filepath.Rel(root, p) + rel = filepath.ToSlash(rel) + if err != nil { + if p == root { + return err + } + r.Unreadable = append(r.Unreadable, rel) + if d != nil && d.IsDir() { + return filepath.SkipDir + } + return nil + } + if d.IsDir() && p == rootGit { + return filepath.SkipDir + } + if d.Type()&os.ModeSymlink != 0 { + r.Symlinks = append(r.Symlinks, rel) + } + return nil + }) + sort.Strings(r.Symlinks) + sort.Strings(r.Unreadable) + if walkErr != nil { + // Only the root's own error reaches here (deeper errors are recorded + // as Unreadable). A root that cannot be listed cannot be shown clean + // (PR #159 round 2: discarding this reported clean while search + // served a planted symlink). + return r, fmt.Errorf("scan brain dir: %w", walkErr) + } + return r, nil +} + +// Watch rescans every interval (DefaultInterval if non-positive) until ctx +// ends, and notifies once per distinct non-clean result. A result counts as +// reported only after Notify succeeds, so an alert lost to a notifier outage +// is retried on the next tick. +func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) { + if interval <= 0 { + interval = DefaultInterval + } + t := time.NewTicker(interval) + defer t.Stop() + reported := "" + for { + select { + case <-ctx.Done(): + return + case <-t.C: + } + r, err := Scan(brainDir) + var key, msg string + switch { + case err != nil: + key = "error: " + err.Error() + msg = "brain symlink tripwire (hyperguild#157): cannot scan the brain dir: " + err.Error() + case r.Clean(): + reported = "" + continue + default: + key = r.String() + msg = "brain symlink tripwire (hyperguild#157): " + key + + ". Readers may follow symlinks out of the brain; remove them." + } + if key == reported { + continue + } + slog.Error("brain symlink tripwire", "finding", key) + if n == nil { + reported = key + continue + } + if err := n.Notify(ctx, msg); err != nil { + slog.Warn("symlink tripwire notify failed; will retry", "err", err) + continue + } + reported = key + } +} diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go new file mode 100644 index 0000000..6b46112 --- /dev/null +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -0,0 +1,189 @@ +package symlinkguard + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// hyperguild#157: the brain dir is a Syncthing sendreceive folder and +// Syncthing syncs symlinks; several readers follow them out of the brain. +// Until every reader goes through an os.Root, a symlink anywhere in the brain, +// or a directory the scan cannot read (which could hide one), is an incident. + +func TestScanReportsSymlinksAtEveryDepthIncludingNestedGit(t *testing.T) { + brain := t.TempDir() + outside := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w", "facts", ".git"), 0o755)) + require.NoError(t, os.MkdirAll(filepath.Join(brain, ".git", "objects"), 0o755)) + require.NoError(t, os.Symlink(filepath.Join(outside, "f"), filepath.Join(brain, "wiki", "w", "facts", "leak.md"))) + require.NoError(t, os.Symlink(outside, filepath.Join(brain, "wiki", "dirlink"))) + // PR #159 review: only the ROOT .git is skipped; a nested .git is not + // special to the brain readers, which walk into it. + require.NoError(t, os.Symlink(filepath.Join(outside, "g"), filepath.Join(brain, "wiki", "w", "facts", ".git", "leak.md"))) + require.NoError(t, os.Symlink(outside, filepath.Join(brain, ".git", "objects", "alt"))) + + r, err := Scan(brain) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"wiki/w/facts/leak.md", "wiki/dirlink", "wiki/w/facts/.git/leak.md"}, r.Symlinks) + assert.Empty(t, r.Unreadable) +} + +// PR #159 review: WalkDir stopped at the first unreadable directory and the +// startup check passed, while search (which keeps walking) served a planted +// symlink. Scan now records the unreadable dir and keeps going. +func TestScanReportsUnreadableDirsAndKeepsWalkingPastThem(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root reads mode-000 directories") + } + brain := t.TempDir() + locked := filepath.Join(brain, "aaa") + require.NoError(t, os.MkdirAll(locked, 0o755)) + require.NoError(t, os.Chmod(locked, 0o000)) + t.Cleanup(func() { _ = os.Chmod(locked, 0o755) }) + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "zzz-planted"))) + + r, err := Scan(brain) + require.NoError(t, err) + assert.Equal(t, []string{"aaa"}, r.Unreadable) + assert.Equal(t, []string{"zzz-planted"}, r.Symlinks, "the walk continues past the unreadable dir") +} + +func TestScanResolvesASymlinkedRootAndReportsSymlinkedEntriesInsideIt(t *testing.T) { + real := t.TempDir() + link := filepath.Join(t.TempDir(), "brain") + require.NoError(t, os.Symlink(real, link)) + require.NoError(t, os.WriteFile(filepath.Join(real, "n.md"), []byte("x"), 0o644)) + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(real, "inner"))) + r, err := Scan(link) + require.NoError(t, err) + assert.Equal(t, []string{"inner"}, r.Symlinks) +} + +func TestScanOfAMissingRootIsErrNotExist(t *testing.T) { + _, err := Scan(filepath.Join(t.TempDir(), "nope")) + assert.True(t, errors.Is(err, os.ErrNotExist), "got %v", err) +} + +type recNotifier struct { + mu sync.Mutex + msgs []string + fail int // fail the first N calls +} + +func (r *recNotifier) Notify(_ context.Context, m string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.msgs = append(r.msgs, m) + if r.fail > 0 { + r.fail-- + return errors.New("ntfy down") + } + return nil +} + +func (r *recNotifier) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.msgs) } + +func TestWatchNotifiesOncePerDistinctSetAndAgainForANewSet(t *testing.T) { + brain := t.TempDir() + n := &recNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, 10*time.Millisecond, n) + + time.Sleep(50 * time.Millisecond) + assert.Zero(t, n.count(), "a clean brain raises nothing") + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "first"))) + require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond) + time.Sleep(60 * time.Millisecond) + assert.Equal(t, 1, n.count(), "the same set is reported once, not every tick") + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "second"))) + require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond) +} + +// PR #159 review: last was recorded before Notify, so an alert lost to an +// ntfy outage was never retried while the symlink stayed. +func TestWatchRetriesAnAlertThatFailedToSend(t *testing.T) { + brain := t.TempDir() + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "planted"))) + n := &recNotifier{fail: 2} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, 10*time.Millisecond, n) + require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 10*time.Millisecond) + time.Sleep(60 * time.Millisecond) + assert.Equal(t, 3, n.count(), "after one successful send, no more") +} + +func TestWatchAlertsOnAnUnreadableDir(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root reads mode-000 directories") + } + brain := t.TempDir() + locked := filepath.Join(brain, "locked") + require.NoError(t, os.MkdirAll(locked, 0o755)) + require.NoError(t, os.Chmod(locked, 0o000)) + t.Cleanup(func() { _ = os.Chmod(locked, 0o755) }) + n := &recNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, 10*time.Millisecond, n) + require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond) + assert.Contains(t, n.msgs[0], "locked") +} + +// PR #159 review: a non-positive interval panicked time.NewTicker inside the +// goroutine and crashlooped the server. +func TestWatchToleratesANonPositiveInterval(t *testing.T) { + brain := t.TempDir() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { Watch(ctx, brain, 0, nil); close(done) }() + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Watch did not return on cancel") + } +} + +// PR #159 round 2: a brain root that can be entered but not listed (mode +// 0100) made WalkDir fail at the root; the error was discarded and Scan +// reported clean, while search (which goes straight into wiki/) still served +// a planted symlink. A root that cannot be listed is now an error. +func TestScanOfAnUnlistableRootIsAnError(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root lists mode-0100 directories") + } + brain := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755)) + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "wiki", "leak.md"))) + require.NoError(t, os.Chmod(brain, 0o100)) + t.Cleanup(func() { _ = os.Chmod(brain, 0o755) }) + _, err := Scan(brain) + require.Error(t, err) + assert.False(t, errors.Is(err, os.ErrNotExist), "an unlistable root is not a missing root") +} + +// The same set that disappears and comes back is reported again. +func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) { + brain := t.TempDir() + link := filepath.Join(brain, "planted") + n := &recNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, 10*time.Millisecond, n) + require.NoError(t, os.Symlink(t.TempDir(), link)) + require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond) + require.NoError(t, os.Remove(link)) + time.Sleep(60 * time.Millisecond) + require.NoError(t, os.Symlink(t.TempDir(), link)) + require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond) +}