Clamp maximum preallocation for enveloped messages

For enveloped protocols (gRPC, gRPC-Web, and Connect streaming), this PR
clamps the maximum number of bytes that the Connect runtime will
pre-allocate based on the size portion of the envelope prefix. This
prevents malicious clients from sending a prefix with a very large
message size and triggering large allocations in the handler. Note that
this is a fallback mechanism explicitly intended to protect servers from
fuzzers or malicious payloads - it doesn't protect servers from clients
actually sending very large messages.

For now, there's no configuration knob to adjust this parameter. Servers
can already use `WithReadMaxBytes`, which protects them from both
spurious envelopes and genuinely large messages. If this becomes
problematic later on, we can always adjust it or expose another option.

envelope.go

@@ -25,6 +25,11 @@ import (
 // same meaning in the gRPC-Web, gRPC-HTTP2, and Connect protocols.
 const flagEnvelopeCompressed = 0b00000001
 
+// maxPreallocateBytes is the largest allocation we're willing to make eagerly.
+// Capping this prevents malicious clients from sending an envelope that
+// immediately triggers a large allocation.
+const maxPreallocateBytes = 1024 * 1024 * 4 // 4MiB
+
 var errSpecialEnvelope = errorf(
 	CodeUnknown,
 	"final message has protocol-specific flags: %w",
@@ -268,8 +273,15 @@ func (r *envelopeReader) Read(env *envelope) *Error {
 		}
 		return errorf(CodeResourceExhausted, "message size %d is larger than configured max %d", size, r.readMaxBytes)
 	}
+	preallocateBytes := size
+	// Where possible, avoid many small allocations growing the buffer, but don't
+	// blindly trust the client: malicious clients may try to trigger large
+	// allocations by sending a spurious prefix.
+	if preallocateBytes > maxPreallocateBytes {
+		preallocateBytes = maxPreallocateBytes
+	}
 	if size > 0 {
-		env.Data.Grow(size)
+		env.Data.Grow(preallocateBytes)
 		// At layer 7, we don't know exactly what's happening down in L4. Large
 		// length-prefixed messages may arrive in chunks, so we may need to read
 		// the request body past EOF. We also need to take care that we don't retry

handler_ext_test.go

@@ -15,11 +15,15 @@
 package connect_test
 
 import (
+	"bytes"
 	"context"
+	"encoding/binary"
 	"encoding/json"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 
 	connect "connectrpc.com/connect"
@@ -209,6 +213,47 @@ func TestHandler_ServeHTTP(t *testing.T) {
 	})
 }
 
+func TestHandlerMaliciousPrefix(t *testing.T) {
+	t.Parallel()
+	mux := http.NewServeMux()
+	mux.Handle(pingv1connect.NewPingServiceHandler(successPingServer{}))
+	server := httptest.NewServer(mux)
+	t.Cleanup(server.Close)
+
+	const (
+		concurrency  = 256
+		spuriousSize = 1024 * 1024 * 512 // 512 MB
+	)
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for i := 0; i < concurrency; i++ {
+		body := make([]byte, 16)
+		// Envelope prefix indicates a large payload which we're not actually
+		// sending.
+		binary.BigEndian.PutUint32(body[1:5], spuriousSize)
+		req, err := http.NewRequestWithContext(
+			context.Background(),
+			http.MethodPost,
+			server.URL+pingv1connect.PingServicePingProcedure,
+			bytes.NewReader(body),
+		)
+		assert.Nil(t, err)
+		req.Header.Set("Content-Type", "application/grpc")
+		wg.Add(1)
+		go func(req *http.Request) {
+			defer wg.Done()
+			<-start
+			response, err := server.Client().Do(req)
+			if err == nil {
+				io.Copy(io.Discard, response.Body)
+				response.Body.Close()
+			}
+		}(req)
+	}
+	close(start)
+	wg.Wait()
+}
+
 type successPingServer struct {
 	pingv1connect.UnimplementedPingServiceHandler
 }