[FAB-3556] Throw error for invalid keys at simulation

This CR makes changes in the simulation such that the simulator
raises an error if an invalid key is passed in. The validity of the
key is decided based on the configured db. For instance, couchdb allows
only valid utf-8 strings to be the key.

In the current code, this check is not made during simulation. This causes
to defer the issue at the commit time in which situation, peer has no option
other than raising a panic becasue, it simply cannot commit the block.

Change-Id: Id1439c36e3a142f8ca47574d391ee10b366acd69
Signed-off-by: manish <manish.sethi@gmail.com>

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

@@ -24,6 +24,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"unicode/utf8"
 
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
@@ -113,6 +114,14 @@ func (vdb *VersionedDB) Close() {
 	// no need to close db since a shared couch instance is used
 }
 
+// ValidateKey implements method in VersionedDB interface
+func (vdb *VersionedDB) ValidateKey(key string) error {
+	if !utf8.ValidString(key) {
+		return fmt.Errorf("Key should be a valid utf8 string: [%x]", key)
+	}
+	return nil
+}
+
 // GetState implements method in VersionedDB interface
 func (vdb *VersionedDB) GetState(namespace string, key string) (*statedb.VersionedValue, error) {
 	logger.Debugf("GetState(). ns=%s, key=%s", namespace, key)

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

@@ -51,6 +51,9 @@ type VersionedDB interface {
 	// GetLatestSavePoint returns the height of the highest transaction upto which
 	// the state db is consistent
 	GetLatestSavePoint() (*version.Height, error)
+	// ValidateKey tests whether the key is supported by the db implementation.
+	// For instance, leveldb supports any bytes for the key while the couchdb supports only valid utf-8 string
+	ValidateKey(key string) error
 	// Open opens the db
 	Open() error
 	// Close closes the db

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

@@ -79,6 +79,11 @@ func (vdb *versionedDB) Close() {
 	// do nothing because shared db is used
 }
 
+// ValidateKey implements method in VersionedDB interface
+func (vdb *versionedDB) ValidateKey(key string) error {
+	return nil
+}
+
 // GetState implements method in VersionedDB interface
 func (vdb *versionedDB) GetState(namespace string, key string) (*statedb.VersionedValue, error) {
 	logger.Debugf("GetState(). ns=%s, key=%s", namespace, key)

core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_tx_simulator.go

@@ -45,6 +45,9 @@ func (s *lockBasedTxSimulator) GetState(ns string, key string) ([]byte, error) {
 // SetState implements method in interface `ledger.TxSimulator`
 func (s *lockBasedTxSimulator) SetState(ns string, key string, value []byte) error {
 	s.helper.checkDone()
+	if err := s.helper.txmgr.db.ValidateKey(key); err != nil {
+		return err
+	}
 	s.rwsetBuilder.AddToWriteSet(ns, key, value)
 	return nil
 }

core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/txmgr_test.go

@@ -587,26 +587,37 @@ func testExecuteQuery(t *testing.T, env testEnv) {
 
 	itr, err := queryExecuter.ExecuteQuery("ns1", queryString)
 	testutil.AssertNoError(t, err, "Error upon ExecuteQuery()")
-
 	counter := 0
 	for {
 		queryRecord, _ := itr.Next()
 		if queryRecord == nil {
 			break
 		}
-
 		//Unmarshal the document to Asset structure
 		assetResp := &Asset{}
 		json.Unmarshal(queryRecord.(*queryresult.KV).Value, &assetResp)
-
 		//Verify the owner retrieved matches
 		testutil.AssertEquals(t, assetResp.Owner, "bob")
-
 		counter++
-
 	}
-
 	//Ensure the query returns 3 documents
 	testutil.AssertEquals(t, counter, 3)
+}
 
+func TestValidateKey(t *testing.T) {
+	nonUTF8Key := string([]byte{0xff, 0xff})
+	dummyValue := []byte("dummyValue")
+	for _, testEnv := range testEnvs {
+		testLedgerID := "TestValidateKey"
+		testEnv.init(t, testLedgerID)
+		txSimulator, _ := testEnv.getTxMgr().NewTxSimulator()
+		err := txSimulator.SetState("ns1", nonUTF8Key, dummyValue)
+		if testEnv.getName() == levelDBtestEnvName {
+			testutil.AssertNoError(t, err, "")
+		}
+		if testEnv.getName() == couchDBtestEnvName {
+			testutil.AssertError(t, err, "")
+		}
+		testEnv.cleanup()
+	}
 }