fix(backend): rate-limit and cap report ingest, drop wildcard CORS (closes #20)
check / check (push) Successful in 42s
check / check (push) Successful in 42s
POST /api/v1/reports stays unauthenticated but is bounded. Each client address, as the trusted-proxy logic resolves it, may send REPORTS_PER_MINUTE reports a minute (default 60, counted by go-chi/httprate over a sliding minute); past that it gets 429 with Retry-After. reportbuf refuses a report that would take the report files past DATA_DIR_MAX_BYTES (default 1 GiB), counting the files already in DATA_DIR and unwritten reports at their uncompressed size; the handler answers 507. CORS adds nothing unless CORS_ALLOWED_ORIGINS lists origins. A limit that is not a positive number, or an origin that is not a plain scheme://host[:port], stops the server from starting. Model: opus-5-5
This commit is contained in:
@@ -1,11 +1,17 @@
|
||||
package reportbuf_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/netwatch/internal/config"
|
||||
"sneak.berlin/go/netwatch/internal/globals"
|
||||
@@ -94,6 +100,253 @@ func TestFailedFinalFlushFailsStop(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// startBuffer starts a Buffer through fx, as main does, with the
|
||||
// DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set.
|
||||
func startBuffer(t *testing.T) *reportbuf.Buffer {
|
||||
t.Helper()
|
||||
|
||||
var buf *reportbuf.Buffer
|
||||
|
||||
app := fxtest.New(t,
|
||||
fx.Provide(
|
||||
globals.New,
|
||||
logger.New,
|
||||
config.New,
|
||||
reportbuf.New,
|
||||
),
|
||||
fx.Populate(&buf),
|
||||
)
|
||||
|
||||
app.RequireStart()
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
return buf
|
||||
}
|
||||
|
||||
// lineBytes is what one report takes in the buffer: its JSON and a
|
||||
// newline.
|
||||
func lineBytes(t *testing.T, report any) int {
|
||||
t.Helper()
|
||||
|
||||
line, err := json.Marshal(report)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal report: %v", err)
|
||||
}
|
||||
|
||||
return len(line) + 1
|
||||
}
|
||||
|
||||
func TestAppendPastCapIsRefused(t *testing.T) {
|
||||
report := map[string]string{"id": "cap"}
|
||||
|
||||
t.Setenv("DATA_DIR", t.TempDir())
|
||||
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))
|
||||
|
||||
buf := startBuffer(t)
|
||||
|
||||
err := buf.Append(report)
|
||||
if err != nil {
|
||||
t.Fatalf("report that fills the cap exactly: %v", err)
|
||||
}
|
||||
|
||||
err = buf.Append(report)
|
||||
if !errors.Is(err, reportbuf.ErrFull) {
|
||||
t.Fatalf("report past the cap: error = %v, want ErrFull", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCapCountsReportFilesAlreadyInDataDir starts on a data
|
||||
// directory holding a report file from an earlier run, and a file
|
||||
// that is not a report, which must not count.
|
||||
func TestCapCountsReportFilesAlreadyInDataDir(t *testing.T) {
|
||||
const earlierBytes = 100
|
||||
|
||||
report := map[string]string{"id": "cap"}
|
||||
dir := t.TempDir()
|
||||
|
||||
writeBytes(t, filepath.Join(dir, "reports-2026-01-01T00-00-00.000Z.jsonl.zst"),
|
||||
earlierBytes)
|
||||
writeBytes(t, filepath.Join(dir, "notes.txt"), 10*earlierBytes)
|
||||
|
||||
t.Setenv("DATA_DIR", dir)
|
||||
t.Setenv("DATA_DIR_MAX_BYTES",
|
||||
strconv.Itoa(earlierBytes+lineBytes(t, report)))
|
||||
|
||||
buf := startBuffer(t)
|
||||
|
||||
err := buf.Append(report)
|
||||
if err != nil {
|
||||
t.Fatalf("report that fills the cap exactly: %v", err)
|
||||
}
|
||||
|
||||
err = buf.Append(report)
|
||||
if !errors.Is(err, reportbuf.ErrFull) {
|
||||
t.Fatalf("report past the cap: error = %v, want ErrFull", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWrittenReportsCountAtFileSize checks that once reports are
|
||||
// written, they count as their compressed file, not their
|
||||
// uncompressed size, which frees room under the cap.
|
||||
func TestWrittenReportsCountAtFileSize(t *testing.T) {
|
||||
// Repetitive, so its file is far smaller than its JSON.
|
||||
report := map[string]string{"id": strings.Repeat("a", 1000)}
|
||||
size := lineBytes(t, report)
|
||||
|
||||
t.Setenv("DATA_DIR", t.TempDir())
|
||||
// Room for the report twice over only if the first one counts
|
||||
// at its file's size by the time the second arrives.
|
||||
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*size-1))
|
||||
|
||||
buf := startBuffer(t)
|
||||
|
||||
err := buf.Append(report)
|
||||
if err != nil {
|
||||
t.Fatalf("first report: %v", err)
|
||||
}
|
||||
|
||||
err = buf.Flush()
|
||||
if err != nil {
|
||||
t.Fatalf("flush: %v", err)
|
||||
}
|
||||
|
||||
// The second report is written at shutdown, and must not land in
|
||||
// the first file's millisecond (see TestWrittenReportsKeepCounting).
|
||||
time.Sleep(time.Millisecond)
|
||||
|
||||
err = buf.Append(report)
|
||||
if err != nil {
|
||||
t.Fatalf("second report, after the first was written: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWrittenReportsKeepCounting writes one report file after another
|
||||
// under a small cap: each report must be taken while the files on disk
|
||||
// leave room for it, and refused once they do not.
|
||||
func TestWrittenReportsKeepCounting(t *testing.T) {
|
||||
const maxBytes = 200
|
||||
|
||||
report := map[string]string{"id": "written"}
|
||||
size := int64(lineBytes(t, report))
|
||||
dir := t.TempDir()
|
||||
|
||||
t.Setenv("DATA_DIR", dir)
|
||||
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(maxBytes))
|
||||
|
||||
buf := startBuffer(t)
|
||||
|
||||
// Every file takes at least a byte, so they fill the cap within
|
||||
// maxBytes rounds.
|
||||
for range maxBytes {
|
||||
used := reportFilesBytes(t, dir)
|
||||
|
||||
err := buf.Append(report)
|
||||
if used+size > maxBytes {
|
||||
if !errors.Is(err, reportbuf.ErrFull) {
|
||||
t.Fatalf("with %d bytes of report files: error = %v, "+
|
||||
"want ErrFull", used, err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("with %d bytes of report files: %v", used, err)
|
||||
}
|
||||
|
||||
// Report files are named to the millisecond; two in the same
|
||||
// one collide (https://git.eeqj.de/sneak/netwatch/issues/61).
|
||||
time.Sleep(time.Millisecond)
|
||||
|
||||
err = buf.Flush()
|
||||
if err != nil {
|
||||
t.Fatalf("flush: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
t.Fatal("the report files never filled the cap")
|
||||
}
|
||||
|
||||
// TestConcurrentAppendsStopAtCap appends from many goroutines at once
|
||||
// with room for exactly roomFor reports: exactly that many must be
|
||||
// taken, which holds only if Append checks and counts each report
|
||||
// under one lock.
|
||||
func TestConcurrentAppendsStopAtCap(t *testing.T) {
|
||||
const (
|
||||
roomFor = 5
|
||||
senders = 50
|
||||
)
|
||||
|
||||
// Large, so each Append takes long enough for the senders to
|
||||
// overlap while the cap is reached.
|
||||
report := map[string]string{"id": strings.Repeat("a", 1_000_000)}
|
||||
|
||||
t.Setenv("DATA_DIR", t.TempDir())
|
||||
t.Setenv("DATA_DIR_MAX_BYTES",
|
||||
strconv.Itoa(roomFor*lineBytes(t, report)))
|
||||
|
||||
buf := startBuffer(t)
|
||||
|
||||
var (
|
||||
taken atomic.Int64
|
||||
wg sync.WaitGroup
|
||||
)
|
||||
|
||||
start := make(chan struct{})
|
||||
|
||||
for range senders {
|
||||
wg.Go(func() {
|
||||
<-start
|
||||
|
||||
err := buf.Append(report)
|
||||
if err == nil {
|
||||
taken.Add(1)
|
||||
} else if !errors.Is(err, reportbuf.ErrFull) {
|
||||
t.Errorf("append: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
close(start)
|
||||
wg.Wait()
|
||||
|
||||
if got := taken.Load(); got != roomFor {
|
||||
t.Fatalf("%d reports taken, want %d", got, roomFor)
|
||||
}
|
||||
}
|
||||
|
||||
// reportFilesBytes returns the total size of the report files in dir.
|
||||
func reportFilesBytes(t *testing.T, dir string) int64 {
|
||||
t.Helper()
|
||||
|
||||
paths, err := filepath.Glob(filepath.Join(dir, "reports-*.jsonl.zst"))
|
||||
if err != nil {
|
||||
t.Fatalf("list report files: %v", err)
|
||||
}
|
||||
|
||||
var total int64
|
||||
|
||||
for _, path := range paths {
|
||||
info, statErr := os.Stat(path)
|
||||
if statErr != nil {
|
||||
t.Fatalf("stat %s: %v", path, statErr)
|
||||
}
|
||||
|
||||
total += info.Size()
|
||||
}
|
||||
|
||||
return total
|
||||
}
|
||||
|
||||
func writeBytes(t *testing.T, path string, n int) {
|
||||
t.Helper()
|
||||
|
||||
err := os.WriteFile(path, make([]byte, n), 0o600)
|
||||
if err != nil {
|
||||
t.Fatalf("write %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
|
||||
func hasReportFile(t *testing.T, dir string) bool {
|
||||
t.Helper()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user