notify-gateway/internal/httpserver/operations_test.go

package httpserver

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/example/notify-gateway/internal/auth"
	"github.com/example/notify-gateway/internal/config"
	"github.com/example/notify-gateway/internal/divera"
	"github.com/example/notify-gateway/internal/gateway"
	"github.com/example/notify-gateway/internal/outbox"
)

func queuedServer(t *testing.T) (*Server, *config.Store, *outbox.Store) {
	t.Helper()
	dir := t.TempDir()
	store, err := config.Open(filepath.Join(dir, "config.json"))
	if err != nil {
		t.Fatal(err)
	}
	db, err := outbox.Open(filepath.Join(dir, "outbox.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { db.Close() })
	client := divera.New(func() config.DiveraConfig { return store.Get().Divera })
	dispatcher := gateway.New(store, client)
	s := New(store, dispatcher, client, log.New(io.Discard, "", 0))
	s.UseQueue(&gateway.Queue{Store: db, Dispatcher: dispatcher}, nil)
	return s, store, db
}
func TestAsyncIngressIdempotencyReadinessAndMetrics(t *testing.T) {
	s, store, db := queuedServer(t)
	c := store.Get()
	c.Ingress.WebhookTokens = map[string]string{"ops": "secret"}
	c.Mappings = []config.Mapping{{ID: "route", Enabled: true, Target: "news", TextTemplate: "{{.Message}}"}}
	store.Replace(c)
	h := s.Handler()
	send := func(body string) *httptest.ResponseRecorder {
		r := httptest.NewRequest("POST", "/in/webhook/ops", strings.NewReader(body))
		r.Header.Set("Authorization", "Bearer secret")
		r.Header.Set("Idempotency-Key", "same-request")
		w := httptest.NewRecorder()
		h.ServeHTTP(w, r)
		return w
	}
	for i := 0; i < 2; i++ {
		if w := send("hello"); w.Code != 202 {
			t.Fatalf("accepted=%d %s", w.Code, w.Body)
		}
	}
	if w := send("different"); w.Code != 409 {
		t.Fatal(w.Code)
	}
	rows, _ := db.List(context.Background(), 50, 0)
	if len(rows) != 1 || rows[0].State != "pending" {
		t.Fatal(rows)
	}
	for _, path := range []string{"/readyz", "/metrics"} {
		w := httptest.NewRecorder()
		h.ServeHTTP(w, httptest.NewRequest("GET", path, nil))
		if path == "/readyz" && w.Code != 200 {
			t.Fatal(w.Code)
		}
		if path == "/metrics" && w.Code != 303 {
			t.Fatal("public metrics")
		}
	}
	t.Setenv("GATEWAY_METRICS_TOKEN", "metric-token")
	r := httptest.NewRequest("GET", "/metrics", nil)
	r.Header.Set("Authorization", "Bearer metric-token")
	w := httptest.NewRecorder()
	h.ServeHTTP(w, r)
	if w.Code != 200 || !strings.Contains(w.Body.String(), `state="pending"} 1`) {
		t.Fatalf("metrics %d %s", w.Code, w.Body)
	}
}

func TestCSRFRequestLimitsAndLoginRateLimit(t *testing.T) {
	s, store, _ := queuedServer(t)
	c := store.Get()
	h := s.Handler()
	session := auth.SignSession(c.Server.SessionSecret, c.Server.AdminUsername, time.Now().Add(time.Hour))
	for _, valid := range []bool{false, true} {
		body, _ := json.Marshal(makeEditable(c))
		r := httptest.NewRequest("PUT", "/ui/api/config", bytes.NewReader(body))
		r.AddCookie(&http.Cookie{Name: "ng_session", Value: session})
		if valid {
			r.Header.Set("X-CSRF-Token", csrfToken(c.Server.SessionSecret, session))
		}
		w := httptest.NewRecorder()
		h.ServeHTTP(w, r)
		want := 403
		if valid {
			want = 200
		}
		if w.Code != want {
			t.Fatalf("CSRF valid=%v status=%d %s", valid, w.Code, w.Body)
		}
	}
	r := httptest.NewRequest("POST", "/login", nil)
	r.Header.Set("Origin", "https://evil.example")
	w := httptest.NewRecorder()
	h.ServeHTTP(w, r)
	if w.Code != 403 {
		t.Fatal("cross-site login accepted")
	}
	w = httptest.NewRecorder()
	h.ServeHTTP(w, httptest.NewRequest("POST", "/ntfy/ops", strings.NewReader(strings.Repeat("x", (1<<20)+1))))
	if w.Code != 413 {
		t.Fatal("large body accepted")
	}
	for i := 0; i < 11; i++ {
		w = httptest.NewRecorder()
		h.ServeHTTP(w, httptest.NewRequest("POST", "/login", nil))
	}
	if w.Code != 429 {
		t.Fatalf("rate limit %d", w.Code)
	}
}

func TestDiscordSignatureAllowlistAndReplay(t *testing.T) {
	s, store, db := queuedServer(t)
	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
	c := store.Get()
	c.Ingress.Discord = config.DiscordIngress{Enabled: true, PublicKey: hex.EncodeToString(pub), ApplicationID: "app", GuildIDs: []string{"guild"}, ChannelIDs: []string{"channel"}, Command: "notify"}
	c.Mappings = []config.Mapping{{ID: "discord", Enabled: true, Source: "discord", Target: "news", TextTemplate: "{{.Message}}"}}
	if err := store.Replace(c); err != nil {
		t.Fatal(err)
	}
	h := s.Handler()
	invoke := func(body string, stamp time.Time, valid bool) *httptest.ResponseRecorder {
		r := httptest.NewRequest("POST", "/in/discord", strings.NewReader(body))
		ts := strconv.FormatInt(stamp.Unix(), 10)
		sig := ed25519.Sign(priv, []byte(ts+body))
		if !valid {
			sig[0] ^= 1
		}
		r.Header.Set("X-Signature-Timestamp", ts)
		r.Header.Set("X-Signature-Ed25519", hex.EncodeToString(sig))
		w := httptest.NewRecorder()
		h.ServeHTTP(w, r)
		return w
	}
	ping := `{"type":1,"application_id":"app"}`
	if w := invoke(ping, time.Now(), true); w.Code != 200 || !strings.Contains(w.Body.String(), `"type":1`) {
		t.Fatalf("ping %d %s", w.Code, w.Body)
	}
	if w := invoke(ping, time.Now(), false); w.Code != 401 {
		t.Fatal("bad signature accepted")
	}
	if w := invoke(ping, time.Now().Add(-6*time.Minute), true); w.Code != 401 {
		t.Fatal("stale request accepted")
	}
	body := `{"id":"interaction","application_id":"app","type":2,"guild_id":"guild","channel_id":"channel","data":{"name":"notify","type":1,"options":[{"name":"message","type":3,"value":"hello"}]}}`
	for i := 0; i < 2; i++ {
		if w := invoke(body, time.Now(), true); w.Code != 200 {
			t.Fatalf("command %d %s", w.Code, w.Body)
		}
	}
	if w := invoke(strings.Replace(body, `"guild"`, `"other"`, 1), time.Now(), true); w.Code != 403 {
		t.Fatal("guild allowlist bypass")
	}
	rows, _ := db.List(context.Background(), 50, 0)
	if len(rows) != 1 {
		t.Fatalf("duplicate interaction: %+v", rows)
	}
}