nsqd: add opt-in --http-reject-chunked flag to reject chunked requests

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)")

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"}

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)

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,
 	}
 }