Pass a message size limit to the stream command decoder

Bump github.com/centrifugal/protocol to v0.21.0, which now requires a
positive message size limit when constructing a stream command decoder
(GetStreamCommandDecoder, the unbounded shortcut, was removed). An
unbounded decoder over untrusted input can be driven to allocate
arbitrary memory by a single frame declaring a huge length. See
GHSA-4r3x-2rwr-6w65.

HandleReadFrame now takes a messageSizeLimit and passes it to
GetStreamCommandDecoderLimited, so an oversized declared length is
rejected before it is allocated. Each transport supplies its own limit,
coerced to a positive default:

- WebSocket: MessageSizeLimit, or DecompressedMessageSizeLimit when
  compression is enabled (the decoder sees decoded bytes).
- SSE / HTTP-stream: MaxRequestBodySize.
- Emulation: the length of the already-buffered command bytes, which a
  single command cannot exceed.

All coercions use <= 0 so a negative config cannot reach the decoder.

Author: FZambia

client_test.go

@@ -263,7 +263,7 @@ func TestClientV2PingPong(t *testing.T) {
 		for msg := range messages {
 			if string(msg) == "{}" {
 				// PING
-				HandleReadFrame(client, bytes.NewReader([]byte("{}")))
+				HandleReadFrame(client, bytes.NewReader([]byte("{}")), 1<<20)
 			}
 		}
 	}()
@@ -2834,14 +2834,14 @@ func TestClientHandleEmptyData(t *testing.T) {
 	defer func() { _ = node.Shutdown(context.Background()) }()
 
 	client := newTestClient(t, node, "42")
-	proceed := HandleReadFrame(client, bytes.NewReader([]byte(nil)))
+	proceed := HandleReadFrame(client, bytes.NewReader([]byte(nil)), 1<<20)
 	require.False(t, proceed)
 	select {
 	case <-client.Context().Done():
 	case <-time.After(time.Second):
 		require.Fail(t, "client not closed")
 	}
-	proceed = HandleReadFrame(client, bytes.NewReader([]byte("test")))
+	proceed = HandleReadFrame(client, bytes.NewReader([]byte("test")), 1<<20)
 	require.False(t, proceed)
 	disconnect, proceed := client.dispatchCommand(&protocol.Command{}, 0)
 	require.Nil(t, disconnect)
@@ -2854,7 +2854,7 @@ func TestClientHandleBrokenData(t *testing.T) {
 	defer func() { _ = node.Shutdown(context.Background()) }()
 
 	client := newTestClient(t, node, "42")
-	proceed := HandleReadFrame(client, bytes.NewReader([]byte(`nd3487yt734y38&**&**`)))
+	proceed := HandleReadFrame(client, bytes.NewReader([]byte(`nd3487yt734y38&**&**`)), 1<<20)
 	require.False(t, proceed)
 	select {
 	case <-client.Context().Done():
@@ -2874,7 +2874,7 @@ func TestClientHandleCommandNotAuthenticated(t *testing.T) {
 	}}
 	data, err := json.Marshal(cmd)
 	require.NoError(t, err)
-	proceed := HandleReadFrame(client, bytes.NewReader(data))
+	proceed := HandleReadFrame(client, bytes.NewReader(data), 1<<20)
 	require.False(t, proceed)
 	select {
 	case <-client.Context().Done():
@@ -2932,7 +2932,7 @@ func TestClientHandleCommandWithoutID(t *testing.T) {
 	cmd := &protocol.Command{}
 	data, err := json.Marshal(cmd)
 	require.NoError(t, err)
-	proceed := HandleReadFrame(client, bytes.NewReader(data))
+	proceed := HandleReadFrame(client, bytes.NewReader(data), 1<<20)
 	require.False(t, proceed)
 	select {
 	case <-client.Context().Done():
@@ -2968,7 +2968,7 @@ func TestClientAlreadyAuthenticated(t *testing.T) {
 	cmd := &protocol.Command{Id: 2, Connect: &protocol.ConnectRequest{}}
 	data, err := json.Marshal(cmd)
 	require.NoError(t, err)
-	proceed := HandleReadFrame(client, bytes.NewReader(data))
+	proceed := HandleReadFrame(client, bytes.NewReader(data), 1<<20)
 	require.False(t, proceed)
 	select {
 	case <-client.Context().Done():
@@ -4899,7 +4899,7 @@ func BenchmarkClientRPC(b *testing.B) {
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		decoder := protocol.GetStreamCommandDecoder(protocol.TypeProtobuf, bytes.NewReader(frame))
+		decoder := protocol.GetStreamCommandDecoderLimited(protocol.TypeProtobuf, bytes.NewReader(frame), 1<<20)
 		cmd, cmdProtocolSize, err := decoder.Decode()
 		require.NoError(b, err)
 		client.HandleCommand(cmd, cmdProtocolSize)

emulation.go

@@ -165,7 +165,11 @@ func (h *emulationSurveyHandler) HandleEmulation(e SurveyEvent, cb SurveyCallbac
 	}
 	go func() {
 		reader := readerpool.GetBytesReader(data)
-		_ = HandleReadFrame(client, reader)
+		// data holds the already-buffered command bytes, so a single command
+		// cannot exceed its length; +1 keeps the limit positive (the stream
+		// decoder rejects a non-positive limit) and admits a command of exactly
+		// len(data) bytes. The ingress body size was already bounded upstream.
+		_ = HandleReadFrame(client, reader, int64(len(data))+1)
 		readerpool.PutBytesReader(reader)
 		cb(SurveyReply{})
 	}()

go.mod

@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/FZambia/eagle v0.2.0
-	github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09
+	github.com/centrifugal/protocol v0.21.0
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/google/cel-go v0.30.0
 	github.com/google/uuid v1.6.0

go.sum

@@ -6,8 +6,8 @@ github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYW
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
-github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09 h1:8SJBvmcS/KXNu8h/79VyYgey4WNreVM9NR3NrGNLFSQ=
-github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09/go.mod h1:3pinAfcb+bH8Yp8Ewhf9QTRTzsQ9veP/QOrCk5D5SSE=
+github.com/centrifugal/protocol v0.21.0 h1:yagnxWBH7vK2+0+A0bmPAIOCaFqTvaLVWubkklS7Xyo=
+github.com/centrifugal/protocol v0.21.0/go.mod h1:3pinAfcb+bH8Yp8Ewhf9QTRTzsQ9veP/QOrCk5D5SSE=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=

handler_http_stream.go

@@ -62,18 +62,21 @@ func (h *HTTPStreamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		protocolType = ProtocolTypeProtobuf
 	}
 
+	// messageSizeLimit bounds both the request body and a single decoded command;
+	// it must be positive since the stream decoder rejects a non-positive limit.
+	messageSizeLimit := h.config.MaxRequestBodySize
+	if messageSizeLimit <= 0 {
+		messageSizeLimit = defaultMaxHTTPStreamingBodySize
+	}
+
 	var requestData []byte
 	if r.Method == http.MethodPost {
-		maxBytesSize := h.config.MaxRequestBodySize
-		if maxBytesSize == 0 {
-			maxBytesSize = defaultMaxHTTPStreamingBodySize
-		}
-		r.Body = http.MaxBytesReader(w, r.Body, int64(maxBytesSize))
+		r.Body = http.MaxBytesReader(w, r.Body, int64(messageSizeLimit))
 		var err error
 		requestData, err = io.ReadAll(r.Body)
 		if err != nil {
 			h.node.logger.log(newLogEntry(LogLevelInfo, "error reading http stream request body", map[string]any{"error": err.Error()}))
-			if len(requestData) >= maxBytesSize {
+			if len(requestData) >= messageSizeLimit {
 				w.WriteHeader(http.StatusRequestEntityTooLarge)
 				return
 			}
@@ -120,7 +123,7 @@ func (h *HTTPStreamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	rc := http.NewResponseController(w)
 
 	reader := readerpool.GetBytesReader(requestData)
-	_ = HandleReadFrame(c, reader)
+	_ = HandleReadFrame(c, reader, int64(messageSizeLimit))
 	readerpool.PutBytesReader(reader)
 
 	sendAck := func() {

handler_sse.go

@@ -50,6 +50,13 @@ func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// messageSizeLimit bounds both the request body and a single decoded command;
+	// it must be positive since the stream decoder rejects a non-positive limit.
+	messageSizeLimit := h.config.MaxRequestBodySize
+	if messageSizeLimit <= 0 {
+		messageSizeLimit = defaultMaxSSEBodySize
+	}
+
 	var requestData []byte
 	if r.Method == http.MethodGet {
 		requestDataString := r.URL.Query().Get(connectUrlParam)
@@ -61,16 +68,12 @@ func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	} else if r.Method == http.MethodPost {
-		maxBytesSize := h.config.MaxRequestBodySize
-		if maxBytesSize == 0 {
-			maxBytesSize = defaultMaxSSEBodySize
-		}
-		r.Body = http.MaxBytesReader(w, r.Body, int64(maxBytesSize))
+		r.Body = http.MaxBytesReader(w, r.Body, int64(messageSizeLimit))
 		var err error
 		requestData, err = io.ReadAll(r.Body)
 		if err != nil {
 			h.node.logger.log(newLogEntry(LogLevelInfo, "error reading sse request body", map[string]any{"error": err.Error()}))
-			if len(requestData) >= maxBytesSize {
+			if len(requestData) >= messageSizeLimit {
 				w.WriteHeader(http.StatusRequestEntityTooLarge)
 				return
 			}
@@ -128,7 +131,7 @@ func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	_ = rc.SetWriteDeadline(time.Time{})
 
 	reader := readerpool.GetBytesReader(requestData)
-	_ = HandleReadFrame(c, reader)
+	_ = HandleReadFrame(c, reader, int64(messageSizeLimit))
 	readerpool.PutBytesReader(reader)
 
 	sendAck := func() {

handler_websocket.go

@@ -223,20 +223,22 @@ func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 		writeTimeout = 1 * time.Second
 	}
 	messageSizeLimit := s.config.MessageSizeLimit
-	if messageSizeLimit == 0 {
+	if messageSizeLimit <= 0 {
 		messageSizeLimit = 65536 // 64KB
 	}
-	if messageSizeLimit > 0 {
-		conn.SetReadLimit(int64(messageSizeLimit))
-	}
+	conn.SetReadLimit(int64(messageSizeLimit))
+	// The stream command decoder parses decoded command bytes, so it is bounded
+	// by the decompressed message size when compression is enabled, otherwise by
+	// the wire message size. It must be positive - the decoder rejects a
+	// non-positive limit.
+	decoderMessageSizeLimit := int64(messageSizeLimit)
 	if compression {
 		decompressedMessageSizeLimit := s.config.DecompressedMessageSizeLimit
-		if decompressedMessageSizeLimit == 0 {
+		if decompressedMessageSizeLimit <= 0 {
 			decompressedMessageSizeLimit = messageSizeLimit * defaultWebsocketDecompressedMessageSizeLimitMultiplier
 		}
-		if decompressedMessageSizeLimit > 0 {
-			conn.SetDecompressedReadLimit(int64(decompressedMessageSizeLimit))
-		}
+		conn.SetDecompressedReadLimit(int64(decompressedMessageSizeLimit))
+		decoderMessageSizeLimit = int64(decompressedMessageSizeLimit)
 	}
 
 	if useFramePingPong {
@@ -309,7 +311,7 @@ func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 		// costs a goroutine but no allocation. See ProcessCommandsOffReadLoop.
 		var handoff *frameHandoff
 		if s.config.ProcessCommandsOffReadLoop {
-			handoff = newFrameHandoff(c)
+			handoff = newFrameHandoff(c, decoderMessageSizeLimit)
 		}
 
 		for {
@@ -324,7 +326,7 @@ func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 			if handoff != nil {
 				proceed = handoff.handle(r)
 			} else {
-				proceed = HandleReadFrame(c, r)
+				proceed = HandleReadFrame(c, r, decoderMessageSizeLimit)
 			}
 			if !proceed {
 				break
@@ -365,17 +367,18 @@ func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 // Only the read loop writes r, and it does so before starting run and does not
 // touch it again until run has sent on done, so the two goroutines never race.
 type frameHandoff struct {
-	c    *Client
-	r    io.Reader
-	done chan bool
+	c                *Client
+	r                io.Reader
+	done             chan bool
+	messageSizeLimit int64
 }
 
-func newFrameHandoff(c *Client) *frameHandoff {
-	return &frameHandoff{c: c, done: make(chan bool, 1)}
+func newFrameHandoff(c *Client, messageSizeLimit int64) *frameHandoff {
+	return &frameHandoff{c: c, done: make(chan bool, 1), messageSizeLimit: messageSizeLimit}
 }
 
 func (h *frameHandoff) run() {
-	h.done <- HandleReadFrame(h.c, h.r)
+	h.done <- HandleReadFrame(h.c, h.r, h.messageSizeLimit)
 }
 
 // handle processes one frame off the read loop and reports whether the
@@ -393,9 +396,14 @@ func (h *frameHandoff) handle(r io.Reader) bool {
 // HandleReadFrame is a helper to read Centrifuge commands from frame-based io.Reader and
 // process them. Frame-based means that EOF treated as the end of the frame, not the entire
 // connection close.
-func HandleReadFrame(c *Client, r io.Reader) bool {
+//
+// messageSizeLimit bounds the size of a single command and must be positive: the
+// underlying stream decoder rejects a non-positive limit, since the length prefix
+// is attacker-controlled and used as an allocation size. Callers pass the
+// transport's configured limit (coerced to a positive default).
+func HandleReadFrame(c *Client, r io.Reader, messageSizeLimit int64) bool {
 	protoType := c.Transport().Protocol().toProto()
-	decoder := protocol.GetStreamCommandDecoder(protoType, r)
+	decoder := protocol.GetStreamCommandDecoderLimited(protoType, r, messageSizeLimit)
 	defer protocol.PutStreamCommandDecoder(protoType, decoder)
 
 	hadCommands := false