Classify oversized agent reports correctly

Change-source: pulse-maintainer

internal/api/agent_ingest.go

@@ -72,22 +72,10 @@ func (h *UnifiedAgentHandlers) HandleReport(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	// Limit request body to 256KB to prevent memory exhaustion
-	r.Body = http.MaxBytesReader(w, r.Body, 256*1024)
-	defer r.Body.Close()
-
-	// Support gzip-compressed reports from agents (backward compatible with uncompressed).
-	// Cap decompressed size at 1.5MB (6x compressed limit — generous for legitimate payloads).
-	body, err := utils.DecompressBodyIfGzipped(r, 1536*1024)
-	if err != nil {
-		writeErrorResponse(w, http.StatusUnsupportedMediaType, "unsupported_encoding", err.Error(), nil)
-		return
-	}
-	defer body.Close()
-
 	var report agentshost.Report
-	if err := json.NewDecoder(body).Decode(&report); err != nil {
-		writeErrorResponse(w, http.StatusBadRequest, "invalid_json", "Failed to decode request body", map[string]string{"error": err.Error()})
+	// Support gzip-compressed reports from agents (backward compatible with
+	// uncompressed), with independent encoded and decoded size limits.
+	if !decodeCompressedAgentReport(w, r, 256*1024, 1536*1024, &report) {
 		return
 	}
 

internal/api/agent_report_decode.go

@@ -0,0 +1,85 @@
+package api
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strconv"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+)
+
+const (
+	agentReportSizeEncodedBody = "encoded_http_body"
+	agentReportSizeDecodedJSON = "decoded_json"
+)
+
+// decodeCompressedAgentReport applies independent encoded and decoded body
+// limits and preserves the reason a report could not be decoded. Appliance
+// inventories can be large and compress well, so the two limits are distinct.
+func decodeCompressedAgentReport(
+	w http.ResponseWriter,
+	r *http.Request,
+	encodedLimit int64,
+	decodedLimit int64,
+	destination any,
+) bool {
+	r.Body = http.MaxBytesReader(w, r.Body, encodedLimit)
+	defer r.Body.Close()
+	decompressed, err := utils.DecompressBodyIfGzipped(r, decodedLimit)
+	if err != nil {
+		var maxBytesErr *http.MaxBytesError
+		var encodingErr *utils.UnsupportedContentEncodingError
+		switch {
+		case errors.As(err, &maxBytesErr):
+			writeAgentReportSizeError(w, agentReportSizeEncodedBody, maxBytesErr.Limit)
+		case errors.As(err, &encodingErr):
+			writeErrorResponse(w, http.StatusUnsupportedMediaType, "unsupported_encoding", encodingErr.Error(), nil)
+		default:
+			writeErrorResponse(w, http.StatusBadRequest, "invalid_compression", "Failed to decompress request body", nil)
+		}
+		return false
+	}
+	defer decompressed.Close()
+
+	if err := json.NewDecoder(decompressed).Decode(destination); err != nil {
+		writeAgentReportDecodeError(w, err)
+		return false
+	}
+	// json.Decoder may finish a complete top-level object before it asks the
+	// reader for the proof byte above a body limit. Drain the bounded reader so
+	// compressed bombs and oversized bodies cannot bypass the cap that way.
+	if _, err := io.Copy(io.Discard, decompressed); err != nil {
+		writeAgentReportDecodeError(w, err)
+		return false
+	}
+
+	return true
+}
+
+func writeAgentReportDecodeError(w http.ResponseWriter, err error) {
+	var maxBytesErr *http.MaxBytesError
+	var decodedSizeErr *utils.DecompressedBodyTooLargeError
+	switch {
+	case errors.As(err, &maxBytesErr):
+		writeAgentReportSizeError(w, agentReportSizeEncodedBody, maxBytesErr.Limit)
+	case errors.As(err, &decodedSizeErr):
+		writeAgentReportSizeError(w, agentReportSizeDecodedJSON, decodedSizeErr.Limit)
+	default:
+		writeErrorResponse(w, http.StatusBadRequest, "invalid_json", "Failed to decode request body", map[string]string{"error": err.Error()})
+	}
+}
+
+func writeAgentReportSizeError(w http.ResponseWriter, dimension string, limit int64) {
+	writeErrorResponse(
+		w,
+		http.StatusRequestEntityTooLarge,
+		"report_too_large",
+		"Agent report exceeds the server size limit",
+		map[string]string{
+			"dimension":  dimension,
+			"limitBytes": strconv.FormatInt(limit, 10),
+		},
+	)
+}

internal/api/agent_report_decode_test.go

@@ -0,0 +1,123 @@
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+)
+
+type agentReportErrorResponse struct {
+	Code    string            `json:"code"`
+	Details map[string]string `json:"details"`
+}
+
+func TestAgentReportHandlersClassifyBodyFailures(t *testing.T) {
+	unified, _ := newUnifiedAgentHandlers(t, nil)
+	kubernetes, _ := newKubernetesAgentHandlers(t, nil)
+
+	tests := []struct {
+		name         string
+		handle       http.HandlerFunc
+		path         string
+		encodedLimit int64
+		decodedLimit int64
+	}{
+		{
+			name: "unified agent", handle: unified.HandleReport,
+			path: "/api/agents/agent/report", encodedLimit: 256 * 1024, decodedLimit: 1536 * 1024,
+		},
+		{
+			name: "kubernetes agent", handle: kubernetes.HandleReport,
+			path: "/api/agents/kubernetes/report", encodedLimit: 2 * 1024 * 1024, decodedLimit: 10 * 1024 * 1024,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name+" encoded limit", func(t *testing.T) {
+			body := oversizedJSON(tc.encodedLimit)
+			rec := invokeAgentReportHandler(tc.handle, tc.path, body, "")
+			assertAgentReportError(t, rec, http.StatusRequestEntityTooLarge, "report_too_large", agentReportSizeEncodedBody, tc.encodedLimit)
+		})
+
+		t.Run(tc.name+" decoded limit", func(t *testing.T) {
+			body, err := utils.CompressJSON(oversizedJSON(tc.decodedLimit))
+			if err != nil {
+				t.Fatalf("compress oversized report: %v", err)
+			}
+			if int64(len(body)) >= tc.encodedLimit {
+				t.Fatalf("fixture compressed to %d bytes, must stay below encoded limit %d", len(body), tc.encodedLimit)
+			}
+			rec := invokeAgentReportHandler(tc.handle, tc.path, body, "gzip")
+			assertAgentReportError(t, rec, http.StatusRequestEntityTooLarge, "report_too_large", agentReportSizeDecodedJSON, tc.decodedLimit)
+		})
+
+		t.Run(tc.name+" malformed gzip", func(t *testing.T) {
+			rec := invokeAgentReportHandler(tc.handle, tc.path, []byte("not a gzip stream"), "gzip")
+			assertAgentReportError(t, rec, http.StatusBadRequest, "invalid_compression", "", 0)
+		})
+
+		t.Run(tc.name+" unsupported encoding", func(t *testing.T) {
+			rec := invokeAgentReportHandler(tc.handle, tc.path, []byte(`{}`), "br")
+			assertAgentReportError(t, rec, http.StatusUnsupportedMediaType, "unsupported_encoding", "", 0)
+		})
+	}
+}
+
+func oversizedJSON(limit int64) []byte {
+	prefix := []byte(`{"padding":"`)
+	suffix := []byte(`"}`)
+	padding := limit + 1 - int64(len(prefix)) - int64(len(suffix))
+	if padding < 1 {
+		padding = 1
+	}
+	body := make([]byte, 0, int64(len(prefix))+padding+int64(len(suffix)))
+	body = append(body, prefix...)
+	body = append(body, bytes.Repeat([]byte("x"), int(padding))...)
+	body = append(body, suffix...)
+	return body
+}
+
+func invokeAgentReportHandler(handle http.HandlerFunc, path string, body []byte, encoding string) *httptest.ResponseRecorder {
+	req := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(body))
+	if encoding != "" {
+		req.Header.Set("Content-Encoding", encoding)
+	}
+	rec := httptest.NewRecorder()
+	handle(rec, req)
+	return rec
+}
+
+func assertAgentReportError(
+	t *testing.T,
+	rec *httptest.ResponseRecorder,
+	wantStatus int,
+	wantCode string,
+	wantDimension string,
+	wantLimit int64,
+) {
+	t.Helper()
+	if rec.Code != wantStatus {
+		t.Fatalf("status = %d, want %d: %s", rec.Code, wantStatus, rec.Body.String())
+	}
+	var response agentReportErrorResponse
+	if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
+		t.Fatalf("decode error response: %v", err)
+	}
+	if response.Code != wantCode {
+		t.Fatalf("code = %q, want %q", response.Code, wantCode)
+	}
+	if wantDimension == "" {
+		return
+	}
+	if response.Details["dimension"] != wantDimension {
+		t.Fatalf("dimension = %q, want %q", response.Details["dimension"], wantDimension)
+	}
+	if response.Details["limitBytes"] != strconv.FormatInt(wantLimit, 10) {
+		t.Fatalf("limitBytes = %q, want %d", response.Details["limitBytes"], wantLimit)
+	}
+}

internal/api/kubernetes_agents.go

@@ -40,21 +40,10 @@ func (h *KubernetesAgentHandlers) HandleReport(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	// Limit request body to 2MB to prevent memory exhaustion (pods can be sizable).
-	r.Body = http.MaxBytesReader(w, r.Body, 2*1024*1024)
-	defer r.Body.Close()
-
-	// Support gzip-compressed reports from agents (backward compatible with uncompressed)
-	body, err := utils.DecompressBodyIfGzipped(r, 10*1024*1024)
-	if err != nil {
-		writeErrorResponse(w, http.StatusUnsupportedMediaType, "unsupported_encoding", err.Error(), nil)
-		return
-	}
-	defer body.Close()
-
 	var report agentsk8s.Report
-	if err := json.NewDecoder(body).Decode(&report); err != nil {
-		writeErrorResponse(w, http.StatusBadRequest, "invalid_json", "Failed to decode request body", map[string]string{"error": err.Error()})
+	// Pod inventories can be sizable and compress well, so cap both the
+	// encoded HTTP body and the decoded JSON body independently.
+	if !decodeCompressedAgentReport(w, r, 2*1024*1024, 10*1024*1024, &report) {
 		return
 	}
 

internal/utils/gzip.go

@@ -9,6 +9,28 @@ import (
 	"strings"
 )
 
+// UnsupportedContentEncodingError reports a Content-Encoding value that the
+// server does not know how to decode. Callers can distinguish this from a
+// malformed payload that claims to use a supported encoding.
+type UnsupportedContentEncodingError struct {
+	Encoding string
+}
+
+func (e *UnsupportedContentEncodingError) Error() string {
+	return fmt.Sprintf("unsupported Content-Encoding: %s", e.Encoding)
+}
+
+// DecompressedBodyTooLargeError reports that a compressed body expanded past
+// the decoded payload limit. It is intentionally typed so HTTP handlers do not
+// misclassify the read failure as malformed JSON.
+type DecompressedBodyTooLargeError struct {
+	Limit int64
+}
+
+func (e *DecompressedBodyTooLargeError) Error() string {
+	return fmt.Sprintf("decompressed payload exceeds %d byte limit", e.Limit)
+}
+
 // CompressJSON compresses a JSON payload using gzip BestSpeed.
 // Returns the compressed bytes suitable for use as an HTTP request body.
 func CompressJSON(payload []byte) ([]byte, error) {
@@ -44,7 +66,7 @@ func DecompressBodyIfGzipped(r *http.Request, maxDecompressed int64) (io.ReadClo
 		limited := io.LimitReader(gz, maxDecompressed+1)
 		return &cappedGzipReader{gz: gz, lr: limited, max: maxDecompressed}, nil
 	default:
-		return nil, fmt.Errorf("unsupported Content-Encoding: %s", encoding)
+		return nil, &UnsupportedContentEncodingError{Encoding: encoding}
 	}
 }
 
@@ -54,15 +76,30 @@ type cappedGzipReader struct {
 	lr  io.Reader
 	max int64
 	n   int64
+	err error
 }
 
 func (c *cappedGzipReader) Read(p []byte) (int, error) {
-	n, err := c.lr.Read(p)
-	c.n += int64(n)
-	if c.n > c.max {
-		return n, fmt.Errorf("decompressed payload exceeds %d byte limit", c.max)
+	if c.err != nil {
+		return 0, c.err
 	}
-	return n, err
+	n, err := c.lr.Read(p)
+	if c.n+int64(n) <= c.max {
+		c.n += int64(n)
+		return n, err
+	}
+
+	// LimitReader intentionally permits one byte beyond max so an exact-limit
+	// payload remains distinguishable from an oversized one. Do not expose that
+	// proof byte to callers: decoders are allowed to accept a complete value
+	// even when Read returns data and an error together.
+	allowed := c.max - c.n
+	if allowed < 0 {
+		allowed = 0
+	}
+	c.n = c.max
+	c.err = &DecompressedBodyTooLargeError{Limit: c.max}
+	return int(allowed), c.err
 }
 
 func (c *cappedGzipReader) Close() error {

internal/utils/gzip_test.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"bytes"
 	"compress/gzip"
+	"errors"
 	"io"
 	"net/http"
 	"strings"
@@ -114,6 +115,10 @@ func TestDecompressBodyIfGzipped_UnsupportedEncoding(t *testing.T) {
 	if !strings.Contains(err.Error(), "unsupported Content-Encoding") {
 		t.Fatalf("unexpected error: %v", err)
 	}
+	var encodingErr *UnsupportedContentEncodingError
+	if !errors.As(err, &encodingErr) || encodingErr.Encoding != "deflate" {
+		t.Fatalf("expected typed unsupported encoding error, got %T: %v", err, err)
+	}
 }
 
 func TestDecompressBodyIfGzipped_BombProtection(t *testing.T) {
@@ -135,11 +140,18 @@ func TestDecompressBodyIfGzipped_BombProtection(t *testing.T) {
 	}
 	defer body.Close()
 
-	_, readErr := io.ReadAll(body)
+	data, readErr := io.ReadAll(body)
 	if readErr == nil {
 		t.Fatal("expected error when decompressed size exceeds limit")
 	}
 	if !strings.Contains(readErr.Error(), "exceeds") {
 		t.Fatalf("unexpected error: %v", readErr)
 	}
+	var sizeErr *DecompressedBodyTooLargeError
+	if !errors.As(readErr, &sizeErr) || sizeErr.Limit != 1024 {
+		t.Fatalf("expected typed decoded-size error, got %T: %v", readErr, readErr)
+	}
+	if len(data) > 1024 {
+		t.Fatalf("reader exposed %d decoded bytes past the 1024-byte limit", len(data))
+	}
 }