diff --git a/internal/api/agent_ingest.go b/internal/api/agent_ingest.go index 7defdaf9d..29d1f8d5a 100644 --- a/internal/api/agent_ingest.go +++ b/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 } diff --git a/internal/api/agent_report_decode.go b/internal/api/agent_report_decode.go new file mode 100644 index 000000000..2f6fb87fe --- /dev/null +++ b/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), + }, + ) +} diff --git a/internal/api/agent_report_decode_test.go b/internal/api/agent_report_decode_test.go new file mode 100644 index 000000000..d22bb87be --- /dev/null +++ b/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) + } +} diff --git a/internal/api/kubernetes_agents.go b/internal/api/kubernetes_agents.go index ed54406f6..f439fe7e6 100644 --- a/internal/api/kubernetes_agents.go +++ b/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 } diff --git a/internal/utils/gzip.go b/internal/utils/gzip.go index eaea9d0a4..735513542 100644 --- a/internal/utils/gzip.go +++ b/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 { diff --git a/internal/utils/gzip_test.go b/internal/utils/gzip_test.go index 2bb4afb05..2c4f9c6f4 100644 --- a/internal/utils/gzip_test.go +++ b/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)) + } }