Require addressbook to only store addresses with valid ID

* p2p: peer-id -> peer_id

* update changelog

CHANGELOG_PENDING.md

@@ -25,3 +25,7 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 ### IMPROVEMENTS:
 
 ### BUG FIXES:
+
+- [crypto/merkle] [\#2756](https://github.com/tendermint/tendermint/issues/2756) Fix crypto/merkle ProofOperators.Verify to check bounds on keypath parts.
+- [mempool] fix a bug where we create a WAL despite `wal_dir` being empty
+- [p2p] \#2771 Fix `peer-id` label name in prometheus metrics

config/config.go

@@ -497,6 +497,11 @@ func (cfg *MempoolConfig) WalDir() string {
 	return rootify(cfg.WalPath, cfg.RootDir)
 }
 
+// WalEnabled returns true if the WAL is enabled.
+func (cfg *MempoolConfig) WalEnabled() bool {
+	return cfg.WalPath != ""
+}
+
 // ValidateBasic performs basic validation (checking param bounds, etc.) and
 // returns an error if any check fails.
 func (cfg *MempoolConfig) ValidateBasic() error {

consensus/state.go

@@ -1408,9 +1408,9 @@ func (cs *ConsensusState) defaultSetProposal(proposal *types.Proposal) error {
 		return nil
 	}
 
-	// Verify POLRound, which must be -1 or between 0 and proposal.Round exclusive.
-	if proposal.POLRound != -1 &&
-		(proposal.POLRound < 0 || proposal.Round <= proposal.POLRound) {
+  // Verify POLRound, which must be -1 or in range [0, proposal.Round).
+	if proposal.POLRound < -1 ||
+		(proposal.POLRound >= 0 && proposal.POLRound >= proposal.Round) {
 		return ErrInvalidProposalPOLRound
 	}
 

crypto/merkle/proof.go

@@ -43,6 +43,9 @@ func (poz ProofOperators) Verify(root []byte, keypath string, args [][]byte) (er
 	for i, op := range poz {
 		key := op.GetKey()
 		if len(key) != 0 {
+			if len(keys) == 0 {
+				return cmn.NewError("Key path has insufficient # of parts: expected no more keys but got %+v", string(key))
+			}
 			lastKey := keys[len(keys)-1]
 			if !bytes.Equal(lastKey, key) {
 				return cmn.NewError("Key mismatch on operation #%d: expected %+v but got %+v", i, string(lastKey), string(key))

crypto/merkle/proof_test.go

@@ -107,6 +107,10 @@ func TestProofOperators(t *testing.T) {
 	err = popz.Verify(bz("OUTPUT4"), "//KEY4/KEY2/KEY1", [][]byte{bz("INPUT1")})
 	assert.NotNil(t, err)
 
+	// BAD KEY 5
+	err = popz.Verify(bz("OUTPUT4"), "/KEY2/KEY1", [][]byte{bz("INPUT1")})
+	assert.NotNil(t, err)
+
 	// BAD OUTPUT 1
 	err = popz.Verify(bz("OUTPUT4_WRONG"), "/KEY4/KEY2/KEY1", [][]byte{bz("INPUT1")})
 	assert.NotNil(t, err)

mempool/mempool.go

@@ -25,12 +25,12 @@ import (
 // PreCheckFunc is an optional filter executed before CheckTx and rejects
 // transaction if false is returned. An example would be to ensure that a
 // transaction doesn't exceeded the block size.
-type PreCheckFunc func(types.Tx) bool
+type PreCheckFunc func(types.Tx) error
 
 // PostCheckFunc is an optional filter executed after CheckTx and rejects
 // transaction if false is returned. An example would be to ensure a
 // transaction doesn't require more gas than available for the block.
-type PostCheckFunc func(types.Tx, *abci.ResponseCheckTx) bool
+type PostCheckFunc func(types.Tx, *abci.ResponseCheckTx) error
 
 /*
 
@@ -68,24 +68,48 @@ var (
 	ErrMempoolIsFull = errors.New("Mempool is full")
 )
 
+// ErrPreCheck is returned when tx is too big
+type ErrPreCheck struct {
+	Reason error
+}
+
+func (e ErrPreCheck) Error() string {
+	return e.Reason.Error()
+}
+
+// IsPreCheckError returns true if err is due to pre check failure.
+func IsPreCheckError(err error) bool {
+	_, ok := err.(ErrPreCheck)
+	return ok
+}
+
 // PreCheckAminoMaxBytes checks that the size of the transaction plus the amino
 // overhead is smaller or equal to the expected maxBytes.
 func PreCheckAminoMaxBytes(maxBytes int64) PreCheckFunc {
-	return func(tx types.Tx) bool {
+	return func(tx types.Tx) error {
 		// We have to account for the amino overhead in the tx size as well
 		aminoOverhead := amino.UvarintSize(uint64(len(tx)))
-		return int64(len(tx)+aminoOverhead) <= maxBytes
+		txSize := int64(len(tx) + aminoOverhead)
+		if txSize > maxBytes {
+			return fmt.Errorf("Tx size (including amino overhead) is too big: %d, max: %d",
+				txSize, maxBytes)
+		}
+		return nil
 	}
 }
 
 // PostCheckMaxGas checks that the wanted gas is smaller or equal to the passed
-// maxGas. Returns true if maxGas is -1.
+// maxGas. Returns nil if maxGas is -1.
 func PostCheckMaxGas(maxGas int64) PostCheckFunc {
-	return func(tx types.Tx, res *abci.ResponseCheckTx) bool {
+	return func(tx types.Tx, res *abci.ResponseCheckTx) error {
 		if maxGas == -1 {
-			return true
+			return nil
 		}
-		return res.GasWanted <= maxGas
+		if res.GasWanted > maxGas {
+			return fmt.Errorf("gas wanted %d is greater than max gas %d",
+				res.GasWanted, maxGas)
+		}
+		return nil
 	}
 }
 
@@ -189,39 +213,33 @@ func WithMetrics(metrics *Metrics) MempoolOption {
 	return func(mem *Mempool) { mem.metrics = metrics }
 }
 
+// InitWAL creates a directory for the WAL file and opens a file itself.
+//
+// *panics* if can't create directory or open file.
+// *not thread safe*
+func (mem *Mempool) InitWAL() {
+	walDir := mem.config.WalDir()
+	err := cmn.EnsureDir(walDir, 0700)
+	if err != nil {
+		panic(errors.Wrap(err, "Error ensuring Mempool WAL dir"))
+	}
+	af, err := auto.OpenAutoFile(walDir + "/wal")
+	if err != nil {
+		panic(errors.Wrap(err, "Error opening Mempool WAL file"))
+	}
+	mem.wal = af
+}
+
 // CloseWAL closes and discards the underlying WAL file.
 // Any further writes will not be relayed to disk.
-func (mem *Mempool) CloseWAL() bool {
-	if mem == nil {
-		return false
-	}
-
+func (mem *Mempool) CloseWAL() {
 	mem.proxyMtx.Lock()
 	defer mem.proxyMtx.Unlock()
 
-	if mem.wal == nil {
-		return false
-	}
-	if err := mem.wal.Close(); err != nil && mem.logger != nil {
-		mem.logger.Error("Mempool.CloseWAL", "err", err)
+	if err := mem.wal.Close(); err != nil {
+		mem.logger.Error("Error closing WAL", "err", err)
 	}
 	mem.wal = nil
-	return true
-}
-
-func (mem *Mempool) InitWAL() {
-	walDir := mem.config.WalDir()
-	if walDir != "" {
-		err := cmn.EnsureDir(walDir, 0700)
-		if err != nil {
-			cmn.PanicSanity(errors.Wrap(err, "Error ensuring Mempool wal dir"))
-		}
-		af, err := auto.OpenAutoFile(walDir + "/wal")
-		if err != nil {
-			cmn.PanicSanity(errors.Wrap(err, "Error opening Mempool wal file"))
-		}
-		mem.wal = af
-	}
 }
 
 // Lock locks the mempool. The consensus must be able to hold lock to safely update.
@@ -285,8 +303,10 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
 		return ErrMempoolIsFull
 	}
 
-	if mem.preCheck != nil && !mem.preCheck(tx) {
-		return
+	if mem.preCheck != nil {
+		if err := mem.preCheck(tx); err != nil {
+			return ErrPreCheck{err}
+		}
 	}
 
 	// CACHE
@@ -336,8 +356,11 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
 	switch r := res.Value.(type) {
 	case *abci.Response_CheckTx:
 		tx := req.GetCheckTx().Tx
-		if (r.CheckTx.Code == abci.CodeTypeOK) &&
-			mem.isPostCheckPass(tx, r.CheckTx) {
+		var postCheckErr error
+		if mem.postCheck != nil {
+			postCheckErr = mem.postCheck(tx, r.CheckTx)
+		}
+		if (r.CheckTx.Code == abci.CodeTypeOK) && postCheckErr == nil {
 			mem.counter++
 			memTx := &mempoolTx{
 				counter:   mem.counter,
@@ -346,12 +369,18 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
 				tx:        tx,
 			}
 			mem.txs.PushBack(memTx)
-			mem.logger.Info("Added good transaction", "tx", TxID(tx), "res", r, "total", mem.Size())
+			mem.logger.Info("Added good transaction",
+				"tx", TxID(tx),
+				"res", r,
+				"height", memTx.height,
+				"total", mem.Size(),
+				"counter", memTx.counter,
+			)
 			mem.metrics.TxSizeBytes.Observe(float64(len(tx)))
 			mem.notifyTxsAvailable()
 		} else {
 			// ignore bad transaction
-			mem.logger.Info("Rejected bad transaction", "tx", TxID(tx), "res", r)
+			mem.logger.Info("Rejected bad transaction", "tx", TxID(tx), "res", r, "err", postCheckErr)
 			mem.metrics.FailedTxs.Add(1)
 			// remove from cache (it might be good later)
 			mem.cache.Remove(tx)
@@ -364,6 +393,7 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
 func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) {
 	switch r := res.Value.(type) {
 	case *abci.Response_CheckTx:
+		tx := req.GetCheckTx().Tx
 		memTx := mem.recheckCursor.Value.(*mempoolTx)
 		if !bytes.Equal(req.GetCheckTx().Tx, memTx.tx) {
 			cmn.PanicSanity(
@@ -374,15 +404,20 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) {
 				),
 			)
 		}
-		if (r.CheckTx.Code == abci.CodeTypeOK) && mem.isPostCheckPass(memTx.tx, r.CheckTx) {
+		var postCheckErr error
+		if mem.postCheck != nil {
+			postCheckErr = mem.postCheck(tx, r.CheckTx)
+		}
+		if (r.CheckTx.Code == abci.CodeTypeOK) && postCheckErr == nil {
 			// Good, nothing to do.
 		} else {
 			// Tx became invalidated due to newly committed block.
+			mem.logger.Info("Tx is no longer valid", "tx", TxID(tx), "res", r, "err", postCheckErr)
 			mem.txs.Remove(mem.recheckCursor)
 			mem.recheckCursor.DetachPrev()
 
 			// remove from cache (it might be good later)
-			mem.cache.Remove(req.GetCheckTx().Tx)
+			mem.cache.Remove(tx)
 		}
 		if mem.recheckCursor == mem.recheckEnd {
 			mem.recheckCursor = nil
@@ -565,10 +600,6 @@ func (mem *Mempool) recheckTxs(goodTxs []types.Tx) {
 	mem.proxyAppConn.FlushAsync()
 }
 
-func (mem *Mempool) isPostCheckPass(tx types.Tx, r *abci.ResponseCheckTx) bool {
-	return mem.postCheck == nil || mem.postCheck(tx, r)
-}
-
 //--------------------------------------------------------------------------------
 
 // mempoolTx is a transaction that successfully ran

mempool/mempool_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/stretchr/testify/require"
-	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tendermint/abci/example/counter"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
@@ -66,7 +65,13 @@ func checkTxs(t *testing.T, mempool *Mempool, count int) types.Txs {
 			t.Error(err)
 		}
 		if err := mempool.CheckTx(txBytes, nil); err != nil {
-			t.Fatalf("Error after CheckTx: %v", err)
+			// Skip invalid txs.
+			// TestMempoolFilters will fail otherwise. It asserts a number of txs
+			// returned.
+			if IsPreCheckError(err) {
+				continue
+			}
+			t.Fatalf("CheckTx failed: %v while checking #%d tx", err, i)
 		}
 	}
 	return txs
@@ -126,47 +131,29 @@ func TestMempoolFilters(t *testing.T) {
 	mempool := newMempoolWithApp(cc)
 	emptyTxArr := []types.Tx{[]byte{}}
 
-	nopPreFilter := func(tx types.Tx) bool { return true }
-	nopPostFilter := func(tx types.Tx, res *abci.ResponseCheckTx) bool { return true }
-
-	// This is the same filter we expect to be used within node/node.go and state/execution.go
-	nBytePreFilter := func(n int) func(tx types.Tx) bool {
-		return func(tx types.Tx) bool {
-			// We have to account for the amino overhead in the tx size as well
-			aminoOverhead := amino.UvarintSize(uint64(len(tx)))
-			return (len(tx) + aminoOverhead) <= n
-		}
-	}
-
-	nGasPostFilter := func(n int64) func(tx types.Tx, res *abci.ResponseCheckTx) bool {
-		return func(tx types.Tx, res *abci.ResponseCheckTx) bool {
-			if n == -1 {
-				return true
-			}
-			return res.GasWanted <= n
-		}
-	}
+	nopPreFilter := func(tx types.Tx) error { return nil }
+	nopPostFilter := func(tx types.Tx, res *abci.ResponseCheckTx) error { return nil }
 
 	// each table driven test creates numTxsToCreate txs with checkTx, and at the end clears all remaining txs.
 	// each tx has 20 bytes + amino overhead = 21 bytes, 1 gas
 	tests := []struct {
 		numTxsToCreate int
-		preFilter      func(tx types.Tx) bool
-		postFilter     func(tx types.Tx, res *abci.ResponseCheckTx) bool
+		preFilter      PreCheckFunc
+		postFilter     PostCheckFunc
 		expectedNumTxs int
 	}{
 		{10, nopPreFilter, nopPostFilter, 10},
-		{10, nBytePreFilter(10), nopPostFilter, 0},
-		{10, nBytePreFilter(20), nopPostFilter, 0},
-		{10, nBytePreFilter(21), nopPostFilter, 10},
-		{10, nopPreFilter, nGasPostFilter(-1), 10},
-		{10, nopPreFilter, nGasPostFilter(0), 0},
-		{10, nopPreFilter, nGasPostFilter(1), 10},
-		{10, nopPreFilter, nGasPostFilter(3000), 10},
-		{10, nBytePreFilter(10), nGasPostFilter(20), 0},
-		{10, nBytePreFilter(30), nGasPostFilter(20), 10},
-		{10, nBytePreFilter(21), nGasPostFilter(1), 10},
-		{10, nBytePreFilter(21), nGasPostFilter(0), 0},
+		{10, PreCheckAminoMaxBytes(10), nopPostFilter, 0},
+		{10, PreCheckAminoMaxBytes(20), nopPostFilter, 0},
+		{10, PreCheckAminoMaxBytes(21), nopPostFilter, 10},
+		{10, nopPreFilter, PostCheckMaxGas(-1), 10},
+		{10, nopPreFilter, PostCheckMaxGas(0), 0},
+		{10, nopPreFilter, PostCheckMaxGas(1), 10},
+		{10, nopPreFilter, PostCheckMaxGas(3000), 10},
+		{10, PreCheckAminoMaxBytes(10), PostCheckMaxGas(20), 0},
+		{10, PreCheckAminoMaxBytes(30), PostCheckMaxGas(20), 10},
+		{10, PreCheckAminoMaxBytes(21), PostCheckMaxGas(1), 10},
+		{10, PreCheckAminoMaxBytes(21), PostCheckMaxGas(0), 0},
 	}
 	for tcIndex, tt := range tests {
 		mempool.Update(1, emptyTxArr, tt.preFilter, tt.postFilter)
@@ -385,15 +372,12 @@ func TestMempoolCloseWAL(t *testing.T) {
 
 	// 7. Invoke CloseWAL() and ensure it discards the
 	// WAL thus any other write won't go through.
-	require.True(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
+	mempool.CloseWAL()
 	mempool.CheckTx(types.Tx([]byte("bar")), nil)
 	sum2 := checksumFile(walFilepath, t)
 	require.Equal(t, sum1, sum2, "expected no change to the WAL after invoking CloseWAL() since it was discarded")
 
-	// 8. Second CloseWAL should do nothing
-	require.False(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
-
-	// 9. Sanity check to ensure that the WAL file still exists
+	// 8. Sanity check to ensure that the WAL file still exists
 	m3, err := filepath.Glob(filepath.Join(rootDir, "*"))
 	require.Nil(t, err, "successful globbing expected")
 	require.Equal(t, 1, len(m3), "expecting the wal match in")

node/node.go

@@ -13,8 +13,8 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"github.com/tendermint/go-amino"
 
+	amino "github.com/tendermint/go-amino"
 	abci "github.com/tendermint/tendermint/abci/types"
 	bc "github.com/tendermint/tendermint/blockchain"
 	cfg "github.com/tendermint/tendermint/config"
@@ -279,7 +279,9 @@ func NewNode(config *cfg.Config,
 	)
 	mempoolLogger := logger.With("module", "mempool")
 	mempool.SetLogger(mempoolLogger)
-	mempool.InitWAL() // no need to have the mempool wal during tests
+	if config.Mempool.WalEnabled() {
+		mempool.InitWAL() // no need to have the mempool wal during tests
+	}
 	mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool)
 	mempoolReactor.SetLogger(mempoolLogger)
 
@@ -586,6 +588,11 @@ func (n *Node) OnStop() {
 	// TODO: gracefully disconnect from peers.
 	n.sw.Stop()
 
+	// stop mempool WAL
+	if n.config.Mempool.WalEnabled() {
+		n.mempoolReactor.Mempool.CloseWAL()
+	}
+
 	if err := n.transport.Close(); err != nil {
 		n.Logger.Error("Error closing transport", "err", err)
 	}

p2p/netaddress.go

@@ -218,10 +218,22 @@ func (na *NetAddress) Routable() bool {
 // For IPv4 these are either a 0 or all bits set address. For IPv6 a zero
 // address or one that matches the RFC3849 documentation address format.
 func (na *NetAddress) Valid() bool {
+	if string(na.ID) != "" {
+		data, err := hex.DecodeString(string(na.ID))
+		if err != nil || len(data) != IDByteLength {
+			return false
+		}
+	}
 	return na.IP != nil && !(na.IP.IsUnspecified() || na.RFC3849() ||
 		na.IP.Equal(net.IPv4bcast))
 }
 
+// HasID returns true if the address has an ID.
+// NOTE: It does not check whether the ID is valid or not.
+func (na *NetAddress) HasID() bool {
+	return string(na.ID) != ""
+}
+
 // Local returns true if it is a local address.
 func (na *NetAddress) Local() bool {
 	return na.IP.IsLoopback() || zero4.Contains(na.IP)

p2p/peer.go

@@ -240,7 +240,7 @@ func (p *peer) Send(chID byte, msgBytes []byte) bool {
 	}
 	res := p.mconn.Send(chID, msgBytes)
 	if res {
-		p.metrics.PeerSendBytesTotal.With("peer-id", string(p.ID())).Add(float64(len(msgBytes)))
+		p.metrics.PeerSendBytesTotal.With("peer_id", string(p.ID())).Add(float64(len(msgBytes)))
 	}
 	return res
 }
@@ -255,7 +255,7 @@ func (p *peer) TrySend(chID byte, msgBytes []byte) bool {
 	}
 	res := p.mconn.TrySend(chID, msgBytes)
 	if res {
-		p.metrics.PeerSendBytesTotal.With("peer-id", string(p.ID())).Add(float64(len(msgBytes)))
+		p.metrics.PeerSendBytesTotal.With("peer_id", string(p.ID())).Add(float64(len(msgBytes)))
 	}
 	return res
 }
@@ -330,7 +330,7 @@ func (p *peer) metricsReporter() {
 				sendQueueSize += float64(chStatus.SendQueueSize)
 			}
 
-			p.metrics.PeerPendingSendBytes.With("peer-id", string(p.ID())).Set(sendQueueSize)
+			p.metrics.PeerPendingSendBytes.With("peer_id", string(p.ID())).Set(sendQueueSize)
 		case <-p.Quit():
 			return
 		}

p2p/pex/addrbook.go

@@ -652,6 +652,10 @@ func (a *addrBook) addAddress(addr, src *p2p.NetAddress) error {
 		return ErrAddrBookInvalidAddr{addr}
 	}
 
+	if !addr.HasID() {
+		return ErrAddrBookInvalidAddrNoID{addr}
+	}
+
 	// TODO: we should track ourAddrs by ID and by IP:PORT and refuse both.
 	if _, ok := a.ourAddrs[addr.String()]; ok {
 		return ErrAddrBookSelf{addr}

p2p/pex/errors.go

@@ -54,3 +54,11 @@ type ErrAddrBookInvalidAddr struct {
 func (err ErrAddrBookInvalidAddr) Error() string {
 	return fmt.Sprintf("Cannot add invalid address %v", err.Addr)
 }
+
+type ErrAddrBookInvalidAddrNoID struct {
+	Addr *p2p.NetAddress
+}
+
+func (err ErrAddrBookInvalidAddrNoID) Error() string {
+	return fmt.Sprintf("Cannot add address with no ID %v", err.Addr)
+}

types/vote_set.go

@@ -158,7 +158,7 @@ func (voteSet *VoteSet) addVote(vote *Vote) (added bool, err error) {
 	if (vote.Height != voteSet.height) ||
 		(vote.Round != voteSet.round) ||
 		(vote.Type != voteSet.type_) {
-		return false, errors.Wrapf(ErrVoteUnexpectedStep, "Got %d/%d/%d, expected %d/%d/%d",
+		return false, errors.Wrapf(ErrVoteUnexpectedStep, "Expected %d/%d/%d, but got %d/%d/%d",
 			voteSet.height, voteSet.round, voteSet.type_,
 			vote.Height, vote.Round, vote.Type)
 	}