peer,server: Hash mix messages ASAP

When a mixing message (implementing the WriteHash(hash.Hash) method) is read
from another peer, it is improper to use its Hash() method before calculating
and storing the hash using WriteHash.  Do this as early as possible in
readMessage in the peer package.  This avoids the need for callers to remember
to do this step manually in the OnMix* callbacks.

peer/go.mod

@@ -5,6 +5,7 @@ go 1.17
 require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/decred/dcrd/chaincfg/chainhash v1.0.4
+	github.com/decred/dcrd/crypto/blake256 v1.0.1
 	github.com/decred/dcrd/lru v1.1.2
 	github.com/decred/dcrd/txscript/v4 v4.1.1
 	github.com/decred/dcrd/wire v1.7.0
@@ -15,7 +16,6 @@ require (
 require (
 	github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412 // indirect
 	github.com/dchest/siphash v1.2.3 // indirect
-	github.com/decred/dcrd/crypto/blake256 v1.0.1 // indirect
 	github.com/decred/dcrd/crypto/ripemd160 v1.0.2 // indirect
 	github.com/decred/dcrd/dcrec v1.0.1 // indirect
 	github.com/decred/dcrd/dcrec/edwards/v2 v2.0.3 // indirect

peer/peer.go

@@ -9,6 +9,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"hash"
 	"io"
 	"math/rand"
 	"net"
@@ -20,6 +21,7 @@ import (
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/decred/dcrd/chaincfg/chainhash"
+	"github.com/decred/dcrd/crypto/blake256"
 	"github.com/decred/dcrd/lru"
 	"github.com/decred/dcrd/wire"
 	"github.com/decred/go-socks/socks"
@@ -457,6 +459,11 @@ type Peer struct {
 
 	conn net.Conn
 
+	// blake256Hasher is the hash.Hash object that is used by readMessage
+	// to calculate the hash of read mixing messages.  Every peer's hasher
+	// is a distinct object and does not require locking.
+	blake256Hasher hash.Hash
+
 	// These fields are set at creation time and never modified, so they are
 	// safe to read from concurrently without a mutex.
 	addr    string
@@ -967,6 +974,15 @@ func (p *Peer) handlePongMsg(msg *wire.MsgPong) {
 	p.statsMtx.Unlock()
 }
 
+// hashable is a wire message which can be hashed but requires the hash to be
+// calculated upfront after creating any message or (importantly for peer)
+// deserializing a message off the wire.
+type hashable interface {
+	wire.Message
+	WriteHash(hash.Hash)
+	Hash() chainhash.Hash
+}
+
 // readMessage reads the next wire message from the peer with logging.
 func (p *Peer) readMessage() (wire.Message, []byte, error) {
 	err := p.conn.SetReadDeadline(time.Now().Add(p.cfg.IdleTimeout))
@@ -976,10 +992,19 @@ func (p *Peer) readMessage() (wire.Message, []byte, error) {
 	n, msg, buf, err := wire.ReadMessageN(p.conn, p.ProtocolVersion(),
 		p.cfg.Net)
 	atomic.AddUint64(&p.bytesReceived, uint64(n))
+
+	// Calculate and store the message hash of any mixing message
+	// immediately after deserializing it.
+	if err == nil {
+		if msg, ok := msg.(hashable); ok {
+			msg.WriteHash(p.blake256Hasher)
+		}
+	}
+
 	if p.cfg.Listeners.OnRead != nil {
 		p.cfg.Listeners.OnRead(p, n, msg, err)
 	}
-	if err != nil {
+	if err != nil { // Check ReadMessageN error
 		return nil, nil, err
 	}
 
@@ -2171,6 +2196,7 @@ func newPeerBase(cfgOrig *Config, inbound bool) *Peer {
 	}
 
 	p := Peer{
+		blake256Hasher:  blake256.New(),
 		inbound:         inbound,
 		knownInventory:  lru.NewCache(maxKnownInventory),
 		stallControl:    make(chan stallControlMsg, 1), // nonblocking sync

server.go

@@ -14,7 +14,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"hash"
 	"math"
 	"net"
 	"os"
@@ -34,7 +33,6 @@ import (
 	"github.com/decred/dcrd/chaincfg/v3"
 	"github.com/decred/dcrd/connmgr/v3"
 	"github.com/decred/dcrd/container/apbf"
-	"github.com/decred/dcrd/crypto/blake256"
 	"github.com/decred/dcrd/database/v3"
 	"github.com/decred/dcrd/dcrutil/v4"
 	"github.com/decred/dcrd/internal/blockchain"
@@ -641,12 +639,6 @@ type serverPeer struct {
 	// data item requests that still need to be served.
 	getDataQueue              chan []*wire.InvVect
 	numPendingGetDataItemReqs atomic.Uint32
-
-	// blake256Hasher is the hash.Hash object that is reused by the
-	// message listener callbacks (the serverPeer's On* methods) to hash
-	// mixing messages.  It does not require locking, as the message
-	// listeners are executed serially.
-	blake256Hasher hash.Hash
 }
 
 // newServerPeer returns a new serverPeer instance. The peer needs to be set by
@@ -661,7 +653,6 @@ func newServerPeer(s *server, isPersistent bool) *serverPeer {
 		blockProcessed:  make(chan struct{}, 1),
 		mixMsgProcessed: make(chan error, 1),
 		getDataQueue:    make(chan []*wire.InvVect, maxConcurrentGetDataReqs),
-		blake256Hasher:  blake256.New(),
 	}
 }
 
@@ -1706,12 +1697,8 @@ func (sp *serverPeer) onMixMessage(msg mixing.Message) {
 		return
 	}
 
-	// Calculate the message hash, so it can be added to known inventory
-	// and used by the sync manager.
-	msg.WriteHash(sp.blake256Hasher)
-	hash := msg.Hash()
-
 	// Add the message to the known inventory for the peer.
+	hash := msg.Hash()
 	iv := wire.NewInvVect(wire.InvTypeMix, &hash)
 	sp.AddKnownInventory(iv)