Strictly Limit the size of decompressed values to 1 MiB

The default message size limit in GossipSub is 1 MiB, which is unchanged
in Lotus. This means when decompressing values, we can never have a
valid compressed message that expands to larger than 1 MiB.

Set this limit explicitly in the zstd decoder.

internal/encoding/encoding.go

@@ -2,11 +2,17 @@ package encoding
 
 import (
 	"bytes"
+	"fmt"
 
 	"github.com/klauspost/compress/zstd"
 	cbg "github.com/whyrusleeping/cbor-gen"
 )
 
+// maxDecompressedSize is the default maximum amount of memory allocated by the
+// zstd decoder. The limit of 1MiB is chosen based on the default maximum message
+// size in GossipSub.
+const maxDecompressedSize = 1 << 20
+
 type CBORMarshalUnmarshaler interface {
 	cbg.CBORMarshaler
 	cbg.CBORUnmarshaler
@@ -47,7 +53,7 @@ func NewZSTD[T CBORMarshalUnmarshaler]() (*ZSTD[T], error) {
 	if err != nil {
 		return nil, err
 	}
-	reader, err := zstd.NewReader(nil)
+	reader, err := zstd.NewReader(nil, zstd.WithDecoderMaxMemory(maxDecompressedSize))
 	if err != nil {
 		return nil, err
 	}
@@ -60,6 +66,10 @@ func NewZSTD[T CBORMarshalUnmarshaler]() (*ZSTD[T], error) {
 
 func (c *ZSTD[T]) Encode(m T) ([]byte, error) {
 	cborEncoded, err := c.cborEncoding.Encode(m)
+	if len(cborEncoded) > maxDecompressedSize {
+		// Error out early if the encoded value is too large to be decompressed.
+		return nil, fmt.Errorf("encoded value cannot exceed maximum size: %d > %d", len(cborEncoded), maxDecompressedSize)
+	}
 	if err != nil {
 		return nil, err
 	}

internal/encoding/encoding_test.go

@@ -1,10 +1,12 @@
 package encoding_test
 
 import (
+	"bytes"
 	"io"
 	"testing"
 
 	"github.com/filecoin-project/go-f3/internal/encoding"
+	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/require"
 	cbg "github.com/whyrusleeping/cbor-gen"
 )
@@ -53,3 +55,25 @@ func TestZSTD(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, data.Value, decoded.Value)
 }
+
+func TestZSTDLimits(t *testing.T) {
+	subject, err := encoding.NewZSTD[*testValue]()
+	require.NoError(t, err)
+
+	writer, err := zstd.NewWriter(nil)
+	require.NoError(t, err)
+
+	var v testValue
+	v.Value = string(make([]byte, cbg.ByteArrayMaxLen*2))
+
+	var buf bytes.Buffer
+	require.NoError(t, v.MarshalCBOR(&buf))
+
+	tooLargeACompression := writer.EncodeAll(buf.Bytes(), nil)
+	// Assert the compressed size is less than 1MiB, in other words, transportable by
+	// the default GossipSub message size limit.
+	require.Less(t, len(tooLargeACompression), 1<<20)
+
+	var dest testValue
+	require.ErrorContains(t, subject.Decode(tooLargeACompression, &dest), "decompressed size exceeds configured limit")
+}