Hammerhead/alerts.go

package hammerhead

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"time"

	"git.veillax.com/Veillax/reef/netguard"
)

// Alerts: every few seconds Hammerhead derives the set of current problems from
// what the edges and sites report (an edge that went quiet, a port that can't
// bind, a target failing its health check, a certificate error, a mail relay
// that isn't deliverable). A problem has to persist for the debounce time before
// anyone is told, which keeps a restart or a blip from paging you. You get one
// message when it starts and one when it clears, and optionally a reminder.
// Active alerts are kept in the database so a restart doesn't repeat them.

const (
	defaultDebounce = 2 * time.Minute
	defaultOffline  = 90 * time.Second
)

// AlertChannel is somewhere alerts are sent: an email address or a webhook.
type AlertChannel struct {
	ID           string `json:"id"`
	OrgID        string `json:"orgId"` // "" = a server-level channel (for edges)
	Name         string `json:"name"`
	Kind         string `json:"kind"`   // "email" or "webhook"
	Target       string `json:"target"` // address(es), or a webhook URL (secret: URLs often carry tokens)
	Format       string `json:"format"` // webhook body: "json", "slack", "discord" or "ntfy"
	AllowPrivate bool   `json:"allowPrivate"`
	Enabled      bool   `json:"enabled"`
}

type alertState struct {
	Key, OrgID, Kind, Name, Message string
	Since                           time.Time
	Fired                           bool
	Notified                        time.Time
}

type problem struct{ org, kind, name, msg string }

type alertNote struct {
	Status  string    `json:"status"` // "firing", "resolved" or "test"
	Org     string    `json:"org"`
	Kind    string    `json:"kind"`
	Name    string    `json:"name"`
	Message string    `json:"message"`
	Since   time.Time `json:"since"`
	Time    time.Time `json:"time"`
}

func (s *Server) offlineAfter() time.Duration {
	if s.opts.OfflineAfter > 0 {
		return s.opts.OfflineAfter
	}
	return defaultOffline
}

// problemsLocked derives every current problem. Caller holds s.mu.
func (s *Server) problemsLocked(now time.Time) map[string]problem {
	out := map[string]problem{}
	add := func(key, org, kind, name, msg string) { out[key] = problem{org, kind, name, msg} }
	quiet := func(t time.Time) (bool, string) {
		if t.IsZero() {
			return false, "" // never connected: not set up yet, nothing to alert about
		}
		d := now.Sub(t)
		return d > s.offlineAfter(), d.Round(time.Second).String()
	}

	edgeDown, siteDown := map[string]bool{}, map[string]bool{}
	for _, e := range s.st.Edges {
		if down, d := quiet(e.LastSeen); down {
			edgeDown[e.ID] = true
			msg := fmt.Sprintf("Edge %s has not contacted Hammerhead for %s", e.Name, d)
			add("edge:"+e.ID+"@", "", "edge", e.Name, msg) // server-level
			for _, org := range s.orgsUsingEdgeLocked(e.ID) {
				add("edge:"+e.ID+"@"+org, org, "edge", e.Name, msg)
			}
		}
	}
	for _, st := range s.st.Sites {
		if st.Local {
			continue
		}
		if down, d := quiet(st.LastSeen); down {
			siteDown[st.ID] = true
			add("site:"+st.ID, st.OrgID, "site", st.Name, fmt.Sprintf("Site %s has not contacted Hammerhead for %s", st.Name, d))
		}
	}
	for _, m := range s.st.Mappings {
		if !m.Enabled || edgeDown[m.EdgeID] {
			continue // a dead edge's last report is stale: the edge alert already says it
		}
		st, known := s.status[m.ID]
		if !known {
			continue
		}
		if st.Error != "" {
			add("port:"+m.ID, m.OrgID, "port", m.Name, fmt.Sprintf("%s can't listen: %s", m.Name, st.Error))
		}
		if strings.HasPrefix(st.Cert, "error") {
			add("cert:"+m.ID, m.OrgID, "certificate", m.Name, fmt.Sprintf("%s has a certificate problem: %s", m.Name, st.Cert))
		}
		if m.HealthSeconds != -1 {
			for i, th := range st.Targets {
				if i >= len(m.Targets) || th.Healthy || siteDown[m.Targets[i].SiteID] {
					continue
				}
				t := m.Targets[i]
				add(fmt.Sprintf("target:%s:%d", m.ID, i), m.OrgID, "target", m.Name,
					fmt.Sprintf("%s: the target %s:%d isn't answering (%s)", m.Name, t.Host, t.Port, th.Detail))
			}
		}
	}
	for _, x := range s.st.MailEgress {
		if !x.Enabled || edgeDown[x.EdgeID] || siteDown[x.SiteID] {
			continue
		}
		if msg := s.mailStatus[x.ID]; msg != "" {
			add("mail:"+x.ID, x.OrgID, "mail", x.Name, fmt.Sprintf("Mail relay %s can't run: %s", x.Name, msg))
		}
		if chk, ok := s.mailChecks[x.ID]; ok {
			if !chk.PTRMatchesHelo {
				add("mailptr:"+x.ID, x.OrgID, "mail", x.Name, fmt.Sprintf("%s: reverse DNS of %s doesn't match %s, so receiving servers may reject its mail", x.Name, x.SourceIP, x.HeloName))
			}
			if !chk.HeloResolves {
				add("mailhelo:"+x.ID, x.OrgID, "mail", x.Name, fmt.Sprintf("%s: %s doesn't resolve to %s", x.Name, x.HeloName, x.SourceIP))
			}
			if chk.Port25 != "" && chk.Port25 != "open" {
				add("mail25:"+x.ID, x.OrgID, "mail", x.Name, fmt.Sprintf("%s: can't connect out on port 25 from %s", x.Name, x.SourceIP))
			}
		}
	}
	for _, k := range s.st.DKIM {
		for _, x := range s.st.MailEgress {
			if x.OrgID != k.OrgID || !x.Enabled || edgeDown[x.EdgeID] {
				continue
			}
			if st := s.mailChecks[x.ID].DKIM[k.ID]; st == "missing" || st == "mismatch" {
				add("dkim:"+k.ID, k.OrgID, "mail", k.Domain, fmt.Sprintf("DKIM key %s._domainkey.%s is %s in DNS, so signed mail will fail verification", k.Selector, k.Domain, st))
				break
			}
		}
	}
	return out
}

// judgeableLocked reports whether we currently have the data to decide an alert's
// problem is gone. After a restart the live statuses are empty until edges and
// sites report again; resolving then would announce a recovery that never happened.
func (s *Server) judgeableLocked(st *alertState) bool {
	kind, rest, _ := strings.Cut(st.Key, ":")
	switch kind {
	case "port", "cert":
		if s.findMapping(rest) == nil {
			return true // the thing was deleted: nothing left to be wrong
		}
		_, known := s.status[rest]
		return known
	case "target":
		id, _, _ := strings.Cut(rest, ":")
		if s.findMapping(id) == nil {
			return true
		}
		_, known := s.status[id]
		return known
	case "mail":
		if s.findMailEgress(rest) == nil {
			return true
		}
		_, known := s.mailStatus[rest]
		return known
	case "mailptr", "mailhelo", "mail25":
		if s.findMailEgress(rest) == nil {
			return true
		}
		_, known := s.mailChecks[rest]
		return known
	case "dkim":
		return len(s.mailChecks) > 0
	}
	return true // edges and sites carry their own last-seen time
}

func (s *Server) orgsUsingEdgeLocked(edgeID string) []string {
	seen := map[string]bool{}
	var out []string
	for _, m := range s.st.Mappings {
		if m.EdgeID == edgeID && !seen[m.OrgID] {
			seen[m.OrgID] = true
			out = append(out, m.OrgID)
		}
	}
	for _, x := range s.st.MailEgress {
		if x.EdgeID == edgeID && !seen[x.OrgID] {
			seen[x.OrgID] = true
			out = append(out, x.OrgID)
		}
	}
	return out
}

func (s *Server) debounceFor(orgID string) time.Duration {
	if s.opts.AlertDebounce > 0 {
		return s.opts.AlertDebounce
	}
	if o := s.findOrg(orgID); o != nil && o.AlertDebounceMin > 0 {
		return time.Duration(o.AlertDebounceMin) * time.Minute
	}
	return defaultDebounce
}

func (s *Server) repeatFor(orgID string) time.Duration {
	if o := s.findOrg(orgID); o != nil && o.AlertRepeatHours > 0 {
		return time.Duration(o.AlertRepeatHours) * time.Hour
	}
	return 0
}

type delivery struct {
	note     alertNote
	channels []AlertChannel
}

// evaluateAlerts moves alert state forward to now and sends whatever became due.
func (s *Server) evaluateAlerts(now time.Time) {
	var out []delivery
	s.mu.Lock()
	probs := s.problemsLocked(now)
	orgName := func(id string) string {
		if o := s.findOrg(id); o != nil {
			return o.Name
		}
		return "server"
	}
	queue := func(st *alertState, status string) {
		var chans []AlertChannel
		for _, c := range s.st.Channels {
			if c.OrgID == st.OrgID && c.Enabled {
				chans = append(chans, *c)
			}
		}
		out = append(out, delivery{alertNote{Status: status, Org: orgName(st.OrgID), Kind: st.Kind, Name: st.Name, Message: st.Message, Since: st.Since, Time: now}, chans})
		s.db.addAlertEvent(now, st.OrgID, st.Key, status, st.Name, st.Message)
	}
	keys := make([]string, 0, len(probs))
	for k := range probs {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		p := probs[k]
		st := s.alerts[k]
		if st == nil {
			st = &alertState{Key: k, OrgID: p.org, Kind: p.kind, Since: now}
			s.alerts[k] = st
		}
		st.Name, st.Message = p.name, p.msg
		switch {
		case !st.Fired && now.Sub(st.Since) >= s.debounceFor(p.org):
			st.Fired, st.Notified = true, now
			queue(st, "firing")
		case st.Fired && s.repeatFor(p.org) > 0 && now.Sub(st.Notified) >= s.repeatFor(p.org):
			st.Notified = now
			queue(st, "firing")
		}
		s.db.saveAlert(st)
	}
	for k, st := range s.alerts {
		if _, still := probs[k]; still {
			continue
		}
		if !s.judgeableLocked(st) {
			continue // e.g. just restarted: the agents haven't reported yet, so we can't say it cleared
		}
		if st.Fired {
			queue(st, "resolved")
		}
		delete(s.alerts, k)
		s.db.deleteAlert(k)
	}
	s.mu.Unlock()
	for _, d := range out {
		go s.deliver(d)
	}
}

func (s *Server) loadAlerts() {
	for _, a := range s.db.loadAlertRows() {
		a := a
		s.alerts[a.Key] = &a
	}
}

func (s *Server) alertLoop() {
	every := s.opts.AlertInterval
	if every <= 0 {
		every = 15 * time.Second
	}
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-s.stop:
			return
		case <-t.C:
			s.evaluateAlerts(time.Now())
		}
	}
}

// ---------- delivery ----------

func (n alertNote) subject() string {
	switch n.Status {
	case "resolved":
		return "[Hammerhead] Resolved: " + n.Name
	case "test":
		return "[Hammerhead] Test alert"
	}
	return "[Hammerhead] Problem: " + n.Name
}

func (n alertNote) text() string {
	switch n.Status {
	case "resolved":
		return "Resolved: " + n.Message
	case "test":
		return "This is a test alert from Hammerhead. If you can read this, the channel works."
	}
	return "Problem: " + n.Message
}

func (s *Server) deliver(d delivery) {
	for _, ch := range d.channels {
		if err := s.send(ch, d.note); err != nil {
			log.Printf("alert channel %q: %v", ch.Name, err)
		}
	}
}

func (s *Server) send(ch AlertChannel, n alertNote) error {
	switch ch.Kind {
	case "email":
		var errs []error
		for _, to := range strings.Split(ch.Target, ",") {
			if to = strings.TrimSpace(to); to != "" {
				if err := s.Mailer(to, n.subject(), n.text()+"\n\nOrganization: "+n.Org+"\nSince: "+n.Since.Format(time.RFC1123)+"\n"); err != nil {
					errs = append(errs, err)
				}
			}
		}
		return errors.Join(errs...)
	case "webhook":
		body, ctype, hdr := webhookBody(ch.Format, n)
		req, err := http.NewRequest("POST", ch.Target, bytes.NewReader(body))
		if err != nil {
			return err
		}
		req.Header.Set("Content-Type", ctype)
		for k, v := range hdr {
			req.Header.Set(k, v)
		}
		resp, err := netguard.Client(ch.AllowPrivate).Do(req)
		if err != nil {
			return err
		}
		resp.Body.Close()
		if resp.StatusCode >= 300 {
			return fmt.Errorf("the webhook answered %s", resp.Status)
		}
		return nil
	}
	return errors.New("unknown channel kind")
}

func webhookBody(format string, n alertNote) (body []byte, ctype string, hdr map[string]string) {
	text := n.subject() + "\n" + n.text()
	switch format {
	case "slack":
		body, _ = json.Marshal(map[string]string{"text": text})
	case "discord":
		body, _ = json.Marshal(map[string]string{"content": text})
	case "ntfy":
		prio, tags := "high", "rotating_light"
		if n.Status != "firing" {
			prio, tags = "default", "white_check_mark"
		}
		return []byte(n.text()), "text/plain", map[string]string{"Title": n.subject(), "Priority": prio, "Tags": tags}
	default:
		body, _ = json.Marshal(n)
	}
	return body, "application/json", nil
}

// validateChannel normalises ch. allowPrivate may only be kept if the caller is trusted.
func (s *Server) validateChannel(ch *AlertChannel, trusted bool) error {
	ch.Name = strings.TrimSpace(ch.Name)
	ch.Target = strings.TrimSpace(ch.Target)
	switch ch.Kind {
	case "email":
		for _, a := range strings.Split(ch.Target, ",") {
			if a = strings.TrimSpace(a); !strings.Contains(a, "@") || strings.ContainsAny(a, " \r\n<>") {
				return errors.New("enter one or more email addresses, separated by commas")
			}
		}
		ch.Format, ch.AllowPrivate = "", false
	case "webhook":
		u, err := url.Parse(ch.Target)
		if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
			return errors.New("the webhook must be an http:// or https:// URL")
		}
		switch ch.Format {
		case "json", "slack", "discord", "ntfy":
		case "":
			ch.Format = "json"
		default:
			return errors.New("format must be json, slack, discord or ntfy")
		}
		if ch.AllowPrivate && !trusted {
			return errors.New("sending to private addresses (like a service on your home network) needs a server administrator to allow it for this organization")
		}
	default:
		return errors.New("a channel is an email or a webhook")
	}
	if ch.Name == "" {
		ch.Name = ch.Kind
	}
	return nil
}