// 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 { log *slog.Logger hc Banner logs source.LogReader state *state.State started time.Time ctx context.Context // Run's, for what a reload starts mu sync.Mutex // cfg, never and watches change on a reload. cfg *config.Config never *NeverBan neverStop context.CancelFunc watches []*Watch 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 // Load reads the configuration again for a reload (guard run sets it). Load func() (*config.Config, error) // Sources makes a watch's source (tests); nil: from its configuration. Sources func(*config.Watch) source.Source // Now and RetryEvery can be changed by tests. Now func() time.Time RetryEvery time.Duration } // Watch is one watch at work. type Watch struct { Source source.Source Tracker *bounty.Tracker // mu guards Cfg and Matcher, which a reload replaces. mu sync.Mutex Cfg *config.Watch Matcher *match.Matcher stop context.CancelFunc done chan struct{} lines, matches, bans atomic.Uint64 } func (w *Watch) setup() (*config.Watch, *match.Matcher) { w.mu.Lock() defer w.mu.Unlock() return w.Cfg, w.Matcher } // 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, logs: logs, 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) } d.watches = append(d.watches, &Watch{Cfg: w, Matcher: m, Tracker: bounty.New(w.Hits, w.Within.D())}) } if len(d.watches) == 0 { return nil, errors.New("every watch is disabled: nothing to do") } return d, nil } // newSource makes a watch's source from its configuration. func (d *Daemon) newSource(w *config.Watch) source.Source { if d.Sources != nil { return d.Sources(w) } log := d.log.With("watch", w.Name) switch w.Kind() { case "journal": return &source.Journal{Units: w.Journal.Unit, Identifiers: w.Journal.Identifier, Log: log} case config.SourceHousecarlDrops: return &source.HousecarlDrops{Client: d.logs, Protocol: w.Protocol, Shadow: w.Shadow, Log: log} } return &source.File{Pattern: w.File, Log: log} } // Watches are the watches at work (tests set their sources before Run). func (d *Daemon) Watches() []*Watch { d.mu.Lock() defer d.mu.Unlock() return append([]*Watch(nil), d.watches...) } func (d *Daemon) config() *config.Config { d.mu.Lock() defer d.mu.Unlock() return d.cfg } func (d *Daemon) neverBan() *NeverBan { d.mu.Lock() defer d.mu.Unlock() return d.never } // Run watches until ctx ends. func (d *Daemon) Run(ctx context.Context) error { d.started = d.Now() d.ctx = ctx d.loadGroups(ctx) d.mu.Lock() d.startNever() for _, w := range d.watches { d.start(w) } d.mu.Unlock() var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() d.sender(ctx) }() if sock := d.config().Socket; sock != "" { if err := d.serveStatus(ctx); err != nil { d.log.Warn("no status socket (guard status won't work)", "socket", sock, "err", err) } } <-ctx.Done() wg.Wait() for _, w := range d.Watches() { <-w.done } d.savePending() return nil } // startNever looks never_ban's names up until a reload replaces it // (d.mu held). func (d *Daemon) startNever() { c, stop := context.WithCancel(d.ctx) d.neverStop = stop go d.never.Run(c) } // start runs a watch: its source, and its lines through Line, until // stopped or Run's context ends (d.mu held). func (d *Daemon) start(w *Watch) { if w.Source == nil { w.Source = d.newSource(w.Cfg) } ctx, stop := context.WithCancel(d.ctx) w.stop, w.done = stop, make(chan struct{}) lines := make(chan source.Line, 256) src := w.Source go func() { defer close(lines) for { err := src.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 close(w.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) } // 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() { cfg, _ := w.setup() list := 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) cfg, m := w.setup() addr := l.Addr if !addr.IsValid() { var ok bool if addr, ok = m.Match(l.Text); !ok { return } } w.matches.Add(1) if why := d.neverBan().Why(addr); why != "" { d.log.Debug("match, but never banned", "watch", cfg.Name, "address", addr, "why", why) return } g := match.Group(addr, d.group(cfg.Ban.List)) n, reached := w.Tracker.Add(g, l.Time) d.log.Debug("match", "watch", cfg.Name, "address", addr, "bounty", n, "hits", cfg.Hits, "line", clip(l.Text, 300)) if !reached { return } w.bans.Add(1) d.ban(cfg, 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.config().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 *config.Watch, addr netip.Addr, g netip.Prefix, line string) { group := g.String() t := d.BanTime(w, group) b := state.Ban{Time: d.Now(), Watch: w.Name, List: w.Ban.List, Address: addr.String(), Group: group, Protocol: strings.ToLower(w.Ban.Protocol), Ports: strings.Join(w.Ban.Port, ","), Seconds: int(t / time.Second), Reason: clip(line, 300)} d.log.Info("bounty reached; banning", "watch", w.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() cfg := d.config() d.state.SetPending(p) d.state.Forget(d.Now().Add(-cfg.Defaults.Repeat.Forget.D())) if err := d.state.Save(); err != nil { d.log.Warn("saving the state failed", "file", 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() { cfg, _ := w.setup() s.Watches = append(s.Watches, WatchStatus{Name: cfg.Name, Source: Short(cfg), List: cfg.Ban.List, Hits: cfg.Hits, Within: 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.config().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()) }) mux.HandleFunc("POST /reload", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") res, err := d.ReloadFile() if err != nil { w.WriteHeader(http.StatusUnprocessableEntity) json.NewEncoder(w).Encode(map[string]string{"error": err.Error()}) return } json.NewEncoder(w).Encode(res) }) 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 "" }