hyperguild/ingestion/internal/symlinkguard/symlinkguard.go

// 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"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
	"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

// Alert bounds (PR #160 review): the planter chooses the paths, so neither an
// alert's size nor the queue may grow with what was planted.
const (
	maxPathsPerAlert = 10 // further paths are counted, not listed
	maxPending       = 5  // queued alerts; the rest become one summary
)

// 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
	// Special holds FIFOs, sockets and devices (hyperguild#179): a reader
	// that opens one by path can hang, and a FIFO at classification.yaml
	// stops startup.
	Special []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 && len(r.Special) == 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 := r.quoted
	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.Special) > 0 {
		parts = append(parts, fmt.Sprintf("%d special file(s) (FIFO, socket or device; delete them): %s", len(r.Special), q(r.Special)))
	}
	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, "; ")
}

// quoted joins at most maxPathsPerAlert paths, each strconv-quoted, and
// counts the rest.
func (Result) quoted(xs []string) string {
	shown := xs
	if len(shown) > maxPathsPerAlert {
		shown = shown[:maxPathsPerAlert]
	}
	out := make([]string, len(shown))
	for i, x := range shown {
		out[i] = strconv.Quote(x)
	}
	s := strings.Join(out, ", ")
	if extra := len(xs) - len(shown); extra > 0 {
		s += fmt.Sprintf(" and %d more", extra)
	}
	return s
}

// 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
		}
		switch {
		case d.Type()&os.ModeSymlink != 0:
			r.Symlinks = append(r.Symlinks, rel)
		case !d.IsDir() && !d.Type().IsRegular():
			r.Special = append(r.Special, rel)
		}
		return nil
	})
	sort.Strings(r.Symlinks)
	sort.Strings(r.Special)
	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. Every symlink found is removed (the link, never its target:
// hyperguild#157 decision), so one is readable for at most one interval.
// 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 (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}
	reported := ""
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		r, err := Scan(brainDir)
		key := ""
		if err != nil {
			key = "cannot scan the brain dir: " + err.Error()
		} else {
			removed, failed := removeSymlinks(brainDir, r.Symlinks)
			if len(removed) > 0 {
				slog.Error("brain symlink tripwire removed symlinks", "count", len(removed))
				q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
					len(removed), Result{}.quoted(removed)))
			}
			sRemoved, sFailed := removeSpecial(brainDir, r.Special)
			if len(sRemoved) > 0 {
				slog.Error("brain symlink tripwire removed special files", "count", len(sRemoved))
				q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#179): removed %d special file(s) (FIFO, socket or device) from the brain dir: %s",
					len(sRemoved), Result{}.quoted(sRemoved)))
			}
			key = Result{Symlinks: failed, Unreadable: r.Unreadable, Special: sFailed}.String()
		}
		if key != reported {
			reported = key
			if key != "" {
				slog.Error("brain symlink tripwire", "finding", key)
				q.push("brain symlink tripwire (hyperguild#157, #179): " + key +
					". Symlinks can lead readers out of the brain and special files can hang them; remove 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)
		}
	}
}

// alertQueue holds alerts until Notify accepts them. It is bounded (at most
// maxPending queued) and loses nothing from the record: every overflow and
// every dropped message increments suppressed, sent as a summary that is
// retried every tick and never dropped.
//
// A message is dropped only when the notifier rejects it permanently (an
// error with Permanent() == true, e.g. ntfy 413). Every other failure is an
// outage, and nothing is dropped however long it lasts (PR #160 round 2:
// dropping after N raw failures lost the only incident of a long outage).
// Each flush tries every queued message, so one undeliverable message cannot
// block the others.
type alertQueue struct {
	items      []string
	suppressed int
}

func (q *alertQueue) push(msg string) {
	if len(q.items) >= maxPending {
		q.suppressed++
		return
	}
	q.items = append(q.items, msg)
}

func (q *alertQueue) flush(ctx context.Context, n Notifier) {
	if n == nil {
		q.items, q.suppressed = nil, 0
		return
	}
	kept := q.items[:0]
	for _, msg := range q.items {
		err := n.Notify(ctx, msg)
		switch {
		case err == nil:
		case permanent(err):
			slog.Error("symlink tripwire alert rejected permanently; counted in the summary", "err", err)
			q.suppressed++
		default:
			slog.Warn("symlink tripwire notify failed; will retry", "err", err)
			kept = append(kept, msg)
		}
	}
	q.items = kept
	if q.suppressed > 0 {
		msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): %d further alert(s) were suppressed or undeliverable; check the ingestion logs", q.suppressed)
		if err := n.Notify(ctx, msg); err == nil {
			q.suppressed = 0
		}
	}
}

// permanent reports whether err says the message itself can never be sent.
func permanent(err error) bool {
	var p interface{ Permanent() bool }
	return errors.As(err, &p) && p.Permanent()
}

// removeSpecial unlinks each brain-relative FIFO, socket or device through
// an os.Root on the brain dir, after re-checking that it is still one: a
// path that is a regular file, a directory or a symlink at the re-check is
// left alone and reported as failed. The re-check narrows, but cannot close,
// the window before the unlink (Linux has no conditional unlink); the same
// window exists in removeSymlinks, and only a local writer can use it.
func removeSpecial(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.IsDir() || fi.Mode().IsRegular() || 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
}

// 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
		}
		// root.Remove, not os.RemoveAll: if the link is swapped for a
		// non-empty directory after the Lstat, Remove refuses rather than
		// deleting it recursively, and it cannot act outside the root.
		if err := root.Remove(rel); err != nil {
			failed = append(failed, rel)
			continue
		}
		removed = append(removed, rel)
	}
	return removed, failed
}