[FAB-10617] Add writeset validation check during commit

Fabric version 1.1, missed one check in the couchdb key-values
during simulation. Couchdb does not allow an underscore as a
first character in the key or first-level fields name in the
JSON value. The implication of this is that during commit, when
the write-set is applied to the couchdb, the peer panic happens.

The problem is severe because this brings network at halt. This
is because, upon peer startup, peer again tries to process the
last block and hence gets into an endless cycle of panic and
restart.

This CR adds a check in the commit path for the validity of the
write-set and marks the transaction as invalid.

Change-Id: I36070bdba0a5012d6e99bc311e84be13863e7d8b
Signed-off-by: manish <manish.sethi@gmail.com>
(cherry picked from commit 5d8a35e)

Author: manish-sethi

core/ledger/kvledger/txmgmt/statedb/statedb.go

@@ -45,6 +45,9 @@ type VersionedDB interface {
 	GetLatestSavePoint() (*version.Height, error)
 	// ValidateKeyValue tests whether the key and value is supported by the db implementation.
 	// For instance, leveldb supports any bytes for the key while the couchdb supports only valid utf-8 string
+	// TODO make the function ValidateKeyValue return a specific error say ErrInvalidKeyValue
+	// However, as of now, the both implementations of this function (leveldb and couchdb) are deterministic in returing an error
+	// i.e., an error is returned only if the key-value are found to be invalid for the underlying db
 	ValidateKeyValue(key string, value []byte) error
 	// BytesKeySuppoted returns true if the implementation (underlying db) supports the any bytes to be used as key.
 	// For instance, leveldb supports any bytes for the key while the couchdb supports only valid utf-8 string

core/ledger/kvledger/txmgmt/validator/valimpl/default_impl.go

@@ -24,13 +24,14 @@ var logger = flogging.MustGetLogger("valimpl")
 // valinternal.InternalValidator) such as statebased validator
 type DefaultImpl struct {
 	txmgr txmgr.TxMgr
+	db    privacyenabledstate.DB
 	valinternal.InternalValidator
 }
 
 // NewStatebasedValidator constructs a validator that internally manages statebased validator and in addition
 // handles the tasks that are agnostic to a particular validation scheme such as parsing the block and handling the pvt data
 func NewStatebasedValidator(txmgr txmgr.TxMgr, db privacyenabledstate.DB) validator.Validator {
-	return &DefaultImpl{txmgr, statebasedval.NewValidator(db)}
+	return &DefaultImpl{txmgr, db, statebasedval.NewValidator(db)}
 }
 
 // ValidateAndPrepareBatch implements the function in interface validator.Validator
@@ -44,7 +45,7 @@ func (impl *DefaultImpl) ValidateAndPrepareBatch(blockAndPvtdata *ledger.BlockAn
 	var err error
 
 	logger.Debug("preprocessing ProtoBlock...")
-	if internalBlock, err = preprocessProtoBlock(impl.txmgr, block, doMVCCValidation); err != nil {
+	if internalBlock, err = preprocessProtoBlock(impl.txmgr, impl.db.ValidateKeyValue, block, doMVCCValidation); err != nil {
 		return nil, err
 	}
 

core/ledger/kvledger/txmgmt/validator/valimpl/helper.go

@@ -88,7 +88,8 @@ func validatePvtdata(tx *valinternal.Transaction, pvtdata *ledger.TxPvtData) err
 
 // preprocessProtoBlock parses the proto instance of block into 'Block' structure.
 // The retuned 'Block' structure contains only transactions that are endorser transactions and are not alredy marked as invalid
-func preprocessProtoBlock(txmgr txmgr.TxMgr, block *common.Block, doMVCCValidation bool) (*valinternal.Block, error) {
+func preprocessProtoBlock(txmgr txmgr.TxMgr, validateKVFunc func(key string, value []byte) error,
+	block *common.Block, doMVCCValidation bool) (*valinternal.Block, error) {
 	b := &valinternal.Block{Num: block.Header.Number}
 	// Committer validator has already set validation flags based on well formed tran checks
 	txsFilter := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
@@ -145,6 +146,14 @@ func preprocessProtoBlock(txmgr txmgr.TxMgr, block *common.Block, doMVCCValidati
 			}
 		}
 		if txRWSet != nil {
+			if err := validateWriteset(txRWSet, validateKVFunc); err != nil {
+				logger.Warningf("Channel [%s]: Block [%d] Transaction index [%d] TxId [%s]"+
+					" marked as invalid. Reason code [%s]. Message: [%s]",
+					chdr.GetChannelId(), block.Header.Number, txIndex, chdr.GetTxId(),
+					peer.TxValidationCode_INVALID_WRITESET, err.Error())
+				txsFilter.SetFlag(txIndex, peer.TxValidationCode_INVALID_WRITESET)
+				continue
+			}
 			b.Txs = append(b.Txs, &valinternal.Transaction{IndexInBlock: txIndex, ID: chdr.TxId, RWSet: txRWSet})
 		}
 	}
@@ -175,6 +184,24 @@ func processNonEndorserTx(txEnv *common.Envelope, txid string, txType common.Hea
 	return simRes.PubSimulationResults, nil
 }
 
+func validateWriteset(txRWSet *rwsetutil.TxRwSet, validateKVFunc func(key string, value []byte) error) error {
+	for _, nsRwSet := range txRWSet.NsRwSets {
+		if nsRwSet == nil {
+			continue
+		}
+		pubWriteset := nsRwSet.KvRwSet
+		if pubWriteset == nil {
+			continue
+		}
+		for _, kvwrite := range pubWriteset.Writes {
+			if err := validateKVFunc(kvwrite.Key, kvwrite.Value); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // postprocessProtoBlock updates the proto block's validation flags (in metadata) by the results of validation process
 func postprocessProtoBlock(block *common.Block, validatedBlock *valinternal.Block) {
 	txsFilter := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])

core/ledger/kvledger/txmgmt/validator/valimpl/helper_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/validator/valinternal"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
+	lgrutil "github.com/hyperledger/fabric/core/ledger/util"
 	lutils "github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/peer"
@@ -117,8 +118,10 @@ func TestValidateAndPreparePvtBatch(t *testing.T) {
 	tx3TxRWSet, err := rwsetutil.TxRwSetFromProtoMsg(tx3SimulationResults.PubSimulationResults)
 	assert.NoError(t, err)
 	expectedPerProcessedBlock.Txs = append(expectedPerProcessedBlock.Txs, &valinternal.Transaction{IndexInBlock: 2, ID: "tx3", RWSet: tx3TxRWSet})
-
-	actualPreProcessedBlock, err := preprocessProtoBlock(nil, block, false)
+	alwaysValidKVFunc := func(key string, value []byte) error {
+		return nil
+	}
+	actualPreProcessedBlock, err := preprocessProtoBlock(nil, alwaysValidKVFunc, block, false)
 	assert.NoError(t, err)
 	assert.Equal(t, expectedPerProcessedBlock, actualPreProcessedBlock)
 
@@ -149,25 +152,27 @@ func TestValidateAndPreparePvtBatch(t *testing.T) {
 }
 
 func TestPreprocessProtoBlock(t *testing.T) {
-
+	allwaysValidKVfunc := func(key string, value []byte) error {
+		return nil
+	}
 	// good block
 	//_, gb := testutil.NewBlockGenerator(t, "testLedger", false)
 	gb := testutil.ConstructTestBlock(t, 10, 1, 1)
-	_, err := preprocessProtoBlock(nil, gb, false)
+	_, err := preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.NoError(t, err)
 	// bad envelope
 	gb = testutil.ConstructTestBlock(t, 11, 1, 1)
 	gb.Data = &common.BlockData{Data: [][]byte{{123}}}
 	gb.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER] =
 		lutils.NewTxValidationFlagsSetValue(len(gb.Data.Data), peer.TxValidationCode_VALID)
-	_, err = preprocessProtoBlock(nil, gb, false)
+	_, err = preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.Error(t, err)
 	t.Log(err)
 	// bad payload
 	gb = testutil.ConstructTestBlock(t, 12, 1, 1)
 	envBytes, _ := putils.GetBytesEnvelope(&common.Envelope{Payload: []byte{123}})
 	gb.Data = &common.BlockData{Data: [][]byte{envBytes}}
-	_, err = preprocessProtoBlock(nil, gb, false)
+	_, err = preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.Error(t, err)
 	t.Log(err)
 	// bad channel header
@@ -177,7 +182,7 @@ func TestPreprocessProtoBlock(t *testing.T) {
 	})
 	envBytes, _ = putils.GetBytesEnvelope(&common.Envelope{Payload: payloadBytes})
 	gb.Data = &common.BlockData{Data: [][]byte{envBytes}}
-	_, err = preprocessProtoBlock(nil, gb, false)
+	_, err = preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.Error(t, err)
 	t.Log(err)
 
@@ -191,7 +196,7 @@ func TestPreprocessProtoBlock(t *testing.T) {
 	flags := lutils.NewTxValidationFlags(len(gb.Data.Data))
 	flags.SetFlag(0, peer.TxValidationCode_BAD_CHANNEL_HEADER)
 	gb.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER] = flags
-	_, err = preprocessProtoBlock(nil, gb, false)
+	_, err = preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.NoError(t, err) // invalid filter should take precendence
 
 	// new block
@@ -205,7 +210,7 @@ func TestPreprocessProtoBlock(t *testing.T) {
 	// set logging backend for test
 	backend := logging.NewMemoryBackend(1)
 	logging.SetBackend(backend)
-	_, err = preprocessProtoBlock(nil, gb, false)
+	_, err = preprocessProtoBlock(nil, allwaysValidKVfunc, gb, false)
 	assert.NoError(t, err)
 	expected := fmt.Sprintf("Channel [%s]: Block [%d] Transaction index [%d] TxId [%s]"+
 		" marked as invalid by committer. Reason code [%s]",
@@ -217,6 +222,42 @@ func TestPreprocessProtoBlock(t *testing.T) {
 
 }
 
+func TestPreprocessProtoBlockInvalidWriteset(t *testing.T) {
+	kvValidationFunc := func(key string, value []byte) error {
+		if value[0] == '_' {
+			return fmt.Errorf("value [%s] found to be invalid by 'kvValidationFunc for testing'", value)
+		}
+		return nil
+	}
+
+	rwSetBuilder := rwsetutil.NewRWSetBuilder()
+	rwSetBuilder.AddToWriteSet("ns", "key", []byte("_invalidValue")) // bad value
+	simulation1, err := rwSetBuilder.GetTxSimulationResults()
+	assert.NoError(t, err)
+	simulation1Bytes, err := simulation1.GetPubSimulationBytes()
+	assert.NoError(t, err)
+
+	rwSetBuilder = rwsetutil.NewRWSetBuilder()
+	rwSetBuilder.AddToWriteSet("ns", "key", []byte("validValue")) // good value
+	simulation2, err := rwSetBuilder.GetTxSimulationResults()
+	assert.NoError(t, err)
+	simulation2Bytes, err := simulation2.GetPubSimulationBytes()
+	assert.NoError(t, err)
+
+	block := testutil.ConstructBlock(t, 1, testutil.ConstructRandomBytes(t, 32),
+		[][]byte{simulation1Bytes, simulation2Bytes}, false) // block with two txs
+	txfilter := lgrutil.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+	assert.True(t, txfilter.IsValid(0))
+	assert.True(t, txfilter.IsValid(1)) // both txs are valid initially at the time of block cutting
+
+	internalBlock, err := preprocessProtoBlock(nil, kvValidationFunc, block, false)
+	assert.NoError(t, err)
+	assert.False(t, txfilter.IsValid(0)) // tx at index 0 should be marked as invalid
+	assert.True(t, txfilter.IsValid(1))  // tx at index 1 should be marked as valid
+	assert.Len(t, internalBlock.Txs, 1)
+	assert.Equal(t, internalBlock.Txs[0].IndexInBlock, 1)
+}
+
 // from go-logging memory_test.go
 func memoryRecordN(b *logging.MemoryBackend, n int) *logging.Record {
 	node := b.Head()

core/ledger/ledgerstorage/store.go

@@ -20,16 +20,18 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
-
+	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage/fsblkstorage"
 	"github.com/hyperledger/fabric/core/ledger"
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
+	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
 	"github.com/hyperledger/fabric/core/ledger/pvtdatastorage"
 	"github.com/hyperledger/fabric/protos/common"
 )
 
+var logger = flogging.MustGetLogger("ledgerstorage")
+
 // Provider encapusaltes two providers 1) block store provider and 2) and pvt data store provider
 type Provider struct {
 	blkStoreProvider     blkstorage.BlockStoreProvider
@@ -95,20 +97,41 @@ func (s *Store) Init(btlPolicy pvtdatapolicy.BTLPolicy) {
 
 // CommitWithPvtData commits the block and the corresponding pvt data in an atomic operation
 func (s *Store) CommitWithPvtData(blockAndPvtdata *ledger.BlockAndPvtData) error {
+	blockNum := blockAndPvtdata.Block.Header.Number
 	s.rwlock.Lock()
 	defer s.rwlock.Unlock()
-	var pvtdata []*ledger.TxPvtData
-	for _, v := range blockAndPvtdata.BlockPvtData {
-		pvtdata = append(pvtdata, v)
-	}
-	if err := s.pvtdataStore.Prepare(blockAndPvtdata.Block.Header.Number, pvtdata); err != nil {
+
+	pvtBlkStoreHt, err := s.pvtdataStore.LastCommittedBlockHeight()
+	if err != nil {
 		return err
 	}
+
+	writtenToPvtStore := false
+	if pvtBlkStoreHt < blockNum+1 { // The pvt data store sanity check does not allow rewriting the pvt data.
+		// when re-processing blocks (rejoin the channel or re-fetching last few block),
+		// skip the pvt data commit to the pvtdata blockstore
+		logger.Debugf("Writing block [%d] to pvt block store", blockNum)
+		var pvtdata []*ledger.TxPvtData
+		for _, v := range blockAndPvtdata.BlockPvtData {
+			pvtdata = append(pvtdata, v)
+		}
+		if err := s.pvtdataStore.Prepare(blockAndPvtdata.Block.Header.Number, pvtdata); err != nil {
+			return err
+		}
+		writtenToPvtStore = true
+	} else {
+		logger.Debugf("Skipping writing block [%d] to pvt block store as the store height is [%d]", blockNum, pvtBlkStoreHt)
+	}
+
 	if err := s.AddBlock(blockAndPvtdata.Block); err != nil {
 		s.pvtdataStore.Rollback()
 		return err
 	}
-	return s.pvtdataStore.Commit()
+
+	if writtenToPvtStore {
+		return s.pvtdataStore.Commit()
+	}
+	return nil
 }
 
 // GetPvtDataAndBlockByNum returns the block and the corresponding pvt data.

core/ledger/ledgerstorage/store_test.go

@@ -241,10 +241,9 @@ func TestAddAfterPvtdataStoreError(t *testing.T) {
 	for _, d := range sampleData[0:9] {
 		assert.NoError(t, store.CommitWithPvtData(d))
 	}
-	// try to write the last block again. The function should pass on the error raised by the private store
-	err = store.CommitWithPvtData(sampleData[8])
-	_, ok := err.(*pvtdatastorage.ErrIllegalArgs)
-	assert.True(t, ok)
+	// try to write the last block again. The function should skip adding block to the private store
+	// as the pvt store but the block storage should return error
+	assert.Error(t, store.CommitWithPvtData(sampleData[8]))
 
 	// At the end, the pvt store status should not have changed
 	pvtStoreCommitHt, err := store.pvtdataStore.LastCommittedBlockHeight()