Author SHA1 Message Date
sneak e5d708cefa fix(backend): report ingest correctness — propagate storage failure, 413 on oversize, global body cap (closes #23)
check / check (push) Successful in 45s
A buffer failure on POST /api/v1/reports now returns 500 instead of a
false `ok`, so clients can retry. Decode errors split: an over-limit
body returns 413 (via errors.As on `*http.MaxBytesError`), malformed
JSON stays 400. A new MaxBodyBytes middleware (1 MiB default) caps
every route — rejecting an oversized Content-Length up front and
capping the read otherwise — so the health check and future routes are
bounded too. The raw attacker-controlled geo blob is no longer logged,
only its length; client_id and timestamp are length-bounded before
logging. A decodeJSON handler helper is added. Panic recovery is now a
local middleware routing the stack through slog as structured JSON.
Storage failure uses 500: a full buffer or write error is server-side
and retryable.

Model: opus-4-8
2026-09-21 17:02:37 +00:00
9 changed files with 433 additions and 33 deletions
+7
View File
@@ -22,6 +22,13 @@ files, so merging it also closes most compliance gaps.
# Completed Steps # Completed Steps
- 2026-09-21: report ingest correctness (issue #23): a storage failure now
returns 500 instead of a false `ok`; oversize bodies return 413 (distinguished
from malformed JSON, which stays 400); a `MaxBodyBytes` middleware caps every
route, not just the report route; the raw attacker-controlled `geo` blob is no
longer logged (only its length) and `client_id`/`timestamp` are length-bounded
before logging; a `decodeJSON` handler helper was added; and panic recovery
now routes the stack through slog instead of chi's plain-text stderr
- 2026-09-21: shutdown lifecycle correctness. The process now shuts down through - 2026-09-21: shutdown lifecycle correctness. The process now shuts down through
fx instead of `os.Exit`, so every component's `OnStop` runs and buffered fx instead of `os.Exit`, so every component's `OnStop` runs and buffered
reports are flushed to disk on `SIGTERM` — previously a full flush window of reports are flushed to disk on `SIGTERM` — previously a full flush window of
+10
View File
@@ -0,0 +1,10 @@
package handlers
import "log/slog"
// NewForTest builds a Handlers around a report sink and logger,
// bypassing the fx graph so handler behaviour (including the
// storage failure path) is exercisable in unit tests.
func NewForTest(buf reportAppender, log *slog.Logger) *Handlers {
return &Handlers{buf: buf, log: log}
}
+20 -1
View File
@@ -18,6 +18,13 @@ import (
const jsonContentType = "application/json; charset=utf-8" const jsonContentType = "application/json; charset=utf-8"
// reportAppender is the subset of the report buffer the handlers
// depend on. Defining it here keeps the storage failure path
// exercisable with a stub in tests.
type reportAppender interface {
Append(v any) error
}
// Params defines the dependencies for Handlers. // Params defines the dependencies for Handlers.
type Params struct { type Params struct {
fx.In fx.In
@@ -30,7 +37,7 @@ type Params struct {
// Handlers provides HTTP handler factories for all endpoints. // Handlers provides HTTP handler factories for all endpoints.
type Handlers struct { type Handlers struct {
buf *reportbuf.Buffer buf reportAppender
hc *healthcheck.Healthcheck hc *healthcheck.Healthcheck
log *slog.Logger log *slog.Logger
params *Params params *Params
@@ -72,3 +79,15 @@ func (s *Handlers) respondJSON(
} }
} }
} }
// decodeJSON decodes the request body into v. The body is
// expected to already be bounded by the body-size middleware, so
// a caller can distinguish an over-limit body from malformed
// JSON by testing the returned error for *http.MaxBytesError.
func (s *Handlers) decodeJSON(
_ http.ResponseWriter,
r *http.Request,
v any,
) error {
return json.NewDecoder(r.Body).Decode(v)
}
+62 -30
View File
@@ -2,10 +2,14 @@ package handlers
import ( import (
"encoding/json" "encoding/json"
"errors"
"net/http" "net/http"
) )
const maxReportBodyBytes = 1 << 20 // 1 MiB // maxLoggedFieldBytes bounds untrusted string fields before they
// are logged, so a caller cannot inflate log volume with an
// oversized value.
const maxLoggedFieldBytes = 128
type reportSample struct { type reportSample struct {
T int64 `json:"t"` T int64 `json:"t"`
@@ -35,48 +39,76 @@ func (s *Handlers) HandleReport() http.HandlerFunc {
} }
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
r.Body = http.MaxBytesReader(
w, r.Body, maxReportBodyBytes,
)
var rpt report var rpt report
err := json.NewDecoder(r.Body).Decode(&rpt) err := s.decodeJSON(w, r, &rpt)
if err != nil { if err != nil {
s.log.Error("failed to decode report",
"error", err,
)
s.respondJSON(w, r, s.respondJSON(w, r,
&response{Status: "error"}, &response{Status: "error"},
http.StatusBadRequest, s.decodeErrorStatus(err),
) )
return return
} }
totalSamples := 0 s.logReportReceived(rpt)
for _, h := range rpt.Hosts {
totalSamples += len(h.History)
}
s.log.Info("report received", err = s.buf.Append(rpt)
"client_id", rpt.ClientID, if err != nil {
"timestamp", rpt.Timestamp, s.log.Error("failed to buffer report", "error", err)
"host_count", len(rpt.Hosts), s.respondJSON(w, r,
"total_samples", totalSamples, &response{Status: "error"},
"geo", string(rpt.Geo), http.StatusInternalServerError,
)
bufErr := s.buf.Append(rpt)
if bufErr != nil {
s.log.Error("failed to buffer report",
"error", bufErr,
) )
return
} }
s.respondJSON(w, r, s.respondJSON(w, r, &response{Status: "ok"}, http.StatusOK)
&response{Status: "ok"},
http.StatusOK,
)
} }
} }
// decodeErrorStatus logs a report decode failure and returns the
// status to send: 413 when the body exceeded the size limit,
// otherwise 400 for malformed JSON.
func (s *Handlers) decodeErrorStatus(err error) int {
var tooLarge *http.MaxBytesError
if errors.As(err, &tooLarge) {
s.log.Warn("report body too large", "limit_bytes", tooLarge.Limit)
return http.StatusRequestEntityTooLarge
}
s.log.Error("failed to decode report", "error", err)
return http.StatusBadRequest
}
// logReportReceived logs an accepted report. Untrusted fields are
// bounded (client_id, timestamp) or reduced to a length
// (geo_bytes) so the raw attacker-controlled body never reaches
// the log.
func (s *Handlers) logReportReceived(rpt report) {
totalSamples := 0
for _, h := range rpt.Hosts {
totalSamples += len(h.History)
}
s.log.Info("report received",
"client_id", boundedForLog(rpt.ClientID),
"timestamp", boundedForLog(rpt.Timestamp),
"host_count", len(rpt.Hosts),
"total_samples", totalSamples,
"geo_bytes", len(rpt.Geo),
)
}
// boundedForLog truncates an untrusted string to a fixed byte
// bound so an attacker-controlled field cannot dominate the log.
func boundedForLog(s string) string {
if len(s) > maxLoggedFieldBytes {
return s[:maxLoggedFieldBytes]
}
return s
}
+141
View File
@@ -0,0 +1,141 @@
package handlers_test
import (
"bytes"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"testing"
"sneak.berlin/go/netwatch/internal/handlers"
)
var errStorageFailed = errors.New("storage failed")
// stubAppender drives the storage success/failure path without a
// real buffer or disk.
type stubAppender struct {
err error
}
func (s stubAppender) Append(any) error { return s.err }
func newTestHandlers(buf stubAppender, out io.Writer) *handlers.Handlers {
return handlers.NewForTest(buf, slog.New(slog.NewJSONHandler(out, nil)))
}
func decodeStatus(t *testing.T, body []byte) string {
t.Helper()
var resp struct {
Status string `json:"status"`
}
err := json.Unmarshal(body, &resp)
if err != nil {
t.Fatalf("response body not JSON: %v (%q)", err, body)
}
return resp.Status
}
func TestHandleReportStorageFailureIsNon2xx(t *testing.T) {
t.Parallel()
h := newTestHandlers(stubAppender{err: errStorageFailed}, io.Discard)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/api/v1/reports",
strings.NewReader(`{"clientId":"c1","hosts":[]}`),
)
h.HandleReport().ServeHTTP(rec, req)
if rec.Code < 500 {
t.Fatalf("storage failure status = %d, want a 5xx", rec.Code)
}
if got := decodeStatus(t, rec.Body.Bytes()); got != "error" {
t.Fatalf("status field = %q, want %q", got, "error")
}
}
func TestHandleReportMalformedJSONIs400(t *testing.T) {
t.Parallel()
h := newTestHandlers(stubAppender{}, io.Discard)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/api/v1/reports",
strings.NewReader(`{not json`),
)
h.HandleReport().ServeHTTP(rec, req)
if rec.Code != http.StatusBadRequest {
t.Fatalf("malformed status = %d, want %d",
rec.Code, http.StatusBadRequest)
}
}
func TestHandleReportOversizeIs413(t *testing.T) {
t.Parallel()
const limit = 32
h := newTestHandlers(stubAppender{}, io.Discard)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/api/v1/reports",
strings.NewReader(`{"clientId":"`+strings.Repeat("x", 200)+`"}`),
)
// Emulate the body-size middleware capping the body so the
// handler observes a *http.MaxBytesError while decoding.
req.Body = http.MaxBytesReader(rec, req.Body, limit)
h.HandleReport().ServeHTTP(rec, req)
if rec.Code != http.StatusRequestEntityTooLarge {
t.Fatalf("oversize status = %d, want %d",
rec.Code, http.StatusRequestEntityTooLarge)
}
}
func TestHandleReportDoesNotLogRawGeo(t *testing.T) {
t.Parallel()
const sentinel = "SENSITIVE-GEO-BLOB"
var logbuf bytes.Buffer
h := newTestHandlers(stubAppender{}, &logbuf)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/api/v1/reports",
strings.NewReader(
`{"clientId":"c1","geo":{"raw":"`+sentinel+`"},"hosts":[]}`,
),
)
h.HandleReport().ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK)
}
if strings.Contains(logbuf.String(), sentinel) {
t.Fatal("raw geo bytes were written to the log")
}
if !strings.Contains(logbuf.String(), "geo_bytes") {
t.Fatal("expected a bounded geo_bytes field in the log")
}
}
@@ -1,6 +1,7 @@
package middleware package middleware
import ( import (
"log/slog"
"net/http" "net/http"
"net/netip" "net/netip"
) )
@@ -8,6 +9,12 @@ import (
// Test-only wrappers exposing unexported helpers to the // Test-only wrappers exposing unexported helpers to the
// external middleware_test package. // external middleware_test package.
// NewWithLogger builds a Middleware around a logger for tests
// that exercise the logging paths without the fx graph.
func NewWithLogger(log *slog.Logger) *Middleware {
return &Middleware{log: log}
}
func ClientIP( func ClientIP(
remoteAddr string, remoteAddr string,
header http.Header, header http.Header,
+84
View File
@@ -3,11 +3,14 @@
package middleware package middleware
import ( import (
"errors"
"fmt" "fmt"
"io"
"log/slog" "log/slog"
"net" "net"
"net/http" "net/http"
"net/netip" "net/netip"
"runtime/debug"
"strings" "strings"
"time" "time"
@@ -22,6 +25,14 @@ import (
const corsMaxAgeSec = 300 const corsMaxAgeSec = 300
// jsonErrorBody is the body written for errors raised inside
// middleware, matching the {"status":"error"} shape the handlers
// return so clients see one error contract across the API.
const (
jsonContentType = "application/json; charset=utf-8"
jsonErrorBody = "{\"status\":\"error\"}\n"
)
// Security header values. The backend is a JSON API with no // Security header values. The backend is a JSON API with no
// HTML surface, so the CSP forbids every resource type and // HTML surface, so the CSP forbids every resource type and
// framing outright. // framing outright.
@@ -236,6 +247,79 @@ func (s *Middleware) SecurityHeaders() func(http.Handler) http.Handler {
} }
} }
// writeJSONError writes the shared JSON error body with the
// given status. Used where middleware must reject a request
// before it reaches a handler.
func writeJSONError(w http.ResponseWriter, status int) {
w.Header().Set("Content-Type", jsonContentType)
w.WriteHeader(status)
_, _ = io.WriteString(w, jsonErrorBody)
}
// MaxBodyBytes returns middleware that caps the request body at
// limit bytes. A declared Content-Length over the limit is
// rejected immediately with 413. Bodies without a declared
// length (or that understate it) are capped as they are read, so
// a handler that reads the body sees a *http.MaxBytesError it can
// map to 413. Mount it with a different limit on a route group
// that needs a different bound.
func (s *Middleware) MaxBodyBytes(
limit int64,
) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
if r.ContentLength > limit {
writeJSONError(
w,
http.StatusRequestEntityTooLarge,
)
return
}
r.Body = http.MaxBytesReader(w, r.Body, limit)
next.ServeHTTP(w, r)
},
)
}
}
// Recoverer returns middleware that recovers from a panic in a
// downstream handler, logs the panic and stack trace through
// slog, and responds 500 with no body. http.ErrAbortHandler is
// re-panicked so the server can abort the response as intended.
func (s *Middleware) Recoverer() func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
defer func() {
rec := recover()
if rec == nil {
return
}
err, ok := rec.(error)
if ok && errors.Is(err, http.ErrAbortHandler) {
panic(rec)
}
s.log.ErrorContext(r.Context(),
"panic recovered",
"panic", fmt.Sprintf("%v", rec),
"stack", string(debug.Stack()),
)
w.WriteHeader(http.StatusInternalServerError)
}()
next.ServeHTTP(w, r)
},
)
}
}
// CORS returns middleware that adds permissive CORS headers. // CORS returns middleware that adds permissive CORS headers.
func (s *Middleware) CORS() func(http.Handler) http.Handler { func (s *Middleware) CORS() func(http.Handler) http.Handler {
return cors.Handler(cors.Options{ return cors.Handler(cors.Options{
@@ -1,9 +1,12 @@
package middleware_test package middleware_test
import ( import (
"bytes"
"log/slog"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"net/netip" "net/netip"
"strings"
"testing" "testing"
"sneak.berlin/go/netwatch/internal/middleware" "sneak.berlin/go/netwatch/internal/middleware"
@@ -137,3 +140,92 @@ func TestSecurityHeaders(t *testing.T) {
} }
} }
} }
// TestMaxBodyBytesRejectsOversizeOnNonReadingRoute confirms the
// limit is enforced even for a handler that never reads the body
// (for example the health check), via the Content-Length check.
func TestMaxBodyBytesRejectsOversizeOnNonReadingRoute(t *testing.T) {
t.Parallel()
const limit = 16
called := false
handler := (&middleware.Middleware{}).MaxBodyBytes(limit)(
http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
called = true
}),
)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/.well-known/healthcheck",
strings.NewReader(strings.Repeat("x", limit+1)),
)
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusRequestEntityTooLarge {
t.Fatalf("status = %d, want %d",
rec.Code, http.StatusRequestEntityTooLarge)
}
if called {
t.Fatal("handler ran despite oversize body")
}
}
func TestMaxBodyBytesAllowsWithinLimit(t *testing.T) {
t.Parallel()
const limit = 64
handler := (&middleware.Middleware{}).MaxBodyBytes(limit)(
http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
}),
)
rec := httptest.NewRecorder()
req := httptest.NewRequest(
http.MethodPost, "/api/v1/reports",
strings.NewReader(`{"clientId":"c1"}`),
)
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK)
}
}
func TestRecovererReturns500AndLogsThroughSlog(t *testing.T) {
t.Parallel()
var logbuf bytes.Buffer
mw := middleware.NewWithLogger(
slog.New(slog.NewJSONHandler(&logbuf, nil)),
)
handler := mw.Recoverer()(
http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
panic("boom")
}),
)
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/", http.NoBody)
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, want %d",
rec.Code, http.StatusInternalServerError)
}
out := logbuf.String()
if !strings.Contains(out, "panic recovered") {
t.Fatalf("panic was not logged through slog: %q", out)
}
if !strings.Contains(out, `"level":"ERROR"`) {
t.Fatalf("panic log was not structured JSON at error level: %q", out)
}
}
+10 -2
View File
@@ -7,18 +7,26 @@ import (
"github.com/go-chi/chi/v5/middleware" "github.com/go-chi/chi/v5/middleware"
) )
const requestTimeout = 60 * time.Second const (
requestTimeout = 60 * time.Second
// maxRequestBodyBytes caps every request body. A route that
// needs a different bound mounts s.mw.MaxBodyBytes with its
// own value on its group.
maxRequestBodyBytes int64 = 1 << 20 // 1 MiB
)
// SetupRoutes configures the chi router with middleware and // SetupRoutes configures the chi router with middleware and
// all application routes. // all application routes.
func (s *Server) SetupRoutes() { func (s *Server) SetupRoutes() {
s.router = chi.NewRouter() s.router = chi.NewRouter()
s.router.Use(middleware.Recoverer) s.router.Use(s.mw.Recoverer())
s.router.Use(middleware.RequestID) s.router.Use(middleware.RequestID)
s.router.Use(s.mw.Logging()) s.router.Use(s.mw.Logging())
s.router.Use(s.mw.SecurityHeaders()) s.router.Use(s.mw.SecurityHeaders())
s.router.Use(s.mw.CORS()) s.router.Use(s.mw.CORS())
s.router.Use(s.mw.MaxBodyBytes(maxRequestBodyBytes))
s.router.Use(middleware.Timeout(requestTimeout)) s.router.Use(middleware.Timeout(requestTimeout))
s.router.Get( s.router.Get(