guard/internal/daemon/daemon.go

// Package daemon is Guard at work: each watch's source feeds its lines
// to its patterns; an address that matches often enough is banned in
// Housecarl; bans Housecarl can't take now are sent again later.
package daemon

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"math"
	"net"
	"net/http"
	"net/netip"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

// Banner is what bans go to: Housecarl (a fake in tests).
type Banner interface {
	Ban(ctx context.Context, list string, in housecarl.BanInput) (*housecarl.BanResult, error)
	BanList(ctx context.Context, list string) (*housecarl.BanList, error)
}

// Daemon is a running Guard.
type Daemon struct {
	cfg     *config.Config
	log     *slog.Logger
	hc      Banner
	state   *state.State
	never   *NeverBan
	watches []*Watch
	started time.Time

	mu     sync.Mutex
	groups map[string]int // ban list → IPv6 grouping
	// pending are the bans Housecarl hasn't taken yet, by list and group.
	pending map[string]state.Ban
	wake    chan struct{}
	// refused remembers addresses Housecarl refused, so each is logged
	// once an hour.
	refused map[string]time.Time

	// Now and RetryEvery can be changed by tests.
	Now        func() time.Time
	RetryEvery time.Duration
}

// Watch is one watch at work.
type Watch struct {
	Cfg     *config.Watch
	Matcher *match.Matcher
	Source  source.Source
	Tracker *bounty.Tracker

	lines, matches, bans atomic.Uint64
}

// New makes a Daemon. hc may be a fake; sources come from the watches'
// configuration unless a test sets them (Watch.Source).
func New(cfg *config.Config, hc Banner, logs source.LogReader, st *state.State, log *slog.Logger) (*Daemon, error) {
	d := &Daemon{cfg: cfg, log: log, hc: hc, state: st, never: NewNeverBan(cfg.NeverBan, log), groups: map[string]int{},
		pending: map[string]state.Ban{}, wake: make(chan struct{}, 1), refused: map[string]time.Time{},
		Now: time.Now, RetryEvery: 30 * time.Second}
	for _, b := range st.Pending {
		d.pending[b.List+" "+b.Group+" "+b.Protocol+" "+b.Ports] = b
	}
	for _, w := range cfg.Watch {
		if w.Disabled {
			continue
		}
		m, err := match.Compile(w.Match, w.Ignore)
		if err != nil {
			return nil, fmt.Errorf("watch %s: %v", w.Name, err)
		}
		wt := &Watch{Cfg: w, Matcher: m, Tracker: bounty.New(w.Hits, w.Within.D())}
		switch w.Kind() {
		case "file":
			wt.Source = &source.File{Pattern: w.File, Log: log.With("watch", w.Name)}
		case "journal":
			wt.Source = &source.Journal{Units: w.Journal.Unit, Identifiers: w.Journal.Identifier, Log: log.With("watch", w.Name)}
		case config.SourceHousecarlDrops:
			wt.Source = &source.HousecarlDrops{Client: logs, Protocol: w.Protocol, Shadow: w.Shadow, Log: log.With("watch", w.Name)}
		}
		d.watches = append(d.watches, wt)
	}
	if len(d.watches) == 0 {
		return nil, errors.New("every watch is disabled: nothing to do")
	}
	return d, nil
}

// Watches are the watches at work (tests set their sources).
func (d *Daemon) Watches() []*Watch { return d.watches }

// Run watches until ctx ends.
func (d *Daemon) Run(ctx context.Context) error {
	d.started = d.Now()
	d.loadGroups(ctx)
	go d.never.Run(ctx)
	var wg sync.WaitGroup
	for _, w := range d.watches {
		lines := make(chan source.Line, 256)
		wg.Add(2)
		go func() {
			defer wg.Done()
			defer close(lines)
			for {
				err := w.Source.Run(ctx, lines)
				if ctx.Err() != nil {
					return
				}
				d.log.Error("the watch's source stopped; starting it again in 10 seconds", "watch", w.Cfg.Name, "err", err)
				select {
				case <-ctx.Done():
					return
				case <-time.After(10 * time.Second):
				}
			}
		}()
		go func() {
			defer wg.Done()
			for l := range lines {
				d.Line(ctx, w, l)
			}
		}()
		d.log.Info("watching", "watch", w.Cfg.Name, "source", Describe(w.Cfg), "patterns", len(w.Cfg.Match), "hits", w.Cfg.Hits, "within", w.Cfg.Within.String(), "list", w.Cfg.Ban.List)
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		d.sender(ctx)
	}()
	if d.cfg.Socket != "" {
		if err := d.serveStatus(ctx); err != nil {
			d.log.Warn("no status socket (guard status won't work)", "socket", d.cfg.Socket, "err", err)
		}
	}
	<-ctx.Done()
	wg.Wait()
	d.savePending()
	return nil
}

// Describe says where a watch's lines come from.
func Describe(w *config.Watch) string {
	switch w.Kind() {
	case "journal":
		var parts []string
		if len(w.Journal.Unit) > 0 {
			parts = append(parts, "unit "+strings.Join(w.Journal.Unit, ", "))
		}
		if len(w.Journal.Identifier) > 0 {
			parts = append(parts, "identifier "+strings.Join(w.Journal.Identifier, ", "))
		}
		return "journal (" + strings.Join(parts, "; ") + ")"
	case config.SourceHousecarlDrops:
		return "Housecarl's dropped packets (" + w.Protocol + ")"
	}
	return "file " + w.File
}

// Short describes a watch's source in a table's few columns: at most
// two journal units or identifiers, then how many more.
func Short(w *config.Watch) string {
	switch w.Kind() {
	case "journal":
		names := w.Journal.Unit
		if len(names) == 0 {
			names = w.Journal.Identifier
		} else {
			names = append(append([]string(nil), names...), w.Journal.Identifier...)
		}
		s := "journal " + strings.Join(names[:min(len(names), 2)], ", ")
		if len(names) > 2 {
			s += fmt.Sprintf(" +%d", len(names)-2)
		}
		return s
	case config.SourceHousecarlDrops:
		return "Housecarl drops (" + w.Protocol + ")"
	}
	return "file " + w.File
}

// loadGroups asks Housecarl how each ban list groups IPv6 addresses, so
// bounties are counted as the list bans. A list that can't be read now
// counts by /64, its default.
func (d *Daemon) loadGroups(ctx context.Context) {
	for _, w := range d.watches {
		list := w.Cfg.Ban.List
		d.mu.Lock()
		_, done := d.groups[list]
		d.mu.Unlock()
		if done {
			continue
		}
		c, cancel := context.WithTimeout(ctx, 10*time.Second)
		bl, err := d.hc.BanList(c, list)
		cancel()
		group := 64
		switch {
		case err != nil:
			d.log.Warn("can't read the ban list from Housecarl (yet); counting IPv6 addresses by /64", "list", list, "err", err)
		case bl.Disabled:
			d.log.Warn("the ban list is disabled in Housecarl: its bans won't be in force", "list", list)
			group = bl.V6Group()
		default:
			group = bl.V6Group()
		}
		d.mu.Lock()
		d.groups[list] = group
		d.mu.Unlock()
	}
}

func (d *Daemon) group(list string) int {
	d.mu.Lock()
	defer d.mu.Unlock()
	if g, ok := d.groups[list]; ok {
		return g
	}
	return 64
}

// Line handles one line of a watch.
func (d *Daemon) Line(ctx context.Context, w *Watch, l source.Line) {
	w.lines.Add(1)
	addr := l.Addr
	if !addr.IsValid() {
		var ok bool
		if addr, ok = w.Matcher.Match(l.Text); !ok {
			return
		}
	}
	w.matches.Add(1)
	if why := d.never.Why(addr); why != "" {
		d.log.Debug("match, but never banned", "watch", w.Cfg.Name, "address", addr, "why", why)
		return
	}
	g := match.Group(addr, d.group(w.Cfg.Ban.List))
	n, reached := w.Tracker.Add(g, l.Time)
	d.log.Debug("match", "watch", w.Cfg.Name, "address", addr, "bounty", n, "hits", w.Cfg.Hits, "line", clip(l.Text, 300))
	if !reached {
		return
	}
	w.bans.Add(1)
	d.ban(w, addr, g, l.Text)
}

// BanTime is how long a group's next ban lasts: the watch's time,
// times the repeat factor for each earlier ban within forget, up to the
// most.
func (d *Daemon) BanTime(w *config.Watch, group string) time.Duration {
	r := d.cfg.Defaults.Repeat
	base := w.Ban.Time.D()
	n := d.state.Repeats(group, d.Now().Add(-r.Forget.D()))
	t := time.Duration(float64(base) * math.Pow(r.Factor, float64(n)))
	if t > r.Max.D() || t < 0 {
		t = r.Max.D()
	}
	if t < base {
		t = base
	}
	return t
}

func (d *Daemon) ban(w *Watch, addr netip.Addr, g netip.Prefix, line string) {
	group := g.String()
	t := d.BanTime(w.Cfg, group)
	b := state.Ban{Time: d.Now(), Watch: w.Cfg.Name, List: w.Cfg.Ban.List, Address: addr.String(), Group: group,
		Protocol: strings.ToLower(w.Cfg.Ban.Protocol), Ports: strings.Join(w.Cfg.Ban.Port, ","), Seconds: int(t / time.Second), Reason: clip(line, 300)}
	d.log.Info("bounty reached; banning", "watch", w.Cfg.Name, "address", addr, "list", b.List, "ports", portsText(b), "for", config.FormatDuration(t))
	d.mu.Lock()
	d.pending[b.List+" "+b.Group+" "+b.Protocol+" "+b.Ports] = b
	d.mu.Unlock()
	select {
	case d.wake <- struct{}{}:
	default:
	}
}

func portsText(b state.Ban) string {
	if b.Ports == "" {
		return "all"
	}
	if b.Protocol == "" {
		return b.Ports
	}
	return b.Protocol + "/" + b.Ports
}

// sender sends the pending bans to Housecarl: at once, and again every
// RetryEvery while Housecarl can't be reached.
func (d *Daemon) sender(ctx context.Context) {
	t := time.NewTicker(d.RetryEvery)
	defer t.Stop()
	failing := false
	for {
		select {
		case <-ctx.Done():
			return
		case <-d.wake:
		case <-t.C:
		}
		d.mu.Lock()
		var todo []state.Ban
		for _, b := range d.pending {
			todo = append(todo, b)
		}
		d.mu.Unlock()
		sort.Slice(todo, func(i, j int) bool { return todo[i].Time.Before(todo[j].Time) })
		for _, b := range todo {
			err := d.send(ctx, b)
			if err != nil && !housecarl.Refused(err) {
				if !failing {
					d.log.Warn("can't reach Housecarl; the bans wait and are sent again", "err", err, "waiting", len(todo))
					failing = true
				}
				break
			}
			if failing {
				d.log.Info("Housecarl is back; sending the bans that waited")
				failing = false
			}
			d.mu.Lock()
			if cur, ok := d.pending[b.List+" "+b.Group+" "+b.Protocol+" "+b.Ports]; ok && cur.Time.Equal(b.Time) {
				delete(d.pending, b.List+" "+b.Group+" "+b.Protocol+" "+b.Ports)
			}
			d.mu.Unlock()
		}
		d.savePending()
	}
}

// send makes one ban. A ban Housecarl refuses (never banned, no
// permission) isn't sent again.
func (d *Daemon) send(ctx context.Context, b state.Ban) error {
	c, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	// What waited is sent for the time it has left.
	secs := b.Seconds - int(d.Now().Sub(b.Time)/time.Second)
	if secs <= 0 {
		return nil
	}
	res, err := d.hc.Ban(c, b.List, housecarl.BanInput{Address: b.Address, Protocol: b.Protocol, Ports: b.Ports, Time: secs,
		Source: "guard/" + b.Watch, Reason: b.Reason})
	if err != nil {
		if housecarl.Refused(err) {
			b.Result = err.Error()
			d.state.Failed(b)
			d.mu.Lock()
			last, seen := d.refused[b.Group]
			d.refused[b.Group] = d.Now()
			d.mu.Unlock()
			if !seen || d.Now().Sub(last) > time.Hour {
				d.log.Warn("Housecarl refused the ban", "watch", b.Watch, "address", b.Address, "list", b.List, "err", err)
			}
		}
		return err
	}
	b.Result = "banned"
	for _, x := range res.Bans {
		if x.Existing {
			b.Result = "renewed"
		}
	}
	d.state.Banned(b)
	d.log.Info("banned", "watch", b.Watch, "address", b.Address, "list", b.List, "ports", portsText(b), "for", config.FormatDuration(time.Duration(secs)*time.Second),
		"result", b.Result, "connections_closed", res.Closed)
	return nil
}

func (d *Daemon) savePending() {
	d.mu.Lock()
	var p []state.Ban
	for _, b := range d.pending {
		p = append(p, b)
	}
	d.mu.Unlock()
	d.state.SetPending(p)
	d.state.Forget(d.Now().Add(-d.cfg.Defaults.Repeat.Forget.D()))
	if err := d.state.Save(); err != nil {
		d.log.Warn("saving the state failed", "file", d.cfg.State, "err", err)
	}
}

func clip(s string, n int) string {
	if len(s) > n {
		return s[:n]
	}
	return s
}

// ---- status

// Status is what guard status shows.
type Status struct {
	Started time.Time     `json:"started"`
	Watches []WatchStatus `json:"watches"`
	Recent  []state.Ban   `json:"recent"`
	Pending []state.Ban   `json:"pending"`
}

// WatchStatus is a watch's.
type WatchStatus struct {
	Name    string          `json:"name"`
	Source  string          `json:"source"`
	List    string          `json:"list"`
	Hits    int             `json:"hits"`
	Within  string          `json:"within"`
	Lines   uint64          `json:"lines"`
	Matches uint64          `json:"matches"`
	Bans    uint64          `json:"bans"`
	Top     []bounty.Bounty `json:"top"`
}

// Status reports the watches, their highest bounties and the recent bans.
func (d *Daemon) Status() Status {
	s := Status{Started: d.started}
	for _, w := range d.watches {
		s.Watches = append(s.Watches, WatchStatus{Name: w.Cfg.Name, Source: Short(w.Cfg), List: w.Cfg.Ban.List, Hits: w.Cfg.Hits,
			Within: w.Cfg.Within.String(), Lines: w.lines.Load(), Matches: w.matches.Load(), Bans: w.bans.Load(), Top: w.Tracker.Top(10, d.Now())})
	}
	s.Recent, s.Pending = d.state.Snapshot()
	if len(s.Recent) > 50 {
		s.Recent = s.Recent[:50]
	}
	return s
}

// serveStatus answers GET /status on the local socket (root and the
// guard user only: the socket is 0660 in /run/guard).
func (d *Daemon) serveStatus(ctx context.Context) error {
	path := d.cfg.Socket
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		return err
	}
	os.Remove(path)
	ln, err := net.Listen("unix", path)
	if err != nil {
		return err
	}
	os.Chmod(path, 0o660)
	mux := http.NewServeMux()
	mux.HandleFunc("GET /status", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(d.Status())
	})
	srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second}
	go srv.Serve(ln)
	go func() {
		<-ctx.Done()
		srv.Close()
		os.Remove(path)
	}()
	return nil
}

// ---- never banned

// NeverBan holds the addresses never banned: prefixes, and host names
// looked up every ten minutes.
type NeverBan struct {
	log      *slog.Logger
	prefixes []netip.Prefix
	names    []string
	mu       sync.RWMutex
	resolved map[string][]netip.Addr
	// Lookup resolves names (tests).
	Lookup func(ctx context.Context, host string) ([]netip.Addr, error)
}

// NewNeverBan reads never_ban (checked by config).
func NewNeverBan(list []string, log *slog.Logger) *NeverBan {
	n := &NeverBan{log: log, resolved: map[string][]netip.Addr{},
		Lookup: func(ctx context.Context, host string) ([]netip.Addr, error) {
			return net.DefaultResolver.LookupNetIP(ctx, "ip", host)
		}}
	for _, e := range list {
		if p, err := netip.ParsePrefix(e); err == nil {
			n.prefixes = append(n.prefixes, p.Masked())
		} else if a, err := netip.ParseAddr(e); err == nil {
			n.prefixes = append(n.prefixes, netip.PrefixFrom(a.Unmap(), a.Unmap().BitLen()))
		} else {
			n.names = append(n.names, e)
		}
	}
	return n
}

// Run looks the names up now and every ten minutes.
func (n *NeverBan) Run(ctx context.Context) {
	if len(n.names) == 0 {
		return
	}
	for {
		n.resolve(ctx)
		select {
		case <-ctx.Done():
			return
		case <-time.After(10 * time.Minute):
		}
	}
}

func (n *NeverBan) resolve(ctx context.Context) {
	for _, h := range n.names {
		c, cancel := context.WithTimeout(ctx, 10*time.Second)
		addrs, err := n.Lookup(c, h)
		cancel()
		if err != nil {
			// Keep what it was: a name that can't be looked up now is still
			// never banned as it was.
			if n.log != nil {
				n.log.Warn("can't look up a never_ban name; keeping its last addresses", "name", h, "err", err)
			}
			continue
		}
		for i := range addrs {
			addrs[i] = addrs[i].Unmap()
		}
		n.mu.Lock()
		n.resolved[h] = addrs
		n.mu.Unlock()
	}
}

// Why says why an address is never banned ("" when it may be).
func (n *NeverBan) Why(a netip.Addr) string {
	a = a.Unmap()
	switch {
	case a.IsLoopback(), a.IsUnspecified(), a.IsMulticast(), a.IsLinkLocalUnicast():
		return "loopback, link-local, multicast or unspecified"
	}
	for _, p := range n.prefixes {
		if p.Contains(a) {
			return "in never_ban " + p.String()
		}
	}
	n.mu.RLock()
	defer n.mu.RUnlock()
	for h, addrs := range n.resolved {
		for _, x := range addrs {
			if x == a {
				return "never_ban " + h
			}
		}
	}
	return ""
}