feat(membership): bound request bodies and add per-IP rate limit

Pre-auth DoS hardening (audit H1, Critical). The control-plane middleware
read the request body with io.ReadAll before authenticating and with no size
cap, so an unauthenticated peer could force the server to buffer an arbitrary
body in RAM (the auditor sent 400 MB and watched RSS climb to ~898 MB).

- ServeHTTP now caps the buffered body before reading: a per-route ceiling
  (1 MiB JSON, 16 MiB /blobs) rejects an over-declared Content-Length outright
  and wraps the body in http.MaxBytesReader so a lying/chunked sender trips at
  the ceiling instead of unbounded.
- handlePutBlob maps the MaxBytesReader cutoff to 413 in every auth mode.
- Per-IP token-bucket rate limiter (golang.org/x/time/rate, already in the
  module graph) sheds floods before auth or body reads. Loopback dev stacks are
  unaffected (burst >> any single client's rate). Kept in-package as transport
  glue, not promoted to the registry, mirroring the nonceCache decision in 0003.
- membershipd sets http.Server.MaxHeaderBytes and ReadHeaderTimeout.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

cmd/membershipd/main.go

@@ -113,7 +113,15 @@ func main() {
 	srv := membership.NewServer(store, blobs, authMode)
 	log.Printf("control-plane auth: %s", authMode)
 	addr := *bind + ":" + *httpPort
-	httpSrv := &http.Server{Addr: addr, Handler: srv}
+	httpSrv := &http.Server{
+		Addr:    addr,
+		Handler: srv,
+		// Bound request header size so a peer cannot exhaust memory with huge
+		// headers before any body limit applies (the body ceilings live in the
+		// membership middleware).
+		MaxHeaderBytes:    membership.MaxHeaderBytes,
+		ReadHeaderTimeout: 10 * time.Second,
+	}
 
 	go func() {
 		log.Printf("HTTP control-plane API: http://%s", addr)

go.mod

@@ -8,7 +8,9 @@ require (
 	fn-registry v0.0.0-00010101000000-000000000000
 	github.com/nats-io/nats-server/v2 v2.10.22
 	github.com/nats-io/nats.go v1.37.0
+	github.com/nats-io/nkeys v0.4.7
 	github.com/oklog/ulid/v2 v2.1.0
+	golang.org/x/time v0.7.0
 	modernc.org/sqlite v1.47.0
 )
 
@@ -19,7 +21,6 @@ require (
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/minio/highwayhash v1.0.3 // indirect
 	github.com/nats-io/jwt/v2 v2.5.8 // indirect
-	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/ncruces/go-strftime v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
@@ -29,7 +30,6 @@ require (
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.44.0 // indirect
 	golang.org/x/text v0.37.0 // indirect
-	golang.org/x/time v0.7.0 // indirect
 	golang.org/x/tools v0.45.0 // indirect
 	modernc.org/libc v1.70.0 // indirect
 	modernc.org/mathutil v1.7.1 // indirect

go.sum

@@ -1,5 +1,7 @@
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -47,6 +49,10 @@ golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
 golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8=
 golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0=
+golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM=
+golang.org/x/tools/go/expect v0.1.1-deprecated/go.mod h1:eihoPOH+FgIqa3FpoTwguz/bVUSGBlGQU67vpBeOrBY=
+golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated h1:1h2MnaIAIXISqTFKdENegdpAgUXz6NrPEsbIeWaBRvM=
+golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated/go.mod h1:RVAQXBGNv1ib0J382/DPCRS/BPnsGebyM1Gj5VSDpG8=
 modernc.org/cc/v4 v4.27.1 h1:9W30zRlYrefrDV2JE2O8VDtJ1yPGownxciz5rrbQZis=
 modernc.org/cc/v4 v4.27.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0=
 modernc.org/ccgo/v4 v4.32.0 h1:hjG66bI/kqIPX1b2yT6fr/jt+QedtP2fqojG2VrFuVw=

pkg/membership/ratelimit.go

@@ -0,0 +1,93 @@
+package membership
+
+import (
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"golang.org/x/time/rate"
+)
+
+// ipRateLimiter is a per-source-IP token-bucket rate limiter for the control
+// plane. It exists to blunt pre-auth flooding: an unauthenticated peer that
+// hammers the HTTP API (signature verification is not free, and io is bounded
+// but still real) is throttled before it can amplify load. Like the nonceCache,
+// this is transport glue specific to unibus, not a registry primitive — the
+// report 0003 made the same call for the nonce cache (it would only drag a NATS
+// dependency into the multi-domain registry go.mod for one helper).
+//
+// Each distinct IP gets its own golang.org/x/time/rate.Limiter (a standard
+// token bucket already in the module graph, so no new dependency). Idle buckets
+// are reaped so the map cannot grow without bound under a churn of source IPs.
+type ipRateLimiter struct {
+	mu      sync.Mutex
+	buckets map[string]*ipBucket
+	r       rate.Limit
+	burst   int
+	ttl     time.Duration
+}
+
+type ipBucket struct {
+	lim  *rate.Limiter
+	seen time.Time
+}
+
+// newIPRateLimiter builds a limiter granting r tokens/second with the given
+// burst per IP. ttl bounds how long an idle bucket is retained before being
+// reaped. r<=0 disables limiting (Allow always true) so dev/loopback stacks are
+// unaffected.
+func newIPRateLimiter(r rate.Limit, burst int, ttl time.Duration) *ipRateLimiter {
+	return &ipRateLimiter{
+		buckets: make(map[string]*ipBucket),
+		r:       r,
+		burst:   burst,
+		ttl:     ttl,
+	}
+}
+
+// allow reports whether a request from ip may proceed now, consuming one token
+// on success. A disabled limiter (r<=0) always allows. Reaping of stale buckets
+// is amortized: it runs only when the map has grown past a small threshold, so
+// the common path is a single map lookup under the mutex.
+func (l *ipRateLimiter) allow(ip string, now time.Time) bool {
+	if l == nil || l.r <= 0 {
+		return true
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if len(l.buckets) > 1024 {
+		l.reapLocked(now)
+	}
+	b, ok := l.buckets[ip]
+	if !ok {
+		b = &ipBucket{lim: rate.NewLimiter(l.r, l.burst)}
+		l.buckets[ip] = b
+	}
+	b.seen = now
+	return b.lim.AllowN(now, 1)
+}
+
+// reapLocked drops buckets idle for longer than ttl. The caller holds l.mu.
+func (l *ipRateLimiter) reapLocked(now time.Time) {
+	for ip, b := range l.buckets {
+		if now.Sub(b.seen) > l.ttl {
+			delete(l.buckets, ip)
+		}
+	}
+}
+
+// clientIP extracts the source IP of an HTTP request, stripping the port. It
+// trusts the transport's RemoteAddr only (no X-Forwarded-For parsing): a public
+// deployment terminates TLS at this process or behind a proxy that the operator
+// controls, and honoring an attacker-supplied header would let a single IP fan
+// its quota across forged identities. If parsing fails the whole RemoteAddr is
+// used as the key (still a stable per-connection bucket).
+func clientIP(r *http.Request) string {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}

pkg/membership/server.go

@@ -15,9 +15,36 @@ import (
 
 	cs "fn-registry/functions/cybersecurity"
 
+	"golang.org/x/time/rate"
+
 	"github.com/enmanuel/unibus/pkg/blobstore"
 )
 
+// Body-size ceilings for the control plane. They bound how much an unauthenticated
+// peer can make the server buffer in RAM before the request is even authenticated
+// (the signature is verified over the full body, so the body must be read — but
+// not unboundedly). maxControlBodyBytes covers JSON metadata requests; /blobs gets
+// a separate, larger ceiling because media ciphertext is legitimately bigger. A
+// request whose declared Content-Length already exceeds its ceiling is rejected
+// before a single byte is buffered.
+const (
+	maxControlBodyBytes = 1 << 20  // 1 MiB for JSON control-plane requests
+	maxBlobBytes        = 16 << 20 // 16 MiB for a single media blob upload
+	// MaxHeaderBytes caps request header size; wired into the http.Server by the
+	// command. Exported so the bound lives next to its body-size siblings.
+	MaxHeaderBytes = 1 << 20 // 1 MiB
+)
+
+// Per-IP rate-limit defaults for the control plane. Tuned for an interactive
+// human/agent bus rather than a high-QPS API: a steady ~20 req/s with a burst of
+// 40 absorbs a chat client's bursty polling while throttling a flood. Loopback
+// dev stacks pass r<=0 to disable limiting entirely.
+const (
+	defaultRatePerSec = rate.Limit(20)
+	defaultRateBurst  = 40
+	rateBucketTTL     = 10 * time.Minute
+)
+
 // Server is the HTTP control plane: the authoritative source of room metadata,
 // membership, and per-epoch sealed keys. The data plane (messages) is NATS.
 //
@@ -32,11 +59,14 @@ type Server struct {
 	mux      *http.ServeMux
 	authMode AuthMode
 	nonces   *nonceCache
+	limiter  *ipRateLimiter
 }
 
 // NewServer wires the membership store and blob store into an http.Handler. The
 // authMode selects the control-plane auth rollout state (AuthOff for callers and
-// tests that have not migrated to signed requests yet).
+// tests that have not migrated to signed requests yet). It installs a per-IP
+// rate limiter with the package defaults; loopback dev behavior is unchanged
+// because the burst comfortably exceeds any single client's request rate.
 func NewServer(store *Store, blobs *blobstore.Store, authMode AuthMode) *Server {
 	s := &Server{
 		store:    store,
@@ -44,6 +74,7 @@ func NewServer(store *Store, blobs *blobstore.Store, authMode AuthMode) *Server
 		mux:      http.NewServeMux(),
 		authMode: authMode,
 		nonces:   newNonceCache(nonceTTL),
+		limiter:  newIPRateLimiter(defaultRatePerSec, defaultRateBurst, rateBucketTTL),
 	}
 	s.routes()
 	return s
@@ -53,23 +84,53 @@ func NewServer(store *Store, blobs *blobstore.Store, authMode AuthMode) *Server
 // (signature verification + anti-replay + allowlist) ahead of the router
 // according to authMode, then dispatches to the matched handler.
 func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	now := time.Now()
+
+	// Per-IP rate limit runs first, ahead of auth and body reads, so a flood is
+	// shed at the cheapest possible point. The health probe is exempt so liveness
+	// checks are never throttled.
+	if !isAuthExempt(r) && !s.limiter.allow(clientIP(r), now) {
+		writeErr(w, http.StatusTooManyRequests, "rate limit exceeded")
+		return
+	}
+
+	// Cap how much body we will buffer, BEFORE reading a single byte. The ceiling
+	// is per-route: /blobs may legitimately carry a media ciphertext, everything
+	// else is small JSON. A declared Content-Length over the ceiling is rejected
+	// outright (no buffering); MaxBytesReader then guards against a lying or
+	// chunked sender by failing the read once the limit is crossed. This is the
+	// fix for the pre-auth DoS: without it an unauthenticated peer could make the
+	// server buffer an unbounded body in RAM before authenticate() ever ran.
+	limit := int64(maxControlBodyBytes)
+	if r.Method == http.MethodPost && r.URL.Path == "/blobs" {
+		limit = int64(maxBlobBytes)
+	}
+	if r.ContentLength > limit {
+		writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
+		return
+	}
+	r.Body = http.MaxBytesReader(w, r.Body, limit)
+
 	if s.authMode == AuthOff || isAuthExempt(r) {
 		s.mux.ServeHTTP(w, r)
 		return
 	}
 
-	// Buffer the body so the signature can be verified over it and the handler
-	// still reads it. Bodies on the control plane are small (JSON metadata or a
-	// media blob already capped upstream), so full buffering is acceptable.
+	// Buffer the (now bounded) body so the signature can be verified over it and
+	// the handler still reads it.
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
-		writeErr(w, http.StatusBadRequest, "read body: "+err.Error())
+		if isBodyTooLarge(err) {
+			writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
+			return
+		}
+		writeErr(w, http.StatusBadRequest, "read body")
 		return
 	}
 	_ = r.Body.Close()
 	r.Body = io.NopCloser(bytes.NewReader(body))
 
-	if _, err := s.authenticate(r, body, time.Now()); err != nil {
+	if _, err := s.authenticate(r, body, now); err != nil {
 		if s.authMode == AuthSoft {
 			log.Printf("[auth] soft: would reject %s %s: %v", r.Method, r.URL.Path, err)
 			s.mux.ServeHTTP(w, r)
@@ -81,6 +142,13 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	s.mux.ServeHTTP(w, r)
 }
 
+// isBodyTooLarge reports whether err is the sentinel returned by MaxBytesReader
+// when the body exceeds its limit, so the middleware can map it to 413.
+func isBodyTooLarge(err error) bool {
+	var maxErr *http.MaxBytesError
+	return errors.As(err, &maxErr)
+}
+
 // isAuthExempt lists requests that bypass control-plane auth even under enforce.
 // Only the unauthenticated health probe qualifies: it carries no data and is
 // needed by load balancers / smoke checks / systemd before any identity exists.
@@ -401,9 +469,18 @@ func (s *Server) handleRekey(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) handlePutBlob(w http.ResponseWriter, r *http.Request) {
+	// The body arrives already bounded: ServeHTTP wraps it in a MaxBytesReader
+	// (maxBlobBytes) and rejects an over-declared Content-Length before this
+	// handler runs, in every auth mode. Reading here therefore cannot buffer
+	// more than the ceiling; a sender that lies about its length (e.g. chunked)
+	// trips MaxBytesReader and we map that to 413 rather than a generic 400.
 	data, err := io.ReadAll(r.Body)
 	if err != nil {
-		writeErr(w, http.StatusBadRequest, "read body: "+err.Error())
+		if isBodyTooLarge(err) {
+			writeErr(w, http.StatusRequestEntityTooLarge, "request body too large")
+			return
+		}
+		writeErr(w, http.StatusBadRequest, "read body")
 		return
 	}
 	hash, err := s.blobs.Put(data)