Compare commits

..
Author SHA1 Message Date
mathiasandClaude Opus 5.5 4c7fcb6bdc test(ingestion): /mcp's JWT validator and resource_metadata; narrow comments
CI / Lint / Test / Vet (pull_request) Successful in 1m2s
CI / Mirror to GitHub (pull_request) Skipped
PR #183 review (comment 15474):
- TestBuildMuxAnswers503ToAJWTOnMCPWhileTheIssuerIsUnresolved: building
  /mcp's middleware with a nil validator turns it red.
- TestBuildMuxMCP401CarriesResourceMetadataInWWWAuthenticate: building it
  with resourceMetadataURL "" turns it red.
- main.go, routes.go and the testMux comment now say exactly what is
  untested: main's restAuth fields and public map passed to buildMux, and
  ListenAndServe (hyperguild#182). Stale public-route list in the
  TestBuildMuxDoesNotWrapPublicRoutes comment corrected.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 16:41:33 +02:00
mathiasandClaude Opus 5.5 4a1abafd8f fix(ingestion): build /mcp's bearer and the metrics wrap inside buildMux
/mcp's bearer was built in main.go and the buildMux tests used stubs, so
dropping it would have kept CI green. buildMux now takes the MCP handler
and wraps it with the same BearerMiddleware and credential as the REST
API. "/mcp" and "GET /metrics" left the public allowlist, so passing
either as a public route is refused. buildMux also registers GET
/metrics and returns the metrics-wrapped handler, so main passes its
result to ListenAndServe unchanged.

Flip-tested: registering /mcp unwrapped turns
TestBuildMuxRefusesMCPWithoutAValidToken red, and dropping the metrics
wrap turns the four metrics tests red.

Comments now claim only what is tested. The ListenAndServe line and the
symlink tripwire wiring in main (#154 comment) are still untested and
tracked in #182.

Closes #154

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 16:41:33 +02:00
13 changed files with 32 additions and 1170 deletions
+1 -1
View File
@@ -68,7 +68,7 @@ require (
github.com/segmentio/asm v1.2.0 // indirect
golang.org/x/crypto v0.51.0 // indirect
golang.org/x/sync v0.21.0 // indirect
golang.org/x/sys v0.45.0
golang.org/x/sys v0.45.0 // indirect
golang.org/x/text v0.39.0 // indirect
gopkg.in/yaml.v3 v3.0.1
)
+5 -17
View File
@@ -3,18 +3,15 @@ package api
import (
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io/fs"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"os"
"path/filepath"
"strings"
"time"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/claim"
"github.com/mathiasbq/hyperguild/ingestion/internal/notes"
)
// ContentHash returns the lowercase hex sha256 of b. It is the note's
@@ -94,23 +91,14 @@ func UpdateNote(brainDir string, opts UpdateNoteOptions) (relPath, contentHash,
rel = filepath.ToSlash(rel)
}
// The path is caller-supplied (brain_update): resolve it as brain_edit
// does, knowledge/ and wiki/ notes only, and read and rewrite it through
// the brain root, never through a symlink (hyperguild#169).
t, err := notes.Resolve(brainDir, rel, notes.ScopeNotes)
rel, abs, err := resolveWithin(brainDir, rel)
if err != nil {
return "", "", "", err
}
rel = t.Rel
bfs, err := brainfs.Open(brainDir)
if err != nil {
return "", "", "", fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
prior, err := bfs.ReadRegular(rel)
prior, err := os.ReadFile(abs)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
if os.IsNotExist(err) {
return "", "", "", fmt.Errorf("note %q does not exist: use brain_write to create", rel)
}
return "", "", "", fmt.Errorf("read target: %w", err)
@@ -176,7 +164,7 @@ func UpdateNote(brainDir string, opts UpdateNoteOptions) (relPath, contentHash,
return "", "", "", err
}
out := []byte(stamped)
if err := bfs.WriteRegular(rel, out); err != nil {
if err := os.WriteFile(abs, out, 0o644); err != nil {
return "", "", "", fmt.Errorf("write: %w", err)
}
return rel, ContentHash(out), priorHash, nil
@@ -1,36 +0,0 @@
package api_test
import (
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
)
// hyperguild#173: a directory symlink inside wiki/ that points elsewhere in
// the brain redirected brain_update's write (os.Root follows in-root
// intermediate symlinks). Every directory above the note must be real.
func TestUpdateNoteRefusesANoteBehindASymlinkedDirectory(t *testing.T) {
brain := t.TempDir()
victim := filepath.Join(brain, "raw", "facts", "v.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("---\ntitle: v\n---\nuntouched\n"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "facts"), filepath.Join(brain, "wiki", "w", "alias")))
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/alias/v.md", Content: "replaced"})
assert.ErrorIs(t, err, brainfs.ErrNotRegular)
got, _ := os.ReadFile(victim)
assert.Contains(t, string(got), "untouched")
real := filepath.Join(brain, "wiki", "w", "facts", "r.md")
require.NoError(t, os.MkdirAll(filepath.Dir(real), 0o755))
require.NoError(t, os.WriteFile(real, []byte("---\ntitle: r\n---\nold\n"), 0o644))
_, _, _, err = api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/facts/r.md", Content: "new"})
require.NoError(t, err, "positive control: a note in a real directory is updated")
}
-112
View File
@@ -1,112 +0,0 @@
package api_test
import (
"os"
"path/filepath"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
)
// brain_update passed the caller's path through resolveWithin, a lexical
// containment check only: any file in the brain could be superseded
// (classification.yaml, raw/, sessions/), and a final symlink was written
// through. It now resolves like brain_edit: knowledge/ and wiki/ notes only,
// read and rewritten through the brain root.
func TestUpdateNoteRefusesPathsOutsideKnowledgeAndWikiNotes(t *testing.T) {
brain := t.TempDir()
files := map[string]string{
"classification.yaml": "levels: {}\n",
"raw/r.md": "raw\n",
"sessions/s.jsonl": "{}\n",
"wiki/w/facts/ok.md": "---\ntitle: ok\n---\nold\n",
"knowledge/k.md": "---\ntitle: k\n---\nold\n",
"wiki/w/facts/x.txt": "txt\n",
}
for rel, body := range files {
p := filepath.Join(brain, filepath.FromSlash(rel))
require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
require.NoError(t, os.WriteFile(p, []byte(body), 0o644))
}
for _, rel := range []string{"classification.yaml", "raw/r.md", "sessions/s.jsonl", "wiki/w/facts/x.txt", "wiki/w/../../raw/r.md"} {
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: rel, Content: "replaced"})
assert.Error(t, err, rel)
}
for rel, body := range files {
if rel == "wiki/w/facts/ok.md" || rel == "knowledge/k.md" {
continue
}
got, err := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
require.NoError(t, err)
assert.Equal(t, body, string(got), rel)
}
for _, rel := range []string{"wiki/w/facts/ok.md", "knowledge/k.md"} {
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: rel, Content: "new body"})
require.NoError(t, err, "positive control: %s", rel)
got, _ := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
assert.Contains(t, string(got), "new body")
}
}
func TestUpdateNoteDoesNotWriteThroughASymlinkedNote(t *testing.T) {
brain := t.TempDir()
facts := filepath.Join(brain, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
victim := filepath.Join(brain, "raw", "victim.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), filepath.Join(facts, "l.md")))
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/facts/l.md", Content: "replaced"})
assert.Error(t, err)
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched\n", string(got))
}
// The note is swapped between a regular file and an in-brain symlink to a
// raw/ file while UpdateNote runs: the read and the write each refuse a
// symlink, so the raw/ file is never rewritten.
func TestUpdateNoteNeverRewritesASymlinkSwappedInForTheNote(t *testing.T) {
brain := t.TempDir()
facts := filepath.Join(brain, "wiki", "w", "facts")
require.NoError(t, os.MkdirAll(facts, 0o755))
victim := filepath.Join(brain, "raw", "victim.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
note := filepath.Join(facts, "n.md")
body := "---\ntitle: n\n---\nold\n"
require.NoError(t, os.WriteFile(note, []byte(body), 0o644))
var stop atomic.Bool
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
tmp := note + ".swap"
for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp)
if i%2 == 0 {
_ = os.WriteFile(tmp, []byte(body), 0o644)
} else {
_ = os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), tmp)
}
_ = os.Rename(tmp, note)
}
}()
ok := 0
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
if _, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/facts/n.md", Content: "new"}); err == nil {
ok++
}
}
stop.Store(true)
wg.Wait()
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched\n", string(got))
assert.Positive(t, ok, "positive control: some updates reach the regular note")
}
+20 -54
View File
@@ -4,12 +4,9 @@ import (
"bufio"
"fmt"
"os"
"path"
"path/filepath"
"strings"
"time"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
)
// seeAlsoHeader is the markdown heading used to group cross-wing links.
@@ -200,12 +197,6 @@ func fileExists(p string) bool {
// section headers. Rejects same-wing pairs (a tunnel is by definition
// cross-wing) and missing notes.
func WriteTunnel(brainDir, sourcePath, targetPath string) error {
if err := checkNotePath(sourcePath); err != nil {
return fmt.Errorf("source: %w", err)
}
if err := checkNotePath(targetPath); err != nil {
return fmt.Errorf("target: %w", err)
}
srcWing, err := wingOf(sourcePath)
if err != nil {
return fmt.Errorf("source: %w", err)
@@ -218,50 +209,20 @@ func WriteTunnel(brainDir, sourcePath, targetPath string) error {
return fmt.Errorf("tunnel must cross wings; got both in %q", srcWing)
}
// Both notes are read and rewritten through the brain root, regular
// files only: brain_tunnel passes caller-supplied paths, and the brain is
// a Syncthing folder that can hold symlinks.
bfs, err := brainfs.Open(brainDir)
if err != nil {
return fmt.Errorf("open brain: %w", err)
}
defer func() { _ = bfs.Close() }()
srcContent, err := bfs.ReadRegular(sourcePath)
if err != nil {
srcFull := filepath.Join(brainDir, filepath.FromSlash(sourcePath))
tgtFull := filepath.Join(brainDir, filepath.FromSlash(targetPath))
if _, err := os.Stat(srcFull); err != nil {
return fmt.Errorf("source note: %w", err)
}
tgtContent, err := bfs.ReadRegular(targetPath)
if err != nil {
if _, err := os.Stat(tgtFull); err != nil {
return fmt.Errorf("target note: %w", err)
}
if out, changed, err := withSeeAlso(srcContent, wikilinkOf(targetPath)); err != nil {
if err := appendSeeAlso(srcFull, wikilinkOf(targetPath)); err != nil {
return fmt.Errorf("update source: %w", err)
} else if changed {
if err := bfs.WriteRegular(sourcePath, out); err != nil {
return fmt.Errorf("update source: %w", err)
}
}
if out, changed, err := withSeeAlso(tgtContent, wikilinkOf(sourcePath)); err != nil {
if err := appendSeeAlso(tgtFull, wikilinkOf(sourcePath)); err != nil {
return fmt.Errorf("update target: %w", err)
} else if changed {
if err := bfs.WriteRegular(targetPath, out); err != nil {
return fmt.Errorf("update target: %w", err)
}
}
return nil
}
// checkNotePath accepts only a clean, local wiki/<wing>/<hall>/.../<slug>.md
// path: no "." or ".." segments, no empty segments, no leading slash.
func checkNotePath(p string) error {
if p == "" || path.Clean(p) != p || !filepath.IsLocal(p) || !strings.HasSuffix(p, ".md") {
return fmt.Errorf("not a clean wiki note path: %q", p)
}
for _, part := range strings.Split(p, "/") {
if part == "" || part == "." || part == ".." {
return fmt.Errorf("not a clean wiki note path: %q", p)
}
}
return nil
}
@@ -287,13 +248,16 @@ func wingOf(relPath string) (string, error) {
return parts[1], nil
}
// withSeeAlso returns content with `- [[link]]` under its See also section,
// creating the section if absent. changed is false when the link is already
// present.
func withSeeAlso(content []byte, link string) (out []byte, changed bool, err error) {
// appendSeeAlso inserts `- [[link]]` under the file's See also section,
// creating the section if absent. No-op when the link is already present.
func appendSeeAlso(filePath, link string) error {
content, err := os.ReadFile(filePath)
if err != nil {
return err
}
wikilink := "[[" + link + "]]"
if strings.Contains(string(content), wikilink) {
return content, false, nil
return nil
}
bullet := "- " + wikilink
@@ -302,7 +266,8 @@ func withSeeAlso(content []byte, link string) (out []byte, changed bool, err err
// No section yet — append a fresh one. Always emit a trailing
// newline so subsequent appends don't merge into the previous line.
trimmed := strings.TrimRight(string(content), "\n")
return []byte(trimmed + "\n\n" + seeAlsoHeader + "\n\n" + bullet + "\n"), true, nil
out := trimmed + "\n\n" + seeAlsoHeader + "\n\n" + bullet + "\n"
return os.WriteFile(filePath, []byte(out), 0o644)
}
// Section exists — splice the bullet in just before the next `## `
@@ -328,11 +293,12 @@ func withSeeAlso(content []byte, link string) (out []byte, changed bool, err err
b.WriteByte('\n')
}
if err := scanner.Err(); err != nil {
return nil, false, err
return err
}
if !inserted {
// section was the last thing in the file — just append bullet
return []byte(strings.TrimRight(b.String(), "\n") + "\n" + bullet + "\n"), true, nil
out := strings.TrimRight(b.String(), "\n") + "\n" + bullet + "\n"
return os.WriteFile(filePath, []byte(out), 0o644)
}
return []byte(b.String()), true, nil
return os.WriteFile(filePath, []byte(b.String()), 0o644)
}
@@ -1,38 +0,0 @@
package brain_test
import (
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
)
// hyperguild#173: brain_tunnel's write followed a directory symlink inside
// wiki/ that points elsewhere in the brain.
func TestWriteTunnelRefusesATargetBehindASymlinkedDirectory(t *testing.T) {
b := t.TempDir()
src := filepath.Join(b, "wiki", "a", "facts", "s.md")
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
require.NoError(t, os.WriteFile(src, []byte("x\n"), 0o644))
victim := filepath.Join(b, "raw", "facts", "v.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(b, "wiki", "b"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "facts"), filepath.Join(b, "wiki", "b", "facts")))
assert.ErrorIs(t, brain.WriteTunnel(b, "wiki/a/facts/s.md", "wiki/b/facts/v.md"), brainfs.ErrNotRegular)
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched\n", string(got))
got, _ = os.ReadFile(src)
assert.Equal(t, "x\n", string(got), "the source is not rewritten either")
real := filepath.Join(b, "wiki", "c", "facts", "t.md")
require.NoError(t, os.MkdirAll(filepath.Dir(real), 0o755))
require.NoError(t, os.WriteFile(real, []byte("t\n"), 0o644))
require.NoError(t, brain.WriteTunnel(b, "wiki/a/facts/s.md", "wiki/c/facts/t.md"), "positive control: a tunnel between real directories is written")
}
@@ -1,141 +0,0 @@
package brain_test
import (
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
)
// brain_tunnel passes caller-supplied source and target paths to WriteTunnel,
// which checked only "wiki/<wing>/..." and then joined them onto the brain
// dir: ".." segments resolved outside the brain and appendSeeAlso rewrote the
// file there.
func TestWriteTunnelRefusesATargetThatTraversesOutOfTheBrain(t *testing.T) {
brain0 := t.TempDir()
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
require.NoError(t, os.WriteFile(src, []byte("---\ntitle: S\n---\nbody\n"), 0o644))
victim := filepath.Join(t.TempDir(), "victim.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
up := strings.Repeat("../", strings.Count(filepath.Clean(brain0), "/")+3)
target := "wiki/b/x/" + up + strings.TrimPrefix(victim, "/")
err := brain.WriteTunnel(brain0, "wiki/a/facts/s.md", target)
assert.Error(t, err)
got, rerr := os.ReadFile(victim)
require.NoError(t, rerr)
assert.Equal(t, "untouched\n", string(got), "a file outside the brain was modified")
}
func TestWriteTunnelRefusesUncleanOrNonWikiPaths(t *testing.T) {
brain0 := t.TempDir()
for _, p := range []string{"wiki/a/facts/s.md", "wiki/b/facts/t.md", "wiki/b/facts/t.txt", "raw/r.md"} {
require.NoError(t, os.MkdirAll(filepath.Dir(filepath.Join(brain0, p)), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(brain0, p), []byte("x\n"), 0o644))
}
for _, target := range []string{
"wiki/b/facts/../../../raw/r.md",
"wiki/b/./facts/t.md",
"wiki/b//facts/t.md",
"/wiki/b/facts/t.md",
"wiki/b/facts/t.txt",
"wiki/b/facts/",
} {
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", target), target)
}
for _, p := range []string{"raw/r.md", "wiki/b/facts/t.txt"} {
got, err := os.ReadFile(filepath.Join(brain0, p))
require.NoError(t, err)
assert.Equal(t, "x\n", string(got), p)
}
require.NoError(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md"), "positive control")
got, err := os.ReadFile(filepath.Join(brain0, "wiki", "b", "facts", "t.md"))
require.NoError(t, err)
assert.Contains(t, string(got), "[[a/facts/s]]")
}
func TestWriteTunnelDoesNotWriteThroughASymlinkedNote(t *testing.T) {
brain0 := t.TempDir()
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
require.NoError(t, os.WriteFile(src, []byte("x\n"), 0o644))
tdir := filepath.Join(brain0, "wiki", "b", "facts")
require.NoError(t, os.MkdirAll(tdir, 0o755))
victim := filepath.Join(t.TempDir(), "victim.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
require.NoError(t, os.Symlink(victim, filepath.Join(tdir, "t.md")))
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md"))
got, err := os.ReadFile(victim)
require.NoError(t, err)
assert.Equal(t, "untouched\n", string(got))
// The same with a link to a file elsewhere in the brain, which os.Root
// would follow.
inVictim := filepath.Join(brain0, "raw", "v.md")
require.NoError(t, os.MkdirAll(filepath.Dir(inVictim), 0o755))
require.NoError(t, os.WriteFile(inVictim, []byte("untouched\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "v.md"), filepath.Join(tdir, "u.md")))
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/u.md"))
got, err = os.ReadFile(inVictim)
require.NoError(t, err)
assert.Equal(t, "untouched\n", string(got))
got, err = os.ReadFile(src)
require.NoError(t, err)
assert.Equal(t, "x\n", string(got), "the source is not touched when the target is refused")
}
// The target is swapped between a regular note and a symlink to an in-brain
// file outside wiki/ while WriteTunnel runs: the read and the write each
// refuse a symlink, so the other file is never rewritten.
func TestWriteTunnelNeverRewritesASymlinkSwappedInForTheTarget(t *testing.T) {
brain0 := t.TempDir()
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
require.NoError(t, os.WriteFile(src, []byte("x\n"), 0o644))
tdir := filepath.Join(brain0, "wiki", "b", "facts")
require.NoError(t, os.MkdirAll(tdir, 0o755))
victim := filepath.Join(brain0, "raw", "victim.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
tgt := filepath.Join(tdir, "t.md")
require.NoError(t, os.WriteFile(tgt, []byte("t\n"), 0o644))
var stop atomic.Bool
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
tmp := tgt + ".swap"
for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp)
if i%2 == 0 {
_ = os.WriteFile(tmp, []byte("t\n"), 0o644)
} else {
_ = os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), tmp)
}
_ = os.Rename(tmp, tgt)
}
}()
ok := 0
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
if brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md") == nil {
ok++
}
_ = os.WriteFile(src, []byte("x\n"), 0o644)
}
stop.Store(true)
wg.Wait()
got, err := os.ReadFile(victim)
require.NoError(t, err)
assert.Equal(t, "untouched\n", string(got), "the swapped-in symlink's target was rewritten")
assert.Positive(t, ok, "positive control: some tunnels were written to the regular target")
}
-71
View File
@@ -104,17 +104,6 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) {
return names, nil
}
// cleanRel refuses a path that is not clean and local. realDir checks the
// cleaned parent, so the I/O must use that same path: os.Root resolves ".."
// after following a symlink, and a ".." after a symlinked directory would
// otherwise pass the check and redirect the read or write.
func cleanRel(rel string) error {
if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) || strings.Contains(rel, "\\") {
return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
}
return nil
}
// ReadDirDirs lists the real subdirectories directly in a brain-relative
// directory, by name. Every component of relDir must be a real directory
// (ErrNotRegular otherwise); symlinks to directories are left out.
@@ -139,12 +128,6 @@ func (b *Brain) ReadDirDirs(relDir string) ([]string, error) {
// ReadRegular reads a brain-relative path, refusing anything that is not a
// regular file, both before the open and on the opened handle.
func (b *Brain) ReadRegular(rel string) ([]byte, error) {
if err := cleanRel(rel); err != nil {
return nil, err
}
if err := b.realDir(path.Dir(rel)); err != nil {
return nil, err
}
fi, err := b.root.Lstat(rel)
if err != nil {
return nil, err
@@ -171,12 +154,6 @@ func (b *Brain) readOpened(rel string) ([]byte, error) { return b.readOpenedMax(
// ReadRegularMax is ReadRegular with a size limit, checked on the opened
// handle before reading (max < 0 means no limit).
func (b *Brain) ReadRegularMax(rel string, max int64) ([]byte, error) {
if err := cleanRel(rel); err != nil {
return nil, err
}
if err := b.realDir(path.Dir(rel)); err != nil {
return nil, err
}
fi, err := b.root.Lstat(rel)
if err != nil {
return nil, err
@@ -272,51 +249,3 @@ func (b *Brain) WalkRegular(relDir string, fn func(rel string) error, onErr func
return fn(p)
})
}
// WriteRegular replaces the contents of an existing regular file at a
// brain-relative path. It never creates a file and never writes through a
// final symlink: os.Root.OpenFile follows an in-root final symlink even with
// O_NOFOLLOW (verified on go1.27), so the path is Lstat-ed through the root
// first and the opened handle must be that same file (os.SameFile), which
// also catches a swap between the two. O_NONBLOCK keeps a FIFO from blocking
// the open. Every directory above the file must be a real directory
// (hyperguild#173); a directory swapped in after that check is the #167
// residual.
func (b *Brain) WriteRegular(rel string, data []byte) error {
if err := cleanRel(rel); err != nil {
return err
}
// Every directory above the file must be real: os.Root follows an
// in-brain intermediate symlink, which would redirect the write
// elsewhere in the brain (hyperguild#173). A swap after this check is
// the #167 residual.
if err := b.realDir(path.Dir(rel)); err != nil {
return err
}
lfi, err := b.root.Lstat(rel)
if err != nil {
return err
}
if !lfi.Mode().IsRegular() {
return fmt.Errorf("%s: %w", rel, ErrNotRegular)
}
f, err := b.root.OpenFile(rel, os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
if err != nil {
return err
}
defer func() { _ = f.Close() }()
fi, err := f.Stat()
if err != nil {
return err
}
if !fi.Mode().IsRegular() || !os.SameFile(lfi, fi) {
return fmt.Errorf("%s: changed while opening: %w", rel, ErrNotRegular)
}
if err := f.Truncate(0); err != nil {
return err
}
if _, err := f.Write(data); err != nil {
return err
}
return f.Close()
}
@@ -263,52 +263,6 @@ func TestReadRegularNeverReturnsAnInBrainFileSwappedInBehindASymlink(t *testing.
assert.Positive(t, ok, "positive control: the regular file is read")
}
func TestWriteRegularRefusesFinalSymlinksFIFOsEscapesAndMissingFilesAndRewritesRegularFiles(t *testing.T) {
dir := t.TempDir()
outside := filepath.Join(t.TempDir(), "o.md")
require.NoError(t, os.WriteFile(outside, []byte("untouched"), 0o644))
require.NoError(t, os.Symlink(outside, filepath.Join(dir, "l.md")))
require.NoError(t, syscall.Mkfifo(filepath.Join(dir, "f.md"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(dir, "n.md"), []byte("old content"), 0o644))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
assert.Error(t, b.WriteRegular("l.md", []byte("x")))
got, _ := os.ReadFile(outside)
assert.Equal(t, "untouched", string(got))
// A symlink to another file inside the brain: os.Root would allow it,
// only O_NOFOLLOW refuses it.
require.NoError(t, os.WriteFile(filepath.Join(dir, "in.md"), []byte("inside"), 0o644))
require.NoError(t, os.Symlink("in.md", filepath.Join(dir, "il.md")))
assert.Error(t, b.WriteRegular("il.md", []byte("x")))
got, _ = os.ReadFile(filepath.Join(dir, "in.md"))
assert.Equal(t, "inside", string(got))
// A reader held open, so a non-blocking write open would succeed. The
// Lstat type check refuses the FIFO first; the handle check and
// O_NONBLOCK matter only if it is swapped in after the Lstat.
r, err := os.OpenFile(filepath.Join(dir, "f.md"), os.O_RDONLY|syscall.O_NONBLOCK, 0)
require.NoError(t, err)
defer func() { _ = r.Close() }()
done := make(chan error, 1)
go func() { done <- b.WriteRegular("f.md", []byte("x")) }()
select {
case err := <-done:
assert.Error(t, err)
case <-time.After(3 * time.Second):
t.Fatal("WriteRegular hung on a FIFO")
}
assert.Error(t, b.WriteRegular("../"+filepath.Base(filepath.Dir(outside))+"/o.md", []byte("x")))
assert.Error(t, b.WriteRegular("missing.md", []byte("x")), "does not create files")
require.NoError(t, b.WriteRegular("n.md", []byte("new")))
got, _ = os.ReadFile(filepath.Join(dir, "n.md"))
assert.Equal(t, "new", string(got), "truncated and rewritten")
}
func TestReadDirDirsListsOnlyRealSubdirsAndRefusesSymlinkedComponents(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "real"), 0o755))
@@ -326,53 +280,3 @@ func TestReadDirDirsListsOnlyRealSubdirsAndRefusesSymlinkedComponents(t *testing
_, err = b.ReadDirDirs("alias")
assert.ErrorIs(t, err, ErrNotRegular)
}
// hyperguild#173: a file reached through a symlinked directory inside the
// brain is refused for reads and writes, so a caller-supplied path cannot be
// redirected to another part of the brain.
func TestReadAndWriteRegularRefuseAFileBehindASymlinkedDirectory(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "v.md"), []byte("untouched"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(dir, "wiki", "alias")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
_, err = b.ReadRegular("wiki/alias/v.md")
assert.ErrorIs(t, err, ErrNotRegular)
_, err = b.ReadRegularMax("wiki/alias/v.md", 100)
assert.ErrorIs(t, err, ErrNotRegular)
assert.ErrorIs(t, b.WriteRegular("wiki/alias/v.md", []byte("x")), ErrNotRegular)
got, _ := os.ReadFile(filepath.Join(dir, "raw", "v.md"))
assert.Equal(t, "untouched", string(got))
got, err = b.ReadRegular("raw/v.md")
require.NoError(t, err, "positive control: the real path still reads")
assert.Equal(t, "untouched", string(got))
}
// realDir checks the cleaned parent, so the I/O must use that same clean
// path: os.Root resolves ".." after following a symlink, and a ".." placed
// after a symlinked directory would otherwise pass the check and redirect.
func TestReadAndWriteRegularRefuseAPathThatIsNotClean(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw", "facts"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "v.md"), []byte("untouched"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "raw", "facts"), filepath.Join(dir, "wiki", "alias")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
for _, rel := range []string{"wiki/alias/../v.md", "./raw/v.md", "raw//v.md", "/raw/v.md"} {
got, err := b.ReadRegular(rel)
assert.ErrorIs(t, err, ErrNotRegular, rel)
assert.NotContains(t, string(got), "untouched", rel)
_, err = b.ReadRegularMax(rel, 100)
assert.ErrorIs(t, err, ErrNotRegular, rel)
assert.ErrorIs(t, b.WriteRegular(rel, []byte("OVERWRITTEN")), ErrNotRegular, rel)
}
got, _ := os.ReadFile(filepath.Join(dir, "raw", "v.md"))
assert.Equal(t, "untouched", string(got))
}
-177
View File
@@ -1,177 +0,0 @@
package session
import (
"errors"
"fmt"
"os"
"path/filepath"
"sync/atomic"
"syscall"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/unix"
)
// session_log passed the caller's session_id straight into
// filepath.Join(sessionsDir, id+".jsonl") and opened it O_CREATE|O_APPEND:
// path segments in the id created or appended files anywhere the pod could
// write. The id is now a plain name, and the open does not follow a final
// symlink.
func TestAppendRefusesSessionIDsThatAreNotPlainNames(t *testing.T) {
root := t.TempDir()
dir := filepath.Join(root, "sessions")
for _, id := range []string{"../x", "a/b", "/abs", ".", "..", ".hidden", "", "a\\b", "a\x00b"} {
assert.Error(t, Append(dir, id, Entry{SessionID: id}), "%q", id)
}
ents, _ := os.ReadDir(root)
for _, e := range ents {
assert.Equal(t, "sessions", e.Name(), "nothing created beside the sessions dir")
}
for _, id := range []string{"1557b2e1-bf0d-5c4f-bbe3-0754d087e356", "_routing", "2026-05-05-infra.plan", "0"} {
require.NoError(t, Append(dir, id, Entry{SessionID: id}), "positive control: %q", id)
}
}
func TestAppendDoesNotWriteThroughASymlinkedLog(t *testing.T) {
root := t.TempDir()
dir := filepath.Join(root, "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
victim := filepath.Join(root, "victim.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("..", "victim.md"), filepath.Join(dir, "s.jsonl")))
assert.Error(t, Append(dir, "s", Entry{SessionID: "s"}))
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched\n", string(got))
}
func TestReadRefusesSessionIDsThatAreNotPlainNames(t *testing.T) {
root := t.TempDir()
require.NoError(t, os.WriteFile(filepath.Join(root, "x.jsonl"), []byte(`{"session_id":"x"}`+"\n"), 0o644))
_, err := Read(filepath.Join(root, "sessions"), "../x")
assert.Error(t, err)
}
// A name that starts plainly and then climbs out with ".." is exactly what the
// ban on "/" in the id's tail stops: filepath.Join cleans "a/../../x" to a
// path beside the sessions dir without needing sessions/a to exist. The
// assertion is on the refusal itself (ErrInvalidID), not on any error.
func TestAppendRefusesAnIDThatClimbsOutAfterAPlainPrefix(t *testing.T) {
root := t.TempDir()
dir := filepath.Join(root, "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
for _, id := range []string{"a/../../escape", "a/../../../escape", "a/../b"} {
err := Append(dir, id, Entry{SessionID: id})
assert.ErrorIs(t, err, ErrInvalidID, "%q", id)
}
_, err := os.Lstat(filepath.Join(root, "escape.jsonl"))
assert.True(t, errors.Is(err, os.ErrNotExist), "nothing created beside the sessions dir")
_, err = os.Lstat(filepath.Join(dir, "b.jsonl"))
assert.True(t, errors.Is(err, os.ErrNotExist), "no in-dir alias created either")
}
// Append does no Lstat type check, so O_NONBLOCK is the only thing that keeps
// a FIFO named <id>.jsonl with no reader from blocking session_log forever.
func TestAppendReturnsPromptlyOnAFIFOLogWithNoReader(t *testing.T) {
dir := filepath.Join(t.TempDir(), "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
fifo := filepath.Join(dir, "s.jsonl")
require.NoError(t, syscall.Mkfifo(fifo, 0o644))
done := make(chan error, 1)
go func() { done <- Append(dir, "s", Entry{SessionID: "s"}) }()
select {
case err := <-done:
assert.Error(t, err, "a FIFO with no reader is refused, not written")
case <-time.After(3 * time.Second):
// Unblock the stuck open so the goroutine and TempDir cleanup finish.
if r, err := os.OpenFile(fifo, os.O_RDONLY|syscall.O_NONBLOCK, 0); err == nil {
defer r.Close() //nolint:errcheck
}
<-done
t.Fatal("Append blocked on a FIFO log with no reader")
}
}
// hyperguild#173: session_log followed a symlinked sessions dir.
func TestAppendRefusesASymlinkedSessionsDir(t *testing.T) {
root := t.TempDir()
elsewhere := filepath.Join(root, "elsewhere")
require.NoError(t, os.MkdirAll(elsewhere, 0o755))
dir := filepath.Join(root, "sessions")
require.NoError(t, os.Symlink("elsewhere", dir))
assert.Error(t, Append(dir, "s", Entry{SessionID: "s"}))
ents, _ := os.ReadDir(elsewhere)
assert.Empty(t, ents, "a log was created through the symlinked dir")
}
// The sessions dir is swapped (RENAME_EXCHANGE) with a symlink to a
// directory outside the brain while Append runs. Append opens through an
// os.Root on the sessions dir's parent, so a swap can redirect it only
// inside that parent (the #167 residual), never outside it.
func TestAppendNeverCreatesALogOutsideTheParentUnderADirSwapRace(t *testing.T) {
brain := t.TempDir()
outside := t.TempDir()
dir := filepath.Join(brain, "sessions")
other := filepath.Join(brain, "sessions.link")
require.NoError(t, os.MkdirAll(dir, 0o755))
require.NoError(t, os.Symlink(outside, other))
var stop atomic.Bool
done := make(chan struct{})
go func() {
defer close(done)
for !stop.Load() {
_ = unix.Renameat2(unix.AT_FDCWD, dir, unix.AT_FDCWD, other, unix.RENAME_EXCHANGE)
}
}()
ok := 0
for i := 0; i < 20000; i++ {
if Append(dir, fmt.Sprintf("s%d", i), Entry{SessionID: "s"}) == nil {
ok++
}
}
stop.Store(true)
<-done
ents, _ := os.ReadDir(outside)
assert.Empty(t, ents, "logs were created outside the brain")
assert.Positive(t, ok, "positive control: appends to the real dir succeed")
}
// The log file is swapped between a regular file and a symlink to another
// file in the brain while Append runs: Append writes only after confirming
// the path is the regular file it opened, so the other file never grows.
func TestAppendNeverWritesThroughALogSwappedForASymlink(t *testing.T) {
brain := t.TempDir()
dir := filepath.Join(brain, "sessions")
require.NoError(t, os.MkdirAll(dir, 0o755))
victim := filepath.Join(brain, "victim.md")
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
log := filepath.Join(dir, "s.jsonl")
var stop atomic.Bool
done := make(chan struct{})
go func() {
defer close(done)
tmp := log + ".swap"
for i := 0; !stop.Load(); i++ {
_ = os.Remove(tmp)
if i%2 == 0 {
_ = os.WriteFile(tmp, nil, 0o644)
} else {
_ = os.Symlink(filepath.Join("..", "victim.md"), tmp)
}
_ = os.Rename(tmp, log)
}
}()
ok := 0
for i := 0; i < 20000; i++ {
if Append(dir, "s", Entry{SessionID: "s"}) == nil {
ok++
}
}
stop.Store(true)
<-done
got, _ := os.ReadFile(victim)
assert.Equal(t, "untouched\n", string(got), "a log line was written through the swapped-in symlink")
assert.Positive(t, ok, "positive control: appends to the regular log succeed")
}
+3 -55
View File
@@ -9,8 +9,6 @@ import (
"io/fs"
"os"
"path/filepath"
"regexp"
"syscall"
"time"
)
@@ -44,63 +42,16 @@ type Attempt struct {
RunnerOutput string `json:"runner_output,omitempty"`
}
// idPattern is a plain file name: session_log takes the id from the caller,
// and a path segment in it created or appended files anywhere the pod could
// write (hyperguild#169).
var idPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$`)
// ErrInvalidID is returned when a session id is not a plain file name.
var ErrInvalidID = errors.New("session id is not a plain name")
func checkID(sessionID string) error {
if !idPattern.MatchString(sessionID) {
return fmt.Errorf("%w: %q", ErrInvalidID, sessionID)
}
return nil
}
// Append writes entry as a single JSON line to sessionsDir/{sessionID}.jsonl.
// The open does not follow a final symlink.
func Append(sessionsDir, sessionID string, entry Entry) error {
if err := checkID(sessionID); err != nil {
return err
}
// Opened through an os.Root on the sessions dir's parent (the brain dir),
// so a sessions dir swapped for a symlink after the check below can
// redirect the log only inside that parent, never outside it (#173; the
// in-brain swap is the #167 residual).
parentDir, name := filepath.Dir(sessionsDir), filepath.Base(sessionsDir)
if err := os.MkdirAll(parentDir, 0o755); err != nil {
return fmt.Errorf("create sessions parent: %w", err)
}
root, err := os.OpenRoot(parentDir)
if err != nil {
return fmt.Errorf("open sessions parent: %w", err)
}
defer func() { _ = root.Close() }()
if err := root.Mkdir(name, 0o755); err != nil && !errors.Is(err, fs.ErrExist) {
if err := os.MkdirAll(sessionsDir, 0o755); err != nil {
return fmt.Errorf("create sessions dir: %w", err)
}
// The sessions dir itself must be a real directory, not a symlink that
// redirects the log elsewhere (hyperguild#173).
if fi, err := root.Lstat(name); err != nil {
return fmt.Errorf("stat sessions dir: %w", err)
} else if !fi.IsDir() {
return fmt.Errorf("sessions dir %s is not a real directory", sessionsDir)
}
logRel := name + "/" + sessionID + ".jsonl"
f, err := root.OpenFile(logRel, os.O_CREATE|os.O_APPEND|os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0o644)
path := filepath.Join(sessionsDir, sessionID+".jsonl")
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("open session log: %w", err)
}
// os.Root follows an in-root final symlink even with O_NOFOLLOW, so
// confirm, before writing, that the path is the regular file we opened.
hfi, herr := f.Stat()
lfi, lerr := root.Lstat(logRel)
if herr != nil || lerr != nil || !lfi.Mode().IsRegular() || !os.SameFile(lfi, hfi) {
_ = f.Close()
return fmt.Errorf("session log %s is not a regular file", logRel)
}
line, err := json.Marshal(entry)
if err != nil {
@@ -119,9 +70,6 @@ func Append(sessionsDir, sessionID string, entry Entry) error {
// Read returns all entries for sessionID. Returns empty slice if no log exists.
func Read(sessionsDir, sessionID string) ([]Entry, error) {
if err := checkID(sessionID); err != nil {
return nil, err
}
path := filepath.Join(sessionsDir, sessionID+".jsonl")
f, err := os.Open(path)
if errors.Is(err, fs.ErrNotExist) {
@@ -18,14 +18,9 @@ import (
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
)
// shutdownFlushTimeout bounds the final flush when Watch stops.
var shutdownFlushTimeout = 5 * time.Second
// DefaultInterval is used when Watch is given a non-positive interval.
const DefaultInterval = 30 * time.Second
@@ -138,32 +133,13 @@ func Scan(brainDir string) (Result, error) {
// queue until Notify succeeds (see alertQueue): an outage delays them, and
// any that do not fit the bounded queue, or are undeliverable, are counted in
// a summary that is never dropped, so every incident is delivered or counted.
//
// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
// a full queue, blocked one flush for several notify timeouts, and while the
// scan waited on it a planted link stayed readable for several intervals.
// The scan hands alerts over without blocking; any that do not fit are
// counted in the summary. When ctx ends, Watch returns after a final flush
// of what was handed over (bounded by shutdownFlushTimeout); alerts that
// flush cannot deliver are counted in one error log line instead. That
// shutdown path only runs for a caller that cancels ctx: cmd/server passes
// context.Background(), so in production the flush on handoff is what keeps
// alerts from waiting on a restart.
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
if interval <= 0 {
interval = DefaultInterval
}
t := time.NewTicker(interval)
defer t.Stop()
alerts := make(chan string, maxPending)
var overflow atomic.Int64
var wg sync.WaitGroup
wg.Add(1)
go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
// The scan is the only sender: closing alerts tells deliver to drain and
// flush once more, and Watch returns only after it has.
defer func() { close(alerts); wg.Wait() }()
q := alertSender{alerts: alerts, overflow: &overflow}
q := &alertQueue{}
reported := ""
for {
select {
@@ -192,76 +168,7 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
". Readers may follow symlinks out of the brain; fix them.")
}
}
}
}
// alertSender hands alerts to deliver without ever blocking the scan; an
// alert that does not fit is counted in overflow, which deliver adds to the
// summary.
type alertSender struct {
alerts chan<- string
overflow *atomic.Int64
}
func (s alertSender) push(msg string) {
select {
case s.alerts <- msg:
default:
s.overflow.Add(1)
}
}
// deliver owns the alertQueue. It flushes as soon as an alert is handed
// over, and again every interval while any are pending (outage retries).
// When alerts is closed (Watch is stopping) it drains it and flushes once
// more, bounded by shutdownFlushTimeout, so an alert handed over just before
// shutdown is not lost (PR #180 review). Whatever that flush cannot deliver
// is logged as one error with its counts, never dropped silently.
func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
t := time.NewTicker(interval)
defer t.Stop()
q := &alertQueue{}
// drain moves everything handed over into the queue; it reports false
// once alerts is closed.
drain := func() bool {
defer func() { q.suppressed += int(overflow.Swap(0)) }()
for {
select {
case msg, ok := <-alerts:
if !ok {
return false
}
q.push(msg)
default:
return true
}
}
}
for {
select {
case msg, ok := <-alerts:
if ok {
q.push(msg)
ok = drain()
}
if !ok {
q.suppressed += int(overflow.Swap(0))
fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
q.flush(fctx, n)
cancel()
if len(q.items) > 0 || q.suppressed > 0 {
slog.Error("symlink tripwire alerts not delivered at shutdown",
"undelivered", len(q.items), "suppressed", q.suppressed)
}
return
}
q.flush(ctx, n)
case <-t.C:
if !drain() {
continue // the closed channel is handled by the case above
}
q.flush(ctx, n)
}
q.flush(ctx, n)
}
}
@@ -1,12 +1,10 @@
package symlinkguard
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"log/slog"
"path/filepath"
"strconv"
"strings"
@@ -347,19 +345,13 @@ func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
}
// toggleNotifier fails while down is set. Like the real audit.NtfyNotifier
// (http.NewRequestWithContext), it also fails, without recording, when its
// ctx is already done (PR #180 round 2: notifiers that ignored ctx let a
// final flush on the cancelled ctx pass).
// toggleNotifier fails while down is set.
type toggleNotifier struct {
recNotifier
down atomic.Bool
}
func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
if err := ctx.Err(); err != nil {
return err
}
if r.down.Load() {
return errors.New("ntfy down")
}
@@ -504,271 +496,3 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
return false
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
}
// hungNotifier answers every Notify only after latency (or when ctx ends),
// with an outage error: an ntfy that accepts connections but never responds
// until its timeout.
type hungNotifier struct{ latency time.Duration }
func (h hungNotifier) Notify(ctx context.Context, _ string) error {
select {
case <-time.After(h.latency):
case <-ctx.Done():
}
return errors.New("ntfy timeout")
}
func waitGone(t *testing.T, link string) {
t.Helper()
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond)
}
// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
// The notifier here is slower than the interval, as in production (10 s
// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
// a scan that waited on it would keep a link for seconds.
func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) {
const interval = 100 * time.Millisecond
brain := t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond})
for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
link := filepath.Join(brain, "fill"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
}
var worst time.Duration
for i := 0; i < 6; i++ {
link := filepath.Join(brain, "probe"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
planted := time.Now()
waitGone(t, link)
worst = max(worst, time.Since(planted))
}
assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst)
}
// Review of PR #180: delivery waited for its own tick, about one interval
// after the removal. An alert is now flushed as soon as it is handed over.
func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) {
const interval = 400 * time.Millisecond
brain := t.TempDir()
n := &recNotifier{}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, interval, n)
link := filepath.Join(brain, "prompt")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
removed := time.Now()
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond)
assert.Less(t, time.Since(removed), interval/4)
}
// Review of PR #180: on ctx end, deliver returned without draining what the
// scan had handed over, so a removal just before shutdown was neither
// delivered nor counted. Watch now drains and flushes before returning.
func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) {
for i := 0; i < 20; i++ {
brain := t.TempDir()
n := &recNotifier{}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "late")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
cancel()
<-done
require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i)
}
}
// blockingNotifier holds every Notify until its ctx ends, tracking calls in
// flight.
type blockingNotifier struct{ inFlight atomic.Int32 }
func (b *blockingNotifier) Notify(ctx context.Context, _ string) error {
b.inFlight.Add(1)
defer b.inFlight.Add(-1)
<-ctx.Done()
return ctx.Err()
}
// Watch does not return while its delivery goroutine is still notifying.
func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) {
prev := shutdownFlushTimeout
shutdownFlushTimeout = 50 * time.Millisecond
t.Cleanup(func() { shutdownFlushTimeout = prev })
brain := t.TempDir()
n := &blockingNotifier{}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "x")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond)
cancel()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("Watch did not return after cancel")
}
assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned")
}
// slowToggleNotifier fails slowly (a hung ntfy) while down, then records.
type slowToggleNotifier struct {
recNotifier
down atomic.Bool
latency time.Duration
}
func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error {
if r.down.Load() {
select {
case <-time.After(r.latency):
case <-ctx.Done():
}
return errors.New("ntfy timeout")
}
return r.recNotifier.Notify(ctx, m)
}
// Review of PR #180: with a slow notifier the scan's handoff overflows, and
// those alerts exist only in the overflow counter. Every removal is still
// delivered or counted in the summary.
func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) {
brain := t.TempDir()
n := &slowToggleNotifier{latency: 30 * time.Millisecond}
n.down.Store(true)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go Watch(ctx, brain, 2*time.Millisecond, n)
const plants = 40
for i := 0; i < plants; i++ {
link := filepath.Join(brain, "ov"+strconv.Itoa(i))
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
}
n.down.Store(false)
delivered, counted := 0, 0
require.Eventually(t, func() bool {
n.mu.Lock()
defer n.mu.Unlock()
delivered, counted = 0, 0
for _, m := range n.msgs {
if strings.Contains(m, "removed 1 symlink(s)") {
delivered++
}
if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
var c int
_, _ = fmt.Sscanf(m[i+3:], "%d", &c)
counted += c
}
}
return delivered+counted == plants
}, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants)
}
// An alert still queued when Watch stops (its send failed, the notifier has
// since recovered, the retry tick has not come) is flushed once more before
// Watch returns, on a ctx that is not the cancelled one: toggleNotifier fails
// on a done ctx, as the real ntfy notifier does.
func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(t *testing.T) {
brain := t.TempDir()
n := &toggleNotifier{}
n.down.Store(true)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }()
link := filepath.Join(brain, "queued")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
time.Sleep(20 * time.Millisecond) // let the immediate send fail
n.down.Store(false)
cancel()
<-done
assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown")
}
// lockedBuffer is a goroutine-safe log sink: Watch goroutines of earlier
// tests may still log while one test captures.
type lockedBuffer struct {
mu sync.Mutex
buf bytes.Buffer
}
func (b *lockedBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.buf.String() }
func captureLog(t *testing.T) *lockedBuffer {
t.Helper()
b := &lockedBuffer{}
prev := slog.Default()
slog.SetDefault(slog.New(slog.NewTextHandler(b, nil)))
t.Cleanup(func() { slog.SetDefault(prev) })
return b
}
// PR #180 round 2: when the final flush fails (ntfy hung at shutdown) the
// undelivered alerts vanished with only "will retry" warnings. Watch now logs
// one error with both counts.
func TestWatchLogsUndeliveredAndSuppressedCountsWhenTheFinalFlushFails(t *testing.T) {
prev := shutdownFlushTimeout
shutdownFlushTimeout = 50 * time.Millisecond
t.Cleanup(func() { shutdownFlushTimeout = prev })
logs := captureLog(t)
brain := t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { Watch(ctx, brain, 5*time.Millisecond, hungNotifier{latency: time.Hour}); close(done) }()
link := filepath.Join(brain, "lost")
require.NoError(t, os.Symlink(t.TempDir(), link))
waitGone(t, link)
cancel()
<-done
out := logs.String()
assert.Contains(t, out, "level=ERROR")
assert.Contains(t, out, `msg="symlink tripwire alerts not delivered at shutdown"`)
assert.Contains(t, out, "undelivered=1")
assert.Contains(t, out, "suppressed=0")
}
// deliver, given a handoff that already holds alerts and is closed, moves them
// into the queue and runs the final flush, even though the ctx is cancelled:
// the drain-on-close path, without depending on when the close lands.
func TestDeliverFlushesAlertsDrainedFromAClosedHandoffOnACancelledCtx(t *testing.T) {
alerts := make(chan string, 3)
alerts <- "a1"
alerts <- "a2"
close(alerts)
ctx, cancel := context.WithCancel(context.Background())
cancel()
n := &toggleNotifier{}
var overflow atomic.Int64
deliver(ctx, time.Hour, n, alerts, &overflow)
assert.Equal(t, []string{"a1", "a2"}, n.msgs)
}
// PR #180 round 2: when the first receive after close finds the handoff empty,
// overflow counted by the scan was never moved into the summary.
func TestDeliverCountsOverflowWhenTheHandoffIsClosedAndEmpty(t *testing.T) {
alerts := make(chan string)
close(alerts)
n := &recNotifier{}
var overflow atomic.Int64
overflow.Store(3)
deliver(context.Background(), time.Hour, n, alerts, &overflow)
require.Len(t, n.msgs, 1)
assert.Contains(t, n.msgs[0], "3 further alert(s)")
}