Hammerhead/alerts_test.go

package hammerhead

import (
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"git.veillax.com/Veillax/hammerhead/internal/secretbox"
	"git.veillax.com/Veillax/reef/model"
)

type hook struct {
	URL  string
	mu   sync.Mutex
	got  []string // raw bodies
	ch   chan string
	hdrs []http.Header
}

func newHook(t *testing.T) *hook {
	h := &hook{ch: make(chan string, 32)}
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		b, _ := io.ReadAll(r.Body)
		h.mu.Lock()
		h.got = append(h.got, string(b))
		h.hdrs = append(h.hdrs, r.Header.Clone())
		h.mu.Unlock()
		h.ch <- string(b)
	}))
	t.Cleanup(ts.Close)
	h.URL = ts.URL + "/hook/secret-token"
	return h
}

// next returns the next delivery, or "" if none arrives soon.
func (h *hook) next(wait time.Duration) string {
	select {
	case b := <-h.ch:
		return b
	case <-time.After(wait):
		return ""
	}
}

type fx struct {
	s    *Server
	org  string
	base time.Time
	hook *hook
	mail chan string
}

func newFx(t *testing.T) *fx {
	s, err := New(t.TempDir(), Options{MasterKey: secretbox.GenerateKey(), AlertInterval: time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(s.Close)
	f := &fx{s: s, base: time.Now(), hook: newHook(t), mail: make(chan string, 16)}
	s.Mailer = func(to, subject, body string) error { f.mail <- to + "|" + subject + "|" + body; return nil }
	s.mu.Lock()
	org, _ := s.newOrgLocked("Acme", true)
	f.org = org.ID
	s.st.Edges = append(s.st.Edges, &model.Edge{ID: "e1", Name: "edge-1", LastSeen: f.base, TunnelIP: "100.89.0.1"})
	s.st.Sites = append(s.st.Sites, &model.Site{ID: "s1", OrgID: org.ID, Name: "home", LastSeen: f.base, TunnelIP: "100.89.0.2"})
	s.st.Mappings = append(s.st.Mappings, &model.Mapping{ID: "m1", OrgID: org.ID, Name: "Minecraft", EdgeID: "e1", Proto: "tcp", Port: 25565, Enabled: true,
		Targets: []model.Target{{SiteID: "s1", Host: "192.168.1.30", Port: 25565}}})
	s.status["m1"] = model.MappingStatus{Targets: []model.TargetHealth{{Healthy: true}}}
	s.st.Channels = append(s.st.Channels, &AlertChannel{ID: "c1", OrgID: org.ID, Name: "hook", Kind: "webhook", Target: f.hook.URL, Format: "json", AllowPrivate: true, Enabled: true})
	s.mu.Unlock()
	return f
}

// touch keeps the edge and site looking alive at the given fake time.
func (f *fx) touch(at time.Time) {
	f.s.mu.Lock()
	f.s.st.Edges[0].LastSeen, f.s.st.Sites[0].LastSeen = at, at
	f.s.mu.Unlock()
}

func (f *fx) setHealthy(h bool) {
	f.s.mu.Lock()
	f.s.status["m1"] = model.MappingStatus{Targets: []model.TargetHealth{{Healthy: h, Detail: "connection refused"}}}
	f.s.mu.Unlock()
}

func (f *fx) eval(at time.Time) {
	f.touch(at)
	f.s.evaluateAlerts(at)
}

func TestAlertLifecycleDebounceOnceAndResolve(t *testing.T) {
	f := newFx(t)
	t0 := f.base
	f.eval(t0)
	if b := f.hook.next(200 * time.Millisecond); b != "" {
		t.Fatalf("a healthy system alerted: %s", b)
	}

	f.setHealthy(false)
	f.eval(t0.Add(10 * time.Second))
	f.eval(t0.Add(70 * time.Second))
	if b := f.hook.next(200 * time.Millisecond); b != "" {
		t.Fatalf("alerted before the debounce time: %s", b)
	}
	f.eval(t0.Add(135 * time.Second)) // 2 minutes after it began
	var note alertNote
	b := f.hook.next(2 * time.Second)
	if b == "" || json.Unmarshal([]byte(b), &note) != nil || note.Status != "firing" || !strings.Contains(note.Message, "Minecraft") || !strings.Contains(note.Message, "192.168.1.30:25565") {
		t.Fatalf("firing notification: %q", b)
	}
	if note.Org != "Acme" || note.Kind != "target" {
		t.Errorf("notification context: %+v", note)
	}
	// the same problem is announced once, not every evaluation
	f.eval(t0.Add(200 * time.Second))
	f.eval(t0.Add(10 * time.Minute))
	if b := f.hook.next(300 * time.Millisecond); b != "" {
		t.Fatalf("a repeat notification arrived with no reminder configured: %s", b)
	}

	f.setHealthy(true)
	f.eval(t0.Add(11 * time.Minute))
	b = f.hook.next(2 * time.Second)
	if json.Unmarshal([]byte(b), &note) != nil || note.Status != "resolved" || !strings.Contains(note.Message, "Minecraft") {
		t.Fatalf("resolved notification: %q", b)
	}
	f.eval(t0.Add(12 * time.Minute))
	if b := f.hook.next(200 * time.Millisecond); b != "" {
		t.Fatalf("resolved twice: %s", b)
	}

	hist := f.s.db.queryAlertEvents(f.org, 10)
	if len(hist) != 2 || hist[0].Status != "resolved" || hist[1].Status != "firing" {
		t.Errorf("alert history: %+v", hist)
	}
}

func TestShortBlipsNeverAlert(t *testing.T) {
	f := newFx(t)
	t0 := f.base
	f.setHealthy(false)
	f.eval(t0)
	f.eval(t0.Add(60 * time.Second))
	f.setHealthy(true) // fixed itself inside the debounce window
	f.eval(t0.Add(90 * time.Second))
	f.eval(t0.Add(10 * time.Minute))
	if b := f.hook.next(300 * time.Millisecond); b != "" {
		t.Fatalf("a blip shorter than the debounce time alerted: %s", b)
	}
	if len(f.s.db.queryAlertEvents(f.org, 10)) != 0 {
		t.Error("a blip left history behind")
	}
}

func TestReminderRepeatsWhenAsked(t *testing.T) {
	f := newFx(t)
	f.s.mu.Lock()
	f.s.findOrg(f.org).AlertRepeatHours = 1
	f.s.mu.Unlock()
	t0 := f.base
	f.setHealthy(false)
	f.eval(t0)
	f.eval(t0.Add(3 * time.Minute))
	if f.hook.next(2*time.Second) == "" {
		t.Fatal("no first alert")
	}
	f.eval(t0.Add(30 * time.Minute))
	if b := f.hook.next(200 * time.Millisecond); b != "" {
		t.Fatalf("reminded too early: %s", b)
	}
	f.eval(t0.Add(64 * time.Minute))
	if f.hook.next(2*time.Second) == "" {
		t.Fatal("no reminder after an hour")
	}
}

func TestDeadEdgeSilencesItsPortsAndRoutesByOrganization(t *testing.T) {
	f := newFx(t)
	other := newHook(t)
	server := newHook(t)
	f.s.mu.Lock()
	o2, _ := f.s.newOrgLocked("Elsewhere", true)
	f.s.st.Channels = append(f.s.st.Channels,
		&AlertChannel{ID: "c2", OrgID: o2.ID, Name: "other org", Kind: "webhook", Target: other.URL, Format: "json", AllowPrivate: true, Enabled: true},
		&AlertChannel{ID: "c3", OrgID: "", Name: "ops", Kind: "webhook", Target: server.URL, Format: "json", AllowPrivate: true, Enabled: true},
		&AlertChannel{ID: "c4", OrgID: f.org, Name: "muted", Kind: "email", Target: "off@x.test", Enabled: false})
	f.s.mu.Unlock()
	t0 := f.base

	// the edge goes silent AND the port's target looks unhealthy: only the edge is reported
	f.setHealthy(false)
	f.s.evaluateAlerts(t0)
	f.s.mu.Lock()
	f.s.st.Edges[0].LastSeen = t0.Add(-10 * time.Minute)
	f.s.mu.Unlock()
	f.s.evaluateAlerts(t0.Add(3 * time.Minute)) // the edge problem starts here...
	f.s.mu.Lock()
	f.s.st.Sites[0].LastSeen = t0.Add(6 * time.Minute) // (the site itself is fine)
	f.s.mu.Unlock()
	f.s.evaluateAlerts(t0.Add(6 * time.Minute)) // ...and has lasted past the debounce here
	var note alertNote
	if b := f.hook.next(2 * time.Second); json.Unmarshal([]byte(b), &note) != nil || note.Kind != "edge" || !strings.Contains(note.Message, "edge-1") {
		t.Fatalf("the org's channel should hear about the dead edge: %q", b)
	}
	if b := f.hook.next(300 * time.Millisecond); b != "" {
		t.Errorf("the port's stale target problem should be hidden while its edge is down: %s", b)
	}
	if b := server.next(2 * time.Second); json.Unmarshal([]byte(b), &note) != nil || note.Org != "server" || note.Kind != "edge" {
		t.Errorf("the server-level channel should hear about the edge: %q", b)
	}
	if b := other.next(300 * time.Millisecond); b != "" {
		t.Errorf("another organization's channel was notified: %s", b)
	}
	select {
	case m := <-f.mail:
		t.Errorf("a disabled channel sent mail: %s", m)
	case <-time.After(200 * time.Millisecond):
	}
}

func TestAlertsSurviveARestartWithoutFalseRecovery(t *testing.T) {
	dir := t.TempDir()
	key := secretbox.GenerateKey()
	s, err := New(dir, Options{MasterKey: key, AlertInterval: time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	h := newHook(t)
	now := time.Now()
	s.mu.Lock()
	org, _ := s.newOrgLocked("Acme", true)
	s.st.Edges = append(s.st.Edges, &model.Edge{ID: "e1", Name: "edge-1", LastSeen: now})
	s.st.Sites = append(s.st.Sites, &model.Site{ID: "s1", OrgID: org.ID, Name: "home", LastSeen: now})
	s.st.Mappings = append(s.st.Mappings, &model.Mapping{ID: "m1", OrgID: org.ID, Name: "Web", EdgeID: "e1", Proto: "tcp", Port: 80, Enabled: true,
		Targets: []model.Target{{SiteID: "s1", Host: "10.0.0.1", Port: 80}}})
	s.st.Channels = append(s.st.Channels, &AlertChannel{ID: "c1", OrgID: org.ID, Name: "hook", Kind: "webhook", Target: h.URL, Format: "json", AllowPrivate: true, Enabled: true})
	s.status["m1"] = model.MappingStatus{Targets: []model.TargetHealth{{Healthy: false, Detail: "down"}}}
	s.mu.Unlock()
	s.save()
	s.evaluateAlerts(now)
	s.mu.Lock()
	s.st.Edges[0].LastSeen, s.st.Sites[0].LastSeen = now.Add(3*time.Minute), now.Add(3*time.Minute) // still alive; only the target is down
	s.mu.Unlock()
	s.evaluateAlerts(now.Add(3 * time.Minute))
	if h.next(2*time.Second) == "" {
		t.Fatal("no first alert")
	}
	s.Close()

	s2, err := New(dir, Options{MasterKey: key, AlertInterval: time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	defer s2.Close()
	s2.mu.Lock()
	restored := s2.alerts["target:m1:0"]
	s2.mu.Unlock()
	if restored == nil || !restored.Fired {
		t.Fatalf("the firing alert was not restored: %+v", restored)
	}
	// nothing has reported since the restart: that is NOT a recovery, and not a new alert either
	later := now.Add(5 * time.Minute)
	s2.mu.Lock()
	s2.st.Edges[0].LastSeen, s2.st.Sites[0].LastSeen = later, later
	s2.mu.Unlock()
	s2.evaluateAlerts(later)
	if b := h.next(400 * time.Millisecond); b != "" {
		t.Fatalf("restarting produced a notification: %s", b)
	}
	s2.mu.Lock()
	_, still := s2.alerts["target:m1:0"]
	s2.mu.Unlock()
	if !still {
		t.Fatal("the alert vanished before anything re-checked the target")
	}
	// once the edge reports the target healthy, it really is resolved
	s2.mu.Lock()
	s2.status["m1"] = model.MappingStatus{Targets: []model.TargetHealth{{Healthy: true}}}
	s2.mu.Unlock()
	s2.evaluateAlerts(later.Add(30 * time.Second))
	var note alertNote
	if b := h.next(2 * time.Second); json.Unmarshal([]byte(b), &note) != nil || note.Status != "resolved" {
		t.Fatalf("expected a recovery after a real healthy report, got %q", b)
	}
}

func TestDeliveryFormatsAndEmail(t *testing.T) {
	f := newFx(t)
	note := alertNote{Status: "firing", Org: "Acme", Kind: "target", Name: "Minecraft", Message: "Minecraft: the target is down", Since: f.base, Time: f.base}
	for _, tc := range []struct {
		format string
		check  func(body string, h http.Header) bool
	}{
		{"json", func(b string, _ http.Header) bool {
			return strings.Contains(b, `"status":"firing"`) && strings.Contains(b, `"org":"Acme"`)
		}},
		{"slack", func(b string, _ http.Header) bool {
			return strings.HasPrefix(b, `{"text":`) && strings.Contains(b, "Minecraft")
		}},
		{"discord", func(b string, _ http.Header) bool {
			return strings.HasPrefix(b, `{"content":`) && strings.Contains(b, "Minecraft")
		}},
		{"ntfy", func(b string, h http.Header) bool {
			return strings.Contains(b, "the target is down") && strings.Contains(h.Get("Title"), "Minecraft") && h.Get("Priority") == "high"
		}},
	} {
		h := newHook(t)
		ch := AlertChannel{Kind: "webhook", Target: h.URL, Format: tc.format, AllowPrivate: true}
		if err := f.s.send(ch, note); err != nil {
			t.Fatalf("%s: %v", tc.format, err)
		}
		body := h.next(2 * time.Second)
		h.mu.Lock()
		hdr := h.hdrs[0]
		h.mu.Unlock()
		if !tc.check(body, hdr) {
			t.Errorf("%s body/headers wrong: %q %v", tc.format, body, hdr)
		}
	}

	if err := f.s.send(AlertChannel{Kind: "email", Target: "a@x.test, b@y.test"}, note); err != nil {
		t.Fatal(err)
	}
	got := map[string]bool{}
	for i := 0; i < 2; i++ {
		select {
		case m := <-f.mail:
			parts := strings.SplitN(m, "|", 3)
			got[parts[0]] = strings.Contains(parts[1], "Minecraft") && strings.Contains(parts[2], "the target is down")
		case <-time.After(2 * time.Second):
		}
	}
	if !got["a@x.test"] || !got["b@y.test"] {
		t.Errorf("email delivery: %v", got)
	}

	// a webhook answering with an error is reported, not swallowed
	bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "no", 500) }))
	defer bad.Close()
	if err := f.s.send(AlertChannel{Kind: "webhook", Target: bad.URL, Format: "json", AllowPrivate: true}, note); err == nil {
		t.Error("a 500 from the webhook should be an error")
	}
}

func TestWebhooksCannotBeAimedAtInternalAddresses(t *testing.T) {
	f := newFx(t)
	h := newHook(t) // a loopback server, standing in for an internal service
	note := alertNote{Status: "test", Time: f.base, Since: f.base}
	for _, target := range []string{h.URL, strings.Replace(h.URL, "127.0.0.1", "localhost", 1), "http://169.254.169.254/latest/meta-data/", "http://10.0.0.5:9/x", "http://[::1]:9/x"} {
		err := f.s.send(AlertChannel{Kind: "webhook", Target: target, Format: "json", AllowPrivate: false}, note)
		if err == nil || !strings.Contains(err.Error(), "not allowed") {
			t.Errorf("%s: %v (should be refused as internal)", target, err)
		}
	}
	if b := h.next(200 * time.Millisecond); b != "" {
		t.Errorf("the internal server received a request: %s", b)
	}
	// the same URL is fine once a trusted admin allowed private targets
	if err := f.s.send(AlertChannel{Kind: "webhook", Target: h.URL, Format: "json", AllowPrivate: true}, note); err != nil {
		t.Errorf("allowed private target: %v", err)
	}
	// a redirect to an internal address is not followed
	redir := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		http.Redirect(w, r, h.URL, http.StatusFound)
	}))
	defer redir.Close()
	h.next(500 * time.Millisecond)
	err := f.s.send(AlertChannel{Kind: "webhook", Target: redir.URL, Format: "json", AllowPrivate: true}, note)
	if err == nil {
		t.Error("a redirect response must not count as delivered")
	}
	if b := h.next(300 * time.Millisecond); b != "" {
		t.Errorf("the redirect was followed: %s", b)
	}

	// validation: only trusted callers may set allowPrivate
	ch := AlertChannel{Kind: "webhook", Target: "http://192.168.1.9/x", AllowPrivate: true}
	if err := f.s.validateChannel(&ch, false); err == nil {
		t.Error("an untrusted caller must not be able to allow private targets")
	}
	if err := f.s.validateChannel(&ch, true); err != nil {
		t.Errorf("a trusted caller may: %v", err)
	}
	for _, bad := range []AlertChannel{{Kind: "webhook", Target: "ftp://x"}, {Kind: "webhook", Target: "not a url"}, {Kind: "email", Target: "nope"}, {Kind: "sms", Target: "1"}, {Kind: "webhook", Target: "http://x.test", Format: "xml"}} {
		if err := f.s.validateChannel(&bad, true); err == nil {
			t.Errorf("accepted %+v", bad)
		}
	}
}

func TestAlertAPIHidesWebhookURLsAndMetricsNeedAToken(t *testing.T) {
	s, err := New(t.TempDir(), Options{AdminEmail: "admin@test.local", AdminPassword: "password123", MetricsToken: "scrape-me", AlertInterval: time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	do := apiClient(t, s)
	_, b := do("GET", "/api/me", nil)
	var me struct{ Orgs []struct{ ID string } }
	json.Unmarshal(b, &me)
	org := me.Orgs[0].ID

	h := newHook(t)
	code, raw := do("POST", "/api/orgs/"+org+"/alerts/channels", map[string]any{"name": "ops", "kind": "webhook", "target": h.URL, "format": "slack", "allowPrivate": true, "enabled": true})
	if code != 200 {
		t.Fatalf("create channel: %d %s", code, raw)
	}
	_, raw = do("GET", "/api/orgs/"+org+"/alerts", nil)
	if strings.Contains(string(raw), "secret-token") {
		t.Errorf("the overview leaks the webhook URL: %s", raw)
	}
	var ov struct {
		Channels []struct{ ID, Preview string }
	}
	json.Unmarshal(raw, &ov)
	if len(ov.Channels) != 1 || !strings.HasPrefix(ov.Channels[0].Preview, "http://127.0.0.1") {
		t.Fatalf("overview: %s", raw)
	}
	id := ov.Channels[0].ID

	// editing without re-entering the URL keeps it
	if code, raw := do("PUT", "/api/orgs/"+org+"/alerts/channels/"+id, map[string]any{"name": "ops renamed", "kind": "webhook", "format": "discord", "allowPrivate": true, "enabled": true}); code != 200 {
		t.Fatalf("edit: %d %s", code, raw)
	}
	if code, raw := do("POST", "/api/orgs/"+org+"/alerts/channels/"+id+"/test", nil); code != 200 {
		t.Fatalf("test send: %d %s", code, raw)
	}
	if body := h.next(2 * time.Second); !strings.HasPrefix(body, `{"content":`) {
		t.Errorf("the edit kept the old format or lost the URL: %q", body)
	}
	if code, _ := do("PUT", "/api/orgs/"+org+"/alerts/settings", map[string]int{"debounceMin": 0, "repeatHours": 1}); code != 400 {
		t.Error("a zero-minute debounce must be rejected")
	}
	if code, _ := do("PUT", "/api/orgs/"+org+"/alerts/settings", map[string]int{"debounceMin": 5, "repeatHours": 6}); code != 200 {
		t.Error("valid settings rejected")
	}
	if code, _ := do("DELETE", "/api/orgs/"+org+"/alerts/channels/"+id, nil); code != 200 {
		t.Error("delete")
	}

	// metrics
	ts := httptest.NewServer(s.Handler())
	defer ts.Close()
	get := func(token string) (int, string) {
		req, _ := http.NewRequest("GET", ts.URL+"/metrics", nil)
		if token != "" {
			req.Header.Set("Authorization", "Bearer "+token)
		}
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		defer resp.Body.Close()
		b, _ := io.ReadAll(resp.Body)
		return resp.StatusCode, string(b)
	}
	if code, _ := get(""); code != 401 {
		t.Errorf("metrics without a token: %d", code)
	}
	if code, _ := get("wrong"); code != 401 {
		t.Errorf("metrics with a wrong token: %d", code)
	}
	code, body := get("scrape-me")
	if code != 200 || !strings.Contains(body, "# TYPE hammerhead_users gauge") || !strings.Contains(body, "hammerhead_users 1") {
		t.Errorf("metrics: %d %s", code, body)
	}
	off, err := New(t.TempDir(), Options{AdminEmail: "a@test.local", AdminPassword: "password123", AlertInterval: time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	defer off.Close()
	ts2 := httptest.NewServer(off.Handler())
	defer ts2.Close()
	if resp, _ := http.Get(ts2.URL + "/metrics"); resp.StatusCode != 404 {
		t.Errorf("metrics should not exist unless a token is configured: %d", resp.StatusCode)
	}
}

func TestPromEscape(t *testing.T) {
	if got := promEscape("a\"b\\c\nd"); got != `a\"b\\c\nd` {
		t.Errorf("promEscape = %q", got)
	}
}