p2p: set RecvMessageCapacity to maxMsgSize in all reactors

to prevent malicious nodes from sending us large messages (~21MB, which
is the default `RecvMessageCapacity`)

This allows us to remove unnecessary `maxMsgSize` check in `decodeMsg`. Since each channel has a msg capacity set to `maxMsgSize`, there's no need to check it again in `decodeMsg`.

Closes #1503

blockchain/v0/reactor.go

@@ -394,9 +394,6 @@ func RegisterBlockchainMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg BlockchainMessage, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

blockchain/v1/reactor.go

@@ -539,9 +539,6 @@ func RegisterBlockchainMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg BlockchainMessage, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

blockchain/v2/reactor.go

@@ -453,9 +453,6 @@ func RegisterBlockchainMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg BlockchainMessage, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

consensus/reactor.go

@@ -1404,9 +1404,6 @@ func RegisterMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg Message, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

evidence/reactor.go

@@ -51,6 +51,7 @@ func (evR *Reactor) GetChannels() []*p2p.ChannelDescriptor {
 		{
 			ID:       EvidenceChannel,
 			Priority: 5,
+			RecvMessageCapacity: maxMsgSize,
 		},
 	}
 }
@@ -235,9 +236,6 @@ func RegisterMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg Message, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

mempool/reactor.go

@@ -137,10 +137,12 @@ func (memR *Reactor) OnStart() error {
 // GetChannels implements Reactor.
 // It returns the list of channels for this reactor.
 func (memR *Reactor) GetChannels() []*p2p.ChannelDescriptor {
+	maxMsgSize := calcMaxMsgSize(memR.config.MaxTxBytes)
 	return []*p2p.ChannelDescriptor{
 		{
-			ID:       MempoolChannel,
-			Priority: 5,
+			ID:                  MempoolChannel,
+			Priority:            5,
+			RecvMessageCapacity: maxMsgSize,
 		},
 	}
 }
@@ -271,10 +273,6 @@ func RegisterMessages(cdc *amino.Codec) {
 }
 
 func (memR *Reactor) decodeMsg(bz []byte) (msg Message, err error) {
-	maxMsgSize := calcMaxMsgSize(memR.config.MaxTxBytes)
-	if l := len(bz); l > maxMsgSize {
-		return msg, ErrTxTooLarge{maxMsgSize, l}
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }

p2p/pex/pex_reactor.go

@@ -180,9 +180,10 @@ func (r *Reactor) OnStop() {
 func (r *Reactor) GetChannels() []*conn.ChannelDescriptor {
 	return []*conn.ChannelDescriptor{
 		{
-			ID:                PexChannel,
-			Priority:          1,
-			SendQueueCapacity: 10,
+			ID:                  PexChannel,
+			Priority:            1,
+			SendQueueCapacity:   10,
+			RecvMessageCapacity: maxMsgSize,
 		},
 	}
 }
@@ -771,9 +772,6 @@ func RegisterMessages(cdc *amino.Codec) {
 }
 
 func decodeMsg(bz []byte) (msg Message, err error) {
-	if len(bz) > maxMsgSize {
-		return msg, fmt.Errorf("msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
-	}
 	err = cdc.UnmarshalBinaryBare(bz, &msg)
 	return
 }