unibus/pkg/membership/dos_test.go

package membership

import (
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"testing"

	"github.com/enmanuel/unibus/pkg/blobstore"
)

// dosServer builds a Server backed by a fresh store + blob store so a test can
// drive ServeHTTP in-process (white-box) and observe its memory behavior without
// a network round trip — the same in-process technique the auditor used.
func dosServer(t *testing.T, mode AuthMode) *Server {
	t.Helper()
	dir := t.TempDir()
	store, err := Open(filepath.Join(dir, "unibus.db"))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	blobs, err := blobstore.New(filepath.Join(dir, "blobs"))
	if err != nil {
		t.Fatalf("open blobs: %v", err)
	}
	t.Cleanup(func() { store.Close() })
	return NewServer(store, blobs, mode)
}

// zeroReader yields up to remaining zero bytes without ever allocating them, so
// the test process itself never materializes a huge buffer (which would taint the
// RSS measurement we are trying to make about the SERVER).
type zeroReader struct{ remaining int64 }

func (z *zeroReader) Read(p []byte) (int, error) {
	if z.remaining <= 0 {
		return 0, io.EOF
	}
	n := int64(len(p))
	if n > z.remaining {
		n = z.remaining
	}
	for i := int64(0); i < n; i++ {
		p[i] = 0
	}
	z.remaining -= n
	return int(n), nil
}

// vmRSSkB reads the resident set size (kB) of this process from /proc. Linux-only;
// the caller skips on other platforms.
func vmRSSkB(t *testing.T) int64 {
	t.Helper()
	b, err := os.ReadFile("/proc/self/status")
	if err != nil {
		t.Skipf("cannot read /proc/self/status: %v", err)
	}
	for _, line := range strings.Split(string(b), "\n") {
		if strings.HasPrefix(line, "VmRSS:") {
			f := strings.Fields(line)
			if len(f) >= 2 {
				v, _ := strconv.ParseInt(f[1], 10, 64)
				return v
			}
		}
	}
	t.Skip("VmRSS not present in /proc/self/status")
	return 0
}

// TestAudit_DoSBodyLimitNoAuth ports the auditor's H1 (Critical) vector: a peer
// with NO valid signature posts an oversized body. Before the fix the middleware
// io.ReadAll'd it unbounded (the auditor sent 400 MB and watched RSS jump from
// 18 MB to 898 MB). Now the request is rejected 413 and the resident set does NOT
// spike. Two shapes are covered:
//
//	(1) a truthful, over-ceiling Content-Length  -> rejected before any byte is read;
//	(2) a lying / unknown length (chunked)       -> MaxBytesReader trips mid-read,
//	    capping the buffered bytes at the ceiling instead of the attacker's 400 MB.
func TestAudit_DoSBodyLimitNoAuth(t *testing.T) {
	if runtime.GOOS != "linux" {
		t.Skip("RSS probe is Linux-only")
	}
	srv := dosServer(t, AuthEnforce) // enforce: the request carries no signature

	const huge = int64(400) << 20 // 400 MiB — the auditor's figure
	// A spike threshold an order of magnitude below the attack. The old code would
	// add ~400 MB+; the fix keeps the delta to at most one bounded buffer.
	const maxSpikeKB = int64(96) << 10 // 96 MiB

	// Shape 1: declared Content-Length over the blob ceiling -> early 413, no read.
	runtime.GC()
	before := vmRSSkB(t)
	req := httptest.NewRequest(http.MethodPost, "/blobs", &zeroReader{remaining: huge})
	req.ContentLength = huge
	rec := httptest.NewRecorder()
	srv.ServeHTTP(rec, req)
	if rec.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("over-declared body should be 413, got %d", rec.Code)
	}
	runtime.GC()
	if d := vmRSSkB(t) - before; d > maxSpikeKB {
		t.Fatalf("RSS spiked %d kB on a pre-declared oversized body (limit %d kB)", d, maxSpikeKB)
	}

	// Shape 2: unknown length (chunked-style). The middleware cannot reject by
	// Content-Length, so MaxBytesReader must cap the read at maxBlobBytes.
	runtime.GC()
	before = vmRSSkB(t)
	req = httptest.NewRequest(http.MethodPost, "/blobs", &zeroReader{remaining: huge})
	req.ContentLength = -1
	rec = httptest.NewRecorder()
	srv.ServeHTTP(rec, req)
	if rec.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("unknown-length oversized body should be 413, got %d", rec.Code)
	}
	runtime.GC()
	if d := vmRSSkB(t) - before; d > maxSpikeKB {
		t.Fatalf("RSS spiked %d kB on a chunked oversized body (limit %d kB)", d, maxSpikeKB)
	}
}

// TestBlobLimitGoldenAndBoundary covers the golden path (a normal blob is stored)
// and the boundary (a body exactly at the ceiling is accepted; one byte over by
// truthful Content-Length is rejected before buffering).
func TestBlobLimitGoldenAndBoundary(t *testing.T) {
	srv := dosServer(t, AuthOff) // AuthOff: the limits apply regardless of auth mode

	// Golden: a small blob is accepted and hashed.
	rec := httptest.NewRecorder()
	srv.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/blobs", strings.NewReader("hello blob")))
	if rec.Code != http.StatusOK {
		t.Fatalf("normal blob should be 200, got %d (%s)", rec.Code, rec.Body.String())
	}

	// Boundary: exactly at the ceiling is allowed (MaxBytesReader permits N bytes).
	atLimit := strings.Repeat("a", maxBlobBytes)
	rec = httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/blobs", strings.NewReader(atLimit))
	req.ContentLength = int64(len(atLimit))
	srv.ServeHTTP(rec, req)
	if rec.Code != http.StatusOK {
		t.Fatalf("blob exactly at the ceiling should be 200, got %d", rec.Code)
	}

	// Error: one byte over the ceiling (truthful Content-Length) -> 413 pre-read.
	rec = httptest.NewRecorder()
	req = httptest.NewRequest(http.MethodPost, "/blobs", &zeroReader{remaining: maxBlobBytes + 1})
	req.ContentLength = maxBlobBytes + 1
	srv.ServeHTTP(rec, req)
	if rec.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("blob one byte over the ceiling should be 413, got %d", rec.Code)
	}
}

// TestControlBodyLimit checks the smaller JSON ceiling on a non-blob route: a body
// over maxControlBodyBytes is rejected 413 before the handler runs.
func TestControlBodyLimit(t *testing.T) {
	srv := dosServer(t, AuthOff)
	rec := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/rooms", &zeroReader{remaining: maxControlBodyBytes + 1})
	req.ContentLength = maxControlBodyBytes + 1
	srv.ServeHTTP(rec, req)
	if rec.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("control body over 1 MiB should be 413, got %d", rec.Code)
	}
}

// TestRateLimitPerIP exercises the per-IP throttle: a burst from one IP eventually
// gets 429 (error path), while a spread across distinct IPs is never throttled
// (edge — the bucket is keyed per source, not global).
func TestRateLimitPerIP(t *testing.T) {
	srv := dosServer(t, AuthOff)

	// Same IP: well past the burst -> at least one 429.
	got429 := false
	for i := 0; i < defaultRateBurst+50; i++ {
		rec := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/rooms/none", nil)
		req.RemoteAddr = "203.0.113.7:5555"
		srv.ServeHTTP(rec, req)
		if rec.Code == http.StatusTooManyRequests {
			got429 = true
			break
		}
	}
	if !got429 {
		t.Fatalf("a flood from one IP should eventually be rate-limited (429)")
	}

	// Distinct IPs: each gets a fresh bucket, so none is throttled.
	for i := 0; i < 100; i++ {
		rec := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/rooms/none", nil)
		req.RemoteAddr = "198.51.100." + strconv.Itoa(i%254+1) + ":4444"
		srv.ServeHTTP(rec, req)
		if rec.Code == http.StatusTooManyRequests {
			t.Fatalf("distinct IPs must not share a rate bucket; IP #%d got 429", i)
		}
	}
}