From 3da4ce1f314a1c186cc53fbf3ca74261a0837384 Mon Sep 17 00:00:00 2001 From: Shipra Gupta Date: Sat, 20 Dec 2025 05:45:14 +0530 Subject: [PATCH] nsqd: add opt-in --http-reject-chunked flag to reject chunked requests --- apps/nsqd/options.go | 3 +++ nsqd/http.go | 32 +++++++++++++++++++++++++---- nsqd/http_test.go | 49 ++++++++++++++++++++++++++++++++++++++++++++ nsqd/options.go | 6 +++++- 4 files changed, 85 insertions(+), 5 deletions(-) diff --git a/apps/nsqd/options.go b/apps/nsqd/options.go index b1380c777..a05f7fced 100644 --- a/apps/nsqd/options.go +++ b/apps/nsqd/options.go @@ -145,6 +145,9 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet { flagSet.String("topology-region", opts.TopologyRegion, "A region represents a larger domain, made up of one or more zones for preferring closer consumer") flagSet.String("topology-zone", opts.TopologyZone, "A zone represents a logical failure domain for preferring closer consumer") + // HTTP behaviour + flagSet.Bool("http-reject-chunked", opts.HTTPRejectChunked, "reject requests with Transfer-Encoding: chunked when enabled") + // diskqueue options flagSet.String("data-path", opts.DataPath, "path to store disk-backed messages") flagSet.Int64("mem-queue-size", opts.MemQueueSize, "number of messages to keep in memory (per topic/channel)") diff --git a/nsqd/http.go b/nsqd/http.go index fdaf1d481..b67b8faf1 100644 --- a/nsqd/http.go +++ b/nsqd/http.go @@ -211,8 +211,20 @@ func (s *httpServer) getTopicFromQuery(req *http.Request) (url.Values, *Topic, e } func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) { - // TODO: one day I'd really like to just error on chunked requests - // to be able to fail "too big" requests before we even read + // Optionally reject chunked requests + if s.nsqd.getOpts().HTTPRejectChunked { + for _, te := range req.TransferEncoding { + if strings.ToLower(te) == "chunked" { + return nil, http_api.Err{400, "CHUNKED_TRANSFER_ENCODING_NOT_SUPPORTED"} + } + } + // also inspect header for older clients + if te := req.Header.Get("Transfer-Encoding"); te != "" { + if strings.Contains(strings.ToLower(te), "chunked") { + return nil, http_api.Err{400, "CHUNKED_TRANSFER_ENCODING_NOT_SUPPORTED"} + } + } + } if req.ContentLength > s.nsqd.getOpts().MaxMsgSize { return nil, http_api.Err{413, "MSG_TOO_BIG"} @@ -264,8 +276,20 @@ func (s *httpServer) doMPUB(w http.ResponseWriter, req *http.Request, ps httprou var msgs []*Message var exit bool - // TODO: one day I'd really like to just error on chunked requests - // to be able to fail "too big" requests before we even read + // Optionally reject chunked requests + if s.nsqd.getOpts().HTTPRejectChunked { + for _, te := range req.TransferEncoding { + if strings.ToLower(te) == "chunked" { + return nil, http_api.Err{400, "CHUNKED_TRANSFER_ENCODING_NOT_SUPPORTED"} + } + } + // also inspect header for older clients + if te := req.Header.Get("Transfer-Encoding"); te != "" { + if strings.Contains(strings.ToLower(te), "chunked") { + return nil, http_api.Err{400, "CHUNKED_TRANSFER_ENCODING_NOT_SUPPORTED"} + } + } + } if req.ContentLength > s.nsqd.getOpts().MaxBodySize { return nil, http_api.Err{413, "BODY_TOO_BIG"} diff --git a/nsqd/http_test.go b/nsqd/http_test.go index 5fdb9b3d3..a57dd3a45 100644 --- a/nsqd/http_test.go +++ b/nsqd/http_test.go @@ -1,6 +1,7 @@ package nsqd import ( + "bufio" "bytes" "crypto/tls" "encoding/json" @@ -59,6 +60,54 @@ func TestHTTPpub(t *testing.T) { test.Equal(t, int64(1), topic.Depth()) } +func TestHTTPpubRejectChunkedWhenEnabled(t *testing.T) { + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.HTTPRejectChunked = true + _, httpAddr, nsqd := mustStartNSQD(opts) + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + topicName := "test_http_pub_chunked" + strconv.Itoa(int(time.Now().Unix())) + + conn, err := net.Dial("tcp", httpAddr.String()) + test.Nil(t, err) + defer conn.Close() + + req := fmt.Sprintf("POST /pub?topic=%s HTTP/1.1\r\nHost: %s\r\nTransfer-Encoding: chunked\r\nContent-Type: application/octet-stream\r\n\r\n5\r\nhello\r\n0\r\n\r\n", topicName, httpAddr.String()) + _, err = conn.Write([]byte(req)) + test.Nil(t, err) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + test.Nil(t, err) + defer resp.Body.Close() + test.Equal(t, 400, resp.StatusCode) +} + +func TestHTTPmpubRejectChunkedWhenEnabled(t *testing.T) { + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.HTTPRejectChunked = true + _, httpAddr, nsqd := mustStartNSQD(opts) + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + topicName := "test_http_mpub_chunked" + strconv.Itoa(int(time.Now().Unix())) + + conn, err := net.Dial("tcp", httpAddr.String()) + test.Nil(t, err) + defer conn.Close() + + req := fmt.Sprintf("POST /mpub?topic=%s HTTP/1.1\r\nHost: %s\r\nTransfer-Encoding: chunked\r\nContent-Type: application/octet-stream\r\n\r\n5\r\nhello\r\n0\r\n\r\n", topicName, httpAddr.String()) + _, err = conn.Write([]byte(req)) + test.Nil(t, err) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + test.Nil(t, err) + defer resp.Body.Close() + test.Equal(t, 400, resp.StatusCode) +} + func TestHTTPpubEmpty(t *testing.T) { opts := NewOptions() opts.Logger = test.NewTestLogger(t) diff --git a/nsqd/options.go b/nsqd/options.go index f81ae23a8..abdaa3081 100644 --- a/nsqd/options.go +++ b/nsqd/options.go @@ -91,6 +91,9 @@ type Options struct { // experimental features Experiments []string `flag:"enable-experiment" cfg:"enable_experiment"` + + // HTTP behavior + HTTPRejectChunked bool `flag:"http-reject-chunked" cfg:"http_reject_chunked"` } type Experiment string @@ -179,6 +182,7 @@ func NewOptions() *Options { MaxDeflateLevel: 6, SnappyEnabled: true, - TLSMinVersion: tls.VersionTLS10, + TLSMinVersion: tls.VersionTLS10, + HTTPRejectChunked: false, } }