node: create SafeRead function to replace io.ReadAll (#4499)

* node: create SafeRead function to replace io.ReadAll

* add unit test

* increase limit and add code comments

* fix syntax error from bad rebase

* fix importShadow violations

---------

Co-authored-by: John Saigle <john@asymmetric.re>

.golangci.yml

@@ -17,6 +17,7 @@ linters:
     # Enum and maps used on switch statements are exhaustive
     - exhaustive
     - exhaustruct
+    - forbidigo
     - forcetypeassert
     - gocritic
     - gosec
@@ -119,6 +120,12 @@ linters:
         - .+/cobra\.Command$
         - .+/http\.Client$
         - .+/prometheus.+
+    forbidigo:
+      forbid:
+        - pattern: 'io\.ReadAll'
+          msg: "method can cause DoS for large user-controlled inputs. Use common.SafeRead instead."
+      exclude-godoc-examples: true
+      analyze-types: true
     gocritic:
       disable-all: true
     # disabled-checks:

node/cmd/ccq/permissions.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"path"
 	"strings"
@@ -198,7 +197,7 @@ func parseConfigFile(fileName string, env common.Environment) (PermissionsMap, e
 	}
 	defer jsonFile.Close()
 
-	byteValue, err := io.ReadAll(jsonFile)
+	byteValue, err := common.SafeRead(jsonFile)
 	if err != nil {
 		return nil, fmt.Errorf(`failed to read permissions file "%s": %w`, fileName, err)
 	}

node/hack/encrypt/encrypt.go

@@ -4,7 +4,6 @@ import (
 	"crypto/rand"
 	"encoding/base64"
 	"fmt"
-	"io"
 	"log"
 	"os"
 
@@ -12,7 +11,7 @@ import (
 )
 
 func main() {
-	in, err := io.ReadAll(os.Stdin)
+	in, err := common.SafeRead(os.Stdin)
 	if err != nil {
 		log.Fatalf("failed to read stdin: %v", err)
 	}

node/hack/release_verification/guardian_vaa_stats.go

@@ -15,12 +15,12 @@ package main
 import (
 	"encoding/base64"
 	"fmt"
-	"io"
 	"net/http"
 	"os"
 	"strings"
 	"time"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/tidwall/gjson"
 	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 )
@@ -45,7 +45,7 @@ func getValidatorIndexForChain(chainId vaa.ChainID, onlyafter time.Time) (map[ui
 	}
 	defer res.Body.Close()
 
-	body, err := io.ReadAll(res.Body)
+	body, err := common.SafeRead(res.Body)
 	if err != nil {
 		return nil, err
 	}

node/hack/repair_eth/repair_eth.go

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
-	"io"
 	"log"
 	"math"
 	"net/http"
@@ -155,7 +154,7 @@ func getCurrentHeight(chainId vaa.ChainID, ctx context.Context, c *http.Client,
 	var r struct {
 		Result string `json:"result"`
 	}
-	body, err := io.ReadAll(resp.Body)
+	body, err := common.SafeRead(resp.Body)
 	if err != nil {
 		return 0, fmt.Errorf("failed to read response body: %w", err)
 	}
@@ -198,7 +197,7 @@ func getLogs(chainId vaa.ChainID, ctx context.Context, c *http.Client, api, key,
 
 	var r logResponse
 
-	body, err := io.ReadAll(resp.Body)
+	body, err := common.SafeRead(resp.Body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read response body: %w", err)
 	}

node/pkg/adminrpc/adminserver.go

@@ -8,7 +8,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"math"
 	"math/big"
 	"math/rand"
@@ -1296,7 +1295,7 @@ func (s *nodePrivilegedService) GetAndObserveMissingVAAs(ctx context.Context, re
 	defer results.Body.Close()
 
 	// Collect the results
-	resBody, err := io.ReadAll(results.Body)
+	resBody, err := common.SafeRead(results.Body)
 	if err != nil {
 		fmt.Printf("GetAndObserveMissingVAAs: could not read response body: %s\n", err)
 		return nil, err

node/pkg/altpub/end2end_test.go

@@ -6,13 +6,13 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"slices"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
 	ethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -222,7 +222,7 @@ func (s *Server) run() {
 
 // handleSignedObservationBatch is the handler for a signed observation.
 func (s *Server) handleSignedObservationBatch(w http.ResponseWriter, r *http.Request) {
-	body, err := io.ReadAll(r.Body)
+	body, err := common.SafeRead(r.Body)
 	r.Body.Close()
 	if err != nil {
 		s.logger.Fatal("error extracting body", zap.Error(err))

node/pkg/common/armoredKey.go

@@ -4,7 +4,6 @@ import (
 	"crypto/ecdsa"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 
 	ethcrypto "github.com/ethereum/go-ethereum/crypto"
@@ -39,7 +38,7 @@ func LoadArmoredKey(filename string, blockType string, unsafeDevMode bool) (*ecd
 		return nil, fmt.Errorf("invalid block type: %s", p.Type)
 	}
 
-	b, err := io.ReadAll(p.Body)
+	b, err := SafeRead(p.Body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read file: %w", err)
 	}

node/pkg/common/chainlock.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"math"
 	"time"
 
@@ -78,6 +79,13 @@ func (e ErrInputSize) Error() string {
 	return fmt.Sprintf("wrong size: %s. expected >= %d bytes, got %d", e.msg, marshaledMsgLenMin, e.got)
 }
 
+// MaxSafeInputSize defines the maximum safe size for untrusted input from `io` Readers.
+// It should be configured so that it can comfortably contain all valid reads while
+// providing a strict upper bound to prevent unlimited reads.
+const MaxSafeInputSize = 128 * 1024 * 1024 // 128MB (arbitrary)
+
+var ErrInputTooLarge = errors.New("input data exceeds maximum allowed size")
+
 // The `VerificationState` is the result of applying transfer verification to the transaction associated with the `MessagePublication`.
 // While this could likely be extended to additional security controls in the future, it is only used for `txverifier` at present.
 // Consequently, its status should be set to `NotVerified` or `NotApplicable` for all messages that aren't token transfers.
@@ -595,3 +603,29 @@ func (msg *MessagePublication) VAAHash() string {
 func validBinaryBool(b byte) bool {
 	return b == 0x00 || b == 0x01
 }
+
+// SafeRead reads from r with a size limit to prevent memory exhaustion attacks.
+// It returns an error if the input exceeds MaxSafeInputSize.
+func SafeRead(r io.Reader) ([]byte, error) {
+	// Create a LimitReader that allows reading up to MaxSafeInputSize + 1 bytes.
+	// The extra byte is specifically to detect if the input stream *exceeds* MaxSafeInputSize.
+	lr := io.LimitReader(r, MaxSafeInputSize+1)
+
+	//nolint:forbidigo // SafeRead is intended as a convenient and safe wrapper for ReadAll.
+	b, err := io.ReadAll(lr)
+	if err != nil {
+		// Propagate any actual read errors from the underlying reader.
+		return nil, err
+	}
+
+	// If the length of the read bytes is greater than MaxSafeInputSize,
+	// it means the original reader contained more data than allowed.
+	// In this case, we return an error instead of silently truncating.
+	if len(b) > MaxSafeInputSize {
+		return nil, ErrInputTooLarge
+	}
+
+	// If err was nil and len(b) <= MaxSafeInputSize, it means we read all
+	// available input (or up to the limit) without exceeding the maximum.
+	return b, nil
+}

node/pkg/common/chainlock_test.go

@@ -1,9 +1,12 @@
 package common
 
 import (
+	"bytes"
 	"encoding/binary"
+	"io"
 	"math"
 	"math/big"
+	"os"
 	"testing"
 	"time"
 
@@ -32,20 +35,20 @@ const (
 )
 
 func encodePayloadBytes(payload *vaa.TransferPayloadHdr) []byte {
-	bytes := make([]byte, 101)
-	bytes[0] = payload.Type
+	bz := make([]byte, 101)
+	bz[0] = payload.Type
 
 	amtBytes := payload.Amount.Bytes()
 	if len(amtBytes) > 32 {
 		panic("amount will not fit in 32 bytes!")
 	}
-	copy(bytes[33-len(amtBytes):33], amtBytes)
+	copy(bz[33-len(amtBytes):33], amtBytes)
 
-	copy(bytes[33:65], payload.OriginAddress.Bytes())
-	binary.BigEndian.PutUint16(bytes[65:67], uint16(payload.OriginChain))
-	copy(bytes[67:99], payload.TargetAddress.Bytes())
-	binary.BigEndian.PutUint16(bytes[99:101], uint16(payload.TargetChain))
-	return bytes
+	copy(bz[33:65], payload.OriginAddress.Bytes())
+	binary.BigEndian.PutUint16(bz[65:67], uint16(payload.OriginChain))
+	copy(bz[67:99], payload.TargetAddress.Bytes())
+	binary.BigEndian.PutUint16(bz[99:101], uint16(payload.TargetChain))
+	return bz
 }
 
 // makeTestMsgPub is a helper function that generates a Message Publication.
@@ -91,11 +94,11 @@ func TestRoundTripMarshal(t *testing.T) {
 	orig := makeTestMsgPub(t)
 	var loaded MessagePublication
 
-	bytes, writeErr := orig.MarshalBinary()
+	bz, writeErr := orig.MarshalBinary()
 	require.NoError(t, writeErr)
-	t.Logf("marshaled bytes: %x", bytes)
+	t.Logf("marshaled bytes: %x", bz)
 
-	readErr := loaded.UnmarshalBinary(bytes)
+	readErr := loaded.UnmarshalBinary(bz)
 	require.NoError(t, readErr)
 
 	require.Equal(t, *orig, loaded)
@@ -402,10 +405,10 @@ func TestDeprecatedSerializeAndDeserializeOfMessagePublication(t *testing.T) {
 		verificationState: Anomalous,
 	}
 
-	bytes, err := msg1.Marshal()
+	bz, err := msg1.Marshal()
 	require.NoError(t, err)
 
-	msg2, err := UnmarshalMessagePublication(bytes)
+	msg2, err := UnmarshalMessagePublication(bz)
 	require.NoError(t, err)
 
 	require.Equal(t, msg1.TxID, msg2.TxID)
@@ -460,10 +463,10 @@ func TestSerializeAndDeserializeOfMessagePublicationWithEmptyTxID(t *testing.T)
 		ConsistencyLevel: 32,
 	}
 
-	bytes, err := msg1.Marshal()
+	bz, err := msg1.Marshal()
 	require.NoError(t, err)
 
-	msg2, err := UnmarshalMessagePublication(bytes)
+	msg2, err := UnmarshalMessagePublication(bz)
 	require.NoError(t, err)
 	assert.Equal(t, msg1, msg2)
 
@@ -507,10 +510,10 @@ func TestSerializeAndDeserializeOfMessagePublicationWithArbitraryTxID(t *testing
 		ConsistencyLevel: 32,
 	}
 
-	bytes, err := msg1.Marshal()
+	bz, err := msg1.Marshal()
 	require.NoError(t, err)
 
-	msg2, err := UnmarshalMessagePublication(bytes)
+	msg2, err := UnmarshalMessagePublication(bz)
 	require.NoError(t, err)
 	assert.Equal(t, msg1, msg2)
 
@@ -568,10 +571,10 @@ func TestSerializeAndDeserializeOfMessagePublicationWithBigPayload(t *testing.T)
 		ConsistencyLevel: 32,
 	}
 
-	bytes, err := msg1.Marshal()
+	bz, err := msg1.Marshal()
 	require.NoError(t, err)
 
-	msg2, err := UnmarshalMessagePublication(bytes)
+	msg2, err := UnmarshalMessagePublication(bz)
 	require.NoError(t, err)
 
 	assert.Equal(t, msg1, msg2)
@@ -609,11 +612,11 @@ func TestMarshalUnmarshalJSONOfMessagePublication(t *testing.T) {
 		ConsistencyLevel: 32,
 	}
 
-	bytes, err := msg1.MarshalJSON()
+	bz, err := msg1.MarshalJSON()
 	require.NoError(t, err)
 
 	var msg2 MessagePublication
-	err = msg2.UnmarshalJSON(bytes)
+	err = msg2.UnmarshalJSON(bz)
 	require.NoError(t, err)
 	assert.Equal(t, *msg1, msg2)
 
@@ -655,11 +658,11 @@ func TestMarshalUnmarshalJSONOfMessagePublicationWithArbitraryTxID(t *testing.T)
 		ConsistencyLevel: 32,
 	}
 
-	bytes, err := msg1.MarshalJSON()
+	bz, err := msg1.MarshalJSON()
 	require.NoError(t, err)
 
 	var msg2 MessagePublication
-	err = msg2.UnmarshalJSON(bytes)
+	err = msg2.UnmarshalJSON(bz)
 	require.NoError(t, err)
 	assert.Equal(t, *msg1, msg2)
 
@@ -802,3 +805,54 @@ func TestMessagePublication_SetVerificationState(t *testing.T) {
 		})
 	}
 }
+
+func TestSafeRead(t *testing.T) {
+	tests := []struct {
+		name    string
+		size    int
+		wantErr bool
+	}{
+		{
+			"happy path",
+			MaxSafeInputSize,
+			false,
+		},
+		{
+			"error: too big",
+			MaxSafeInputSize + 1,
+			true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create temporary file and write bytes to it
+			tmp := os.TempDir()
+
+			f, err := os.CreateTemp(tmp, "tmpfile-")
+			require.NoError(t, err)
+
+			defer f.Close()
+			defer os.Remove(f.Name())
+
+			// Fill slice with zeroes.
+			data := make([]byte, tt.size)
+			if _, err := f.Write(data); err != nil {
+				require.NoError(t, err)
+			}
+
+			// File pointer is at EOF at this point. Reset to the start.
+			_, err = f.Seek(0, io.SeekStart)
+			require.NoError(t, err)
+
+			got, err := SafeRead(f)
+			if tt.wantErr {
+				require.Error(t, err, "SafeRead() should have returned an error")
+				require.Nil(t, got, "got should be nil when error occurs")
+			} else {
+				require.NoError(t, err, "SafeRead() should not have returned an error")
+				require.NotNil(t, got, "got should not be nil when no error occurs")
+				require.True(t, bytes.Equal(got, data), "bytes read are not equal to bytes written")
+			}
+		})
+	}
+}

node/pkg/governor/governor_prices.go

@@ -10,7 +10,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"math/big"
 	"math/rand"
 	"net/http"
@@ -270,7 +269,7 @@ func (gov *ChainGovernor) queryCoinGeckoChunk(query string) (map[string]interfac
 		}
 	}()
 
-	responseData, err := io.ReadAll(response.Body)
+	responseData, err := common.SafeRead(response.Body)
 	if err != nil {
 		return result, fmt.Errorf("failed to read CoinGecko response: %w", err)
 	}

node/pkg/guardiansigner/filesigner.go

@@ -5,13 +5,13 @@ import (
 	"crypto/ecdsa"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 
 	"github.com/ethereum/go-ethereum/crypto"
 	ethcrypto "github.com/ethereum/go-ethereum/crypto"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	nodev1 "github.com/certusone/wormhole/node/pkg/proto/node/v1"
 	"golang.org/x/crypto/openpgp/armor" //nolint:staticcheck // Package is deprecated but we need it in the codebase still.
 )
@@ -49,7 +49,7 @@ func NewFileSigner(_ context.Context, unsafeDevMode bool, signerKeyPath string)
 		return nil, fmt.Errorf("invalid block type: %s", p.Type)
 	}
 
-	b, err := io.ReadAll(p.Body)
+	b, err := common.SafeRead(p.Body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read file: %w", err)
 	}

node/pkg/txverifier/sui.go

@@ -7,11 +7,11 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"math/big"
 	"net/http"
 	"strings"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 	"go.uber.org/zap"
 )
@@ -238,7 +238,7 @@ func suiApiRequest[T SuiApiResponse](rpc string, method string, params string) (
 	defer resp.Body.Close()
 
 	// Read the response
-	body, err := io.ReadAll(resp.Body)
+	body, err := common.SafeRead(resp.Body)
 	if err != nil {
 		return defaultT, fmt.Errorf("cannot read response: %w", err)
 	}

node/pkg/watchers/aptos/watcher.go

@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
-	"io"
 	"math"
 	"net/http"
 	"time"
@@ -276,7 +275,7 @@ func (e *Watcher) retrievePayload(s string) ([]byte, error) {
 		return nil, err
 	}
 	defer res.Body.Close()
-	body, err := io.ReadAll(res.Body)
+	body, err := common.SafeRead(res.Body)
 	if err != nil {
 		return nil, err
 	}

node/pkg/watchers/cosmwasm/watcher.go

@@ -5,7 +5,6 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"fmt"
-	"io"
 	"math"
 	"net/http"
 	"strconv"
@@ -213,7 +212,7 @@ func (e *Watcher) Run(ctx context.Context) error {
 					logger.Error("query latest block response error", zap.String("network", networkName), zap.Error(err))
 					continue
 				}
-				blocksBody, err := io.ReadAll(resp.Body)
+				blocksBody, err := common.SafeRead(resp.Body)
 				if err != nil {
 					logger.Error("query latest block response read error", zap.String("network", networkName), zap.Error(err))
 					errC <- err //nolint:channelcheck // The watcher will exit anyway
@@ -266,7 +265,7 @@ func (e *Watcher) Run(ctx context.Context) error {
 					logger.Error("query tx response error", zap.String("network", networkName), zap.Error(err))
 					continue
 				}
-				txBody, err := io.ReadAll(resp.Body)
+				txBody, err := common.SafeRead(resp.Body)
 				if err != nil {
 					logger.Error("query tx response read error", zap.String("network", networkName), zap.Error(err))
 					resp.Body.Close()

node/pkg/watchers/ibc/watcher.go

@@ -6,7 +6,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"io"
 	"math"
 	"net/http"
 	"net/url"
@@ -390,7 +389,7 @@ func (w *Watcher) handleQueryBlockHeight(ctx context.Context, queryUrl string) e
 			if err != nil {
 				return fmt.Errorf("failed to query latest block: %w", err)
 			}
-			body, err := io.ReadAll(resp.Body)
+			body, err := common.SafeRead(resp.Body)
 			resp.Body.Close()
 			if err != nil {
 				return fmt.Errorf("failed to read latest block body: %w", err)
@@ -455,7 +454,7 @@ func (w *Watcher) handleObservationRequests(ctx context.Context, ce *chainEntry)
 				w.logger.Error("query tx response error", zap.String("chain", ce.chainName), zap.Error(err))
 				continue
 			}
-			txBody, err := io.ReadAll(resp.Body)
+			txBody, err := common.SafeRead(resp.Body)
 			if err != nil {
 				w.logger.Error("query tx response read error", zap.String("chain", ce.chainName), zap.Error(err))
 				resp.Body.Close()
@@ -747,7 +746,7 @@ func (w *Watcher) queryChannelIdToChainIdMapping() (map[string]vaa.ChainID, erro
 	if err != nil {
 		return nil, fmt.Errorf("query failed: %w", err)
 	}
-	body, err := io.ReadAll(resp.Body)
+	body, err := common.SafeRead(resp.Body)
 	if err != nil {
 		return nil, fmt.Errorf("read failed: %w", err)
 	}

node/pkg/watchers/near/nearapi/mock/mock_server.go

@@ -18,6 +18,7 @@ import (
 	"os"
 	"path/filepath"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"go.uber.org/zap"
 )
 
@@ -48,7 +49,7 @@ func panicIfError(e error) {
 }
 
 func serveCache(w http.ResponseWriter, req *http.Request, cacheDir string) (string, error) {
-	reqBody, err := io.ReadAll(req.Body)
+	reqBody, err := common.SafeRead(req.Body)
 	if err != nil {
 		return "", errors.New("error reading request")
 	}
@@ -72,7 +73,7 @@ func returnFile(w http.ResponseWriter, fileName string) {
 }
 
 func (s *ForwardingCachingServer) ProxyReq(_ *zap.Logger, req *http.Request) (*http.Request, error) {
-	reqBody, err := io.ReadAll(req.Body)
+	reqBody, err := common.SafeRead(req.Body)
 
 	if err != nil {
 		return nil, err
@@ -118,7 +119,7 @@ func (s *ForwardingCachingServer) RewriteReq(reqBody []byte) []byte {
 }
 
 func (s *ForwardingCachingServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	origReqBody, err := io.ReadAll(req.Body)
+	origReqBody, err := common.SafeRead(req.Body)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
@@ -156,7 +157,7 @@ func (s *ForwardingCachingServer) ServeHTTP(w http.ResponseWriter, req *http.Req
 			return
 		}
 		defer resp.Body.Close()
-		respBody, err := io.ReadAll(resp.Body)
+		respBody, err := common.SafeRead(resp.Body)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadGateway)
 			return

node/pkg/watchers/near/nearapi/nearapi.go

@@ -5,11 +5,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"math/rand"
 	"net/http"
 	"time"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/mr-tron/base58"
 )
 
@@ -86,7 +86,7 @@ func (n HttpNearRpc) Query(ctx context.Context, s string) ([]byte, error) {
 
 			if err == nil {
 				defer resp.Body.Close()
-				result, err := io.ReadAll(resp.Body)
+				result, err := common.SafeRead(resp.Body)
 				if resp.StatusCode == 200 {
 					return result, err
 				}

node/pkg/watchers/sui/watcher.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"math"
 	"net/http"
 	"strconv"
@@ -641,7 +640,7 @@ func (w *Watcher) createAndExecReq(ctx context.Context, payload string) ([]byte,
 	if err != nil {
 		return retVal, fmt.Errorf("createAndExecReq failed to post: %w", err)
 	}
-	body, err := io.ReadAll(resp.Body)
+	body, err := common.SafeRead(resp.Body)
 	if err != nil {
 		return retVal, fmt.Errorf("createAndExecReq failed to read: %w", err)
 	}