feat(ingestion): runtime symlink tripwire removes what it finds, 30s scans (#157) #160
@@ -21,7 +21,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// DefaultInterval is used when Watch is given a non-positive interval.
|
// DefaultInterval is used when Watch is given a non-positive interval.
|
||||||
const DefaultInterval = 5 * time.Minute
|
const DefaultInterval = 30 * time.Second
|
||||||
|
|
||||||
// Notifier is the out-of-band alert sink (audit.Notifier satisfies it).
|
// Notifier is the out-of-band alert sink (audit.Notifier satisfies it).
|
||||||
type Notifier interface {
|
type Notifier interface {
|
||||||
@@ -41,13 +41,7 @@ func (r Result) Clean() bool { return len(r.Symlinks) == 0 && len(r.Unreadable)
|
|||||||
// String describes the findings for a log line or alert. Paths are quoted, so
|
// String describes the findings for a log line or alert. Paths are quoted, so
|
||||||
// a name containing a newline cannot blur two findings together.
|
// a name containing a newline cannot blur two findings together.
|
||||||
func (r Result) String() string {
|
func (r Result) String() string {
|
||||||
q := func(xs []string) string {
|
q := r.quoted
|
||||||
out := make([]string, len(xs))
|
|
||||||
for i, x := range xs {
|
|
||||||
out[i] = strconv.Quote(x)
|
|
||||||
}
|
|
||||||
return strings.Join(out, ", ")
|
|
||||||
}
|
|
||||||
var parts []string
|
var parts []string
|
||||||
if len(r.Symlinks) > 0 {
|
if len(r.Symlinks) > 0 {
|
||||||
parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
|
parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
|
||||||
@@ -58,6 +52,15 @@ func (r Result) String() string {
|
|||||||
return strings.Join(parts, "; ")
|
return strings.Join(parts, "; ")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// quoted joins paths with each one strconv-quoted.
|
||||||
|
func (Result) quoted(xs []string) string {
|
||||||
|
out := make([]string, len(xs))
|
||||||
|
for i, x := range xs {
|
||||||
|
out[i] = strconv.Quote(x)
|
||||||
|
}
|
||||||
|
return strings.Join(out, ", ")
|
||||||
|
}
|
||||||
|
|
||||||
// Scan walks the brain dir and reports every symlink in it and every
|
// 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
|
// 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
|
// skipped (it is brain-sync's, and no reader walks it); a nested .git is
|
||||||
@@ -106,15 +109,19 @@ func Scan(brainDir string) (Result, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Watch rescans every interval (DefaultInterval if non-positive) until ctx
|
// Watch rescans every interval (DefaultInterval if non-positive) until ctx
|
||||||
// ends, and notifies once per distinct non-clean result. A result counts as
|
// ends. Every symlink found is removed (the link, never its target:
|
||||||
// reported only after Notify succeeds, so an alert lost to a notifier outage
|
// hyperguild#157 decision), so one is readable for at most one interval.
|
||||||
// is retried on the next tick.
|
// Each removal is an alert; a persistent finding (a link it could not remove,
|
||||||
|
// an unreadable dir, a scan error) alerts once per distinct state. Alerts
|
||||||
|
// queue until Notify succeeds, so an outage delays them but loses none, even
|
||||||
|
// though the next scan of a cleaned brain is clean.
|
||||||
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
||||||
if interval <= 0 {
|
if interval <= 0 {
|
||||||
interval = DefaultInterval
|
interval = DefaultInterval
|
||||||
}
|
}
|
||||||
t := time.NewTicker(interval)
|
t := time.NewTicker(interval)
|
||||||
defer t.Stop()
|
defer t.Stop()
|
||||||
|
var pending []string
|
||||||
reported := ""
|
reported := ""
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
@@ -123,31 +130,68 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
|
|||||||
case <-t.C:
|
case <-t.C:
|
||||||
}
|
}
|
||||||
r, err := Scan(brainDir)
|
r, err := Scan(brainDir)
|
||||||
var key, msg string
|
key := ""
|
||||||
switch {
|
if err != nil {
|
||||||
case err != nil:
|
key = "cannot scan the brain dir: " + err.Error()
|
||||||
key = "error: " + err.Error()
|
} else {
|
||||||
msg = "brain symlink tripwire (hyperguild#157): cannot scan the brain dir: " + err.Error()
|
removed, failed := removeSymlinks(brainDir, r.Symlinks)
|
||||||
case r.Clean():
|
if len(removed) > 0 {
|
||||||
reported = ""
|
msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
|
||||||
continue
|
len(removed), Result{Symlinks: removed}.quoted(removed))
|
||||||
default:
|
slog.Error("brain symlink tripwire removed symlinks", "paths", removed)
|
||||||
key = r.String()
|
pending = append(pending, msg)
|
||||||
msg = "brain symlink tripwire (hyperguild#157): " + key +
|
}
|
||||||
". Readers may follow symlinks out of the brain; remove them."
|
key = Result{Symlinks: failed, Unreadable: r.Unreadable}.String()
|
||||||
}
|
}
|
||||||
if key == reported {
|
if key != reported {
|
||||||
continue
|
|
||||||
}
|
|
||||||
slog.Error("brain symlink tripwire", "finding", key)
|
|
||||||
if n == nil {
|
|
||||||
reported = key
|
reported = key
|
||||||
|
if key != "" {
|
||||||
|
slog.Error("brain symlink tripwire", "finding", key)
|
||||||
|
pending = append(pending, "brain symlink tripwire (hyperguild#157): "+key+
|
||||||
|
". Readers may follow symlinks out of the brain; fix them.")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if n == nil {
|
||||||
|
pending = nil
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if err := n.Notify(ctx, msg); err != nil {
|
for len(pending) > 0 {
|
||||||
slog.Warn("symlink tripwire notify failed; will retry", "err", err)
|
if err := n.Notify(ctx, pending[0]); err != nil {
|
||||||
continue
|
slog.Warn("symlink tripwire notify failed; will retry", "err", err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
pending = pending[1:]
|
||||||
}
|
}
|
||||||
reported = key
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// removeSymlinks unlinks each brain-relative symlink through an os.Root on
|
||||||
|
// the brain dir, after re-checking it is still a symlink. Removing a symlink
|
||||||
|
// never touches its target. It returns what was removed and what was not.
|
||||||
|
func removeSymlinks(brainDir string, rels []string) (removed, failed []string) {
|
||||||
|
if len(rels) == 0 {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
resolved, err := filepath.EvalSymlinks(brainDir)
|
||||||
|
if err != nil {
|
||||||
|
return nil, rels
|
||||||
|
}
|
||||||
|
root, err := os.OpenRoot(resolved)
|
||||||
|
if err != nil {
|
||||||
|
return nil, rels
|
||||||
|
}
|
||||||
|
defer func() { _ = root.Close() }()
|
||||||
|
for _, rel := range rels {
|
||||||
|
fi, err := root.Lstat(rel)
|
||||||
|
if err != nil || fi.Mode()&os.ModeSymlink == 0 {
|
||||||
|
failed = append(failed, rel)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := root.Remove(rel); err != nil {
|
||||||
|
failed = append(failed, rel)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
removed = append(removed, rel)
|
||||||
|
}
|
||||||
|
return removed, failed
|
||||||
|
}
|
||||||
|
|||||||
@@ -182,8 +182,47 @@ func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) {
|
|||||||
go Watch(ctx, brain, 10*time.Millisecond, n)
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
||||||
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
|
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
|
||||||
require.NoError(t, os.Remove(link))
|
// The tripwire removed it; planting it again is a new event.
|
||||||
time.Sleep(60 * time.Millisecond)
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 10*time.Millisecond)
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
||||||
require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
|
require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// hyperguild#157 decision: the runtime tripwire removes a symlink it finds
|
||||||
|
// (the link, never its target), so it is readable for at most one interval
|
||||||
|
// instead of until a human acts, and the alert says what was removed.
|
||||||
|
func TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(t *testing.T) {
|
||||||
|
brain := t.TempDir()
|
||||||
|
outside := filepath.Join(t.TempDir(), "target.md")
|
||||||
|
require.NoError(t, os.WriteFile(outside, []byte("keep me"), 0o644))
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
|
||||||
|
link := filepath.Join(brain, "wiki", "leak.md")
|
||||||
|
require.NoError(t, os.Symlink(outside, link))
|
||||||
|
regular := filepath.Join(brain, "wiki", "note.md")
|
||||||
|
require.NoError(t, os.WriteFile(regular, []byte("note"), 0o644))
|
||||||
|
|
||||||
|
n := &recNotifier{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
||||||
|
|
||||||
|
require.Eventually(t, func() bool {
|
||||||
|
_, err := os.Lstat(link)
|
||||||
|
return os.IsNotExist(err)
|
||||||
|
}, 2*time.Second, 10*time.Millisecond, "the symlink is removed")
|
||||||
|
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
|
||||||
|
b, err := os.ReadFile(outside)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, "keep me", string(b), "the target is untouched")
|
||||||
|
b, err = os.ReadFile(regular)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, "note", string(b), "regular files are untouched")
|
||||||
|
n.mu.Lock()
|
||||||
|
defer n.mu.Unlock()
|
||||||
|
assert.Contains(t, n.msgs[0], "removed")
|
||||||
|
assert.Contains(t, n.msgs[0], "wiki/leak.md")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDefaultIntervalIsThirtySeconds(t *testing.T) {
|
||||||
|
assert.Equal(t, 30*time.Second, DefaultInterval)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user