guard/internal/daemon/daemon_test.go

package daemon

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"guard/internal/config"
	"guard/internal/housecarl"
	"guard/internal/source"
	"guard/internal/state"
)

type fakeHC struct {
	mu   sync.Mutex
	bans []housecarl.BanInput
	fail error
}

func (f *fakeHC) Ban(_ context.Context, list string, in housecarl.BanInput) (*housecarl.BanResult, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.fail != nil {
		return nil, f.fail
	}
	f.bans = append(f.bans, in)
	return &housecarl.BanResult{Bans: []housecarl.Ban{{Address: in.Address, List: list}}}, nil
}

func (f *fakeHC) BanList(_ context.Context, list string) (*housecarl.BanList, error) {
	return &housecarl.BanList{Name: list, GroupV6: 64}, nil
}

func (f *fakeHC) setFail(err error) {
	f.mu.Lock()
	f.fail = err
	f.mu.Unlock()
}

func (f *fakeHC) count() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.bans)
}

func (f *fakeHC) last() housecarl.BanInput {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.bans[len(f.bans)-1]
}

type chanSource struct{ ch chan source.Line }

func (c *chanSource) Run(ctx context.Context, out chan<- source.Line) error {
	for {
		select {
		case <-ctx.Done():
			return nil
		case l := <-c.ch:
			out <- l
		}
	}
}

func waitFor(t *testing.T, what string, cond func() bool) {
	t.Helper()
	for i := 0; i < 200; i++ {
		if cond() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("waited in vain: %s", what)
}

// TestDaemon: hits within the window ban, by the watch's ports and time;
// a repeat offender's ban is longer; IPv6 counts by /64; never_ban is
// never banned; bans wait while Housecarl can't be reached, and a
// refused one isn't sent again.
func TestDaemon(t *testing.T) {
	dir := t.TempDir()
	cfg, err := config.Parse([]byte(`housecarl: {url: https://fw:8070, list: web}
never_ban: [198.51.100.0/24]
defaults: {hits: 3, within: 10m, ban: 1h, repeat: {factor: 2, max: 3h}}
watch:
  - {name: ssh, preset: sshd, file: /var/log/auth.log, ban: {protocol: tcp, port: 22}}
state: ` + filepath.Join(dir, "state.json") + `
socket: ` + filepath.Join(dir, "guard.sock") + `
`))
	if err != nil {
		t.Fatal(err)
	}
	st, _ := state.Load(cfg.State)
	hc := &fakeHC{}
	d, err := New(cfg, hc, nil, st, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatal(err)
	}
	d.RetryEvery = 50 * time.Millisecond
	src := &chanSource{ch: make(chan source.Line)}
	d.Watches()[0].Source = src
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { d.Run(ctx); close(done) }()
	fail := func(addr string, n int) {
		for i := 0; i < n; i++ {
			src.ch <- source.Line{Text: "Failed password for root from " + addr + " port 4000 ssh2", Time: time.Now()}
		}
	}
	fail("203.0.113.9", 2)
	time.Sleep(50 * time.Millisecond)
	if hc.count() != 0 {
		t.Fatal("banned below the hits")
	}
	fail("203.0.113.9", 1)
	waitFor(t, "the first ban", func() bool { return hc.count() == 1 })
	if b := hc.last(); b.Address != "203.0.113.9" || b.Protocol != "tcp" || b.Ports != "22" || b.Time < 3590 || b.Time > 3600 || b.Source != "guard/ssh" ||
		!strings.Contains(b.Reason, "Failed password") {
		t.Errorf("ban %+v", b)
	}
	fail("203.0.113.9", 3)
	waitFor(t, "the repeat ban", func() bool { return hc.count() == 2 })
	if b := hc.last(); b.Time < 7190 || b.Time > 7200 {
		t.Errorf("a repeat offender's second ban: %ds, want 2h", b.Time)
	}
	fail("203.0.113.9", 3)
	waitFor(t, "the third ban", func() bool { return hc.count() == 3 })
	if b := hc.last(); b.Time < 10790 || b.Time > 10800 {
		t.Errorf("capped at max: %ds, want 3h", b.Time)
	}
	// IPv6, by /64.
	for _, a := range []string{"2001:db8:1:2::1", "2001:db8:1:2::2", "2001:db8:1:2::3"} {
		fail(a, 1)
	}
	waitFor(t, "the IPv6 ban", func() bool { return hc.count() == 4 })
	// never_ban.
	fail("198.51.100.7", 5)
	time.Sleep(50 * time.Millisecond)
	if hc.count() != 4 {
		t.Errorf("never_ban banned: %+v", hc.last())
	}
	// Housecarl down: the ban waits, then goes.
	hc.setFail(errors.New("connection refused"))
	fail("203.0.113.20", 3)
	time.Sleep(150 * time.Millisecond)
	if _, pending := st.Snapshot(); hc.count() != 4 || len(pending) != 1 {
		t.Errorf("while Housecarl is down: %d bans, %d pending", hc.count(), len(pending))
	}
	hc.setFail(nil)
	waitFor(t, "the ban that waited", func() bool { return hc.count() == 5 && hc.last().Address == "203.0.113.20" })
	// Refused: recorded, not sent again.
	hc.setFail(&housecarl.Error{Status: 422, Code: "never_ban", Message: "203.0.113.30 isn't banned: it is inside"})
	fail("203.0.113.30", 3)
	waitFor(t, "the refusal recorded", func() bool {
		recent, _ := st.Snapshot()
		return len(recent) > 0 && strings.Contains(recent[0].Result, "inside")
	})
	time.Sleep(150 * time.Millisecond)
	if _, pending := st.Snapshot(); len(pending) != 0 {
		t.Errorf("a refused ban waits: %+v", pending)
	}
	s := d.Status()
	if len(s.Watches) != 1 || s.Watches[0].Bans != 6 || s.Watches[0].Matches != 23 {
		t.Errorf("status %+v", s.Watches)
	}
	cancel()
	<-done
	// The state outlives the daemon.
	st2, err := state.Load(cfg.State)
	if err != nil || st2.Repeats("203.0.113.9/32", time.Now().Add(-time.Hour)) != 3 {
		t.Errorf("saved state: %v %+v", err, st2.Offences)
	}
}

// stoppable is a chanSource that says when it stopped.
type stoppable struct {
	chanSource
	stopped chan struct{}
}

func (s *stoppable) Run(ctx context.Context, out chan<- source.Line) error {
	err := s.chanSource.Run(ctx, out)
	close(s.stopped)
	return err
}

// TestReload: a reload keeps the bounties of a watch under the same name
// and applies its new settings; a watch whose source changed gets a new
// one; watches gone stop, new ones start; settings for a restart are
// reported; a configuration with nothing to watch is refused.
func TestReload(t *testing.T) {
	dir := t.TempDir()
	base := `housecarl: {url: https://fw:8070, list: web}
defaults: {hits: 3, within: 10m, ban: 1h}
state: ` + filepath.Join(dir, "state.json") + `
socket: ` + filepath.Join(dir, "guard.sock") + `
`
	parse := func(extra string) *config.Config {
		t.Helper()
		cfg, err := config.Parse([]byte(base + extra))
		if err != nil {
			t.Fatal(err)
		}
		return cfg
	}
	cfg := parse(`watch:
  - {name: ssh, preset: sshd, file: /var/log/auth.log}
  - {name: web, file: /var/log/web.log, match: ['bad <IP>$']}
`)
	st, _ := state.Load(cfg.State)
	hc := &fakeHC{}
	d, err := New(cfg, hc, nil, st, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatal(err)
	}
	var mu sync.Mutex
	srcs := map[string]*stoppable{}
	d.Sources = func(w *config.Watch) source.Source {
		mu.Lock()
		defer mu.Unlock()
		s := &stoppable{chanSource{make(chan source.Line)}, make(chan struct{})}
		srcs[w.Name+" "+w.File] = s
		return s
	}
	src := func(key string) *stoppable {
		mu.Lock()
		defer mu.Unlock()
		return srcs[key]
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { d.Run(ctx); close(done) }()
	waitFor(t, "the sources", func() bool { return src("ssh /var/log/auth.log") != nil && src("web /var/log/web.log") != nil })
	fail := func(s *stoppable, addr string, n int) {
		for i := 0; i < n; i++ {
			s.ch <- source.Line{Text: "Failed password for root from " + addr + " port 4000 ssh2", Time: time.Now()}
		}
	}
	ssh := src("ssh /var/log/auth.log")
	fail(ssh, "203.0.113.9", 2)

	res, err := d.Reload(parse(`never_ban: [198.51.100.0/24]
watch:
  - {name: ssh, preset: sshd, file: /var/log/auth.log, ban: {time: 2h}}
  - {name: mail, preset: dovecot, file: /var/log/mail.log}
`))
	if err != nil {
		t.Fatal(err)
	}
	if got := strings.Join(res.Changes, "; "); got != "never_ban; watch ssh: ban; watch mail: added; watch web: removed" || len(res.Restart) != 0 {
		t.Errorf("reload: %q, restart %v", got, res.Restart)
	}
	select {
	case <-src("web /var/log/web.log").stopped:
	case <-time.After(2 * time.Second):
		t.Error("the removed watch's source still runs")
	}
	if src("mail /var/log/mail.log") == nil {
		t.Error("the added watch has no source")
	}
	// The bounty of 2 was kept: one more bans, for the new time.
	fail(ssh, "203.0.113.9", 1)
	waitFor(t, "the ban after the reload", func() bool { return hc.count() == 1 })
	if b := hc.last(); b.Time < 7190 || b.Time > 7200 {
		t.Errorf("ban after reload: %ds, want 2h", b.Time)
	}
	fail(ssh, "198.51.100.7", 3)

	// A new file: a new source; the bounty stays; the hits change.
	fail(ssh, "203.0.113.50", 2)
	res, err = d.Reload(parse(`log: {level: debug}
watch:
  - {name: ssh, preset: sshd, file: /var/log/secure, hits: 4, ban: {time: 2h}}
`))
	if err != nil {
		t.Fatal(err)
	}
	if got := strings.Join(res.Changes, "; "); got != "never_ban; watch ssh: bounty 4 in 10m, source (started again); watch mail: removed" ||
		strings.Join(res.Restart, ",") != "log" {
		t.Errorf("reload: %q, restart %v", got, res.Restart)
	}
	select {
	case <-ssh.stopped:
	case <-time.After(2 * time.Second):
		t.Error("the old source still runs")
	}
	secure := src("ssh /var/log/secure")
	fail(secure, "203.0.113.50", 1)
	time.Sleep(50 * time.Millisecond)
	if hc.count() != 1 {
		t.Fatal("banned below the new hits")
	}
	fail(secure, "203.0.113.50", 1)
	waitFor(t, "the ban at 4", func() bool { return hc.count() == 2 && hc.last().Address == "203.0.113.50" })
	if d.config().Log.Level != "info" {
		t.Error("the log level changed without a restart")
	}

	// Nothing to watch: refused, the old configuration stays.
	if _, err := d.Reload(parse(`watch:
  - {name: ssh, preset: sshd, disabled: true}
`)); err == nil {
		t.Error("a configuration with every watch disabled was taken")
	}
	if s := d.Status(); len(s.Watches) != 1 || s.Watches[0].Hits != 4 || s.Watches[0].Bans != 2 || s.Watches[0].Lines != 10 {
		t.Errorf("status after reloads %+v", s.Watches)
	}
	cancel()
	<-done
}