[FAB-15450] validate historydb key if it contains nils

There was a logic to skip keys with nil bytes from the historydb results.
However, because some valid blockNumTranNumBytes may also contain
nil bytes, these history keys are filtered out.

This CR updates the logic to handle decoding error and process all keys
returned by range query no matter if blockNumTranNumBytes contains
a nil byte. If there is a decoding error, block:tran cannot be found,
or key is not present in write set, it means that the key is a false key
and will be skipped.

Change-Id: I296479af55b7c750ef8f797faa807ee0f5489b0a
Signed-off-by: Wenjian Qiao <wenjianq@gmail.com>

common/ledger/util/util.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/pkg/errors"
 )
 
 // EncodeOrderPreservingVarUint64 returns a byte-representation for a uint64 number such that
@@ -51,12 +52,23 @@ func EncodeOrderPreservingVarUint64(number uint64) []byte {
 }
 
 // DecodeOrderPreservingVarUint64 decodes the number from the bytes obtained from method 'EncodeOrderPreservingVarUint64'.
-// Also, returns the number of bytes that are consumed in the process
-func DecodeOrderPreservingVarUint64(bytes []byte) (uint64, int) {
-	s, _ := proto.DecodeVarint(bytes)
-	size := int(s)
-	decodedBytes := make([]byte, 8)
-	copy(decodedBytes[8-size:], bytes[1:size+1])
-	numBytesConsumed := size + 1
-	return binary.BigEndian.Uint64(decodedBytes), numBytesConsumed
+// It returns the decoded number, the number of bytes that are consumed in the process, and an error if the input bytes are invalid.
+func DecodeOrderPreservingVarUint64(bytes []byte) (uint64, int, error) {
+	s, numBytes := proto.DecodeVarint(bytes)
+
+	switch {
+	case numBytes != 1:
+		return 0, 0, errors.Errorf("number of consumed bytes from DecodeVarint is invalid, expected 1, but got %d", numBytes)
+	case s > 8:
+		return 0, 0, errors.Errorf("decoded size from DecodeVarint is invalid, expected <=8, but got %d", s)
+	case int(s) > len(bytes)-1:
+		return 0, 0, errors.Errorf("decoded size (%d) from DecodeVarint is more than available bytes (%d)", s, len(bytes)-1)
+	default:
+		// no error
+		size := int(s)
+		decodedBytes := make([]byte, 8)
+		copy(decodedBytes[8-size:], bytes[1:size+1])
+		numBytesConsumed := size + 1
+		return binary.BigEndian.Uint64(decodedBytes), numBytesConsumed, nil
+	}
 }

common/ledger/util/util_test.go

@@ -18,7 +18,11 @@ package util
 
 import (
 	"bytes"
+	"fmt"
 	"testing"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestBasicEncodingDecoding(t *testing.T) {
@@ -29,7 +33,8 @@ func TestBasicEncodingDecoding(t *testing.T) {
 			t.Fatalf("A smaller integer should result into smaller bytes. Encoded bytes for [%d] is [%x] and for [%d] is [%x]",
 				i, i+1, value, nextValue)
 		}
-		decodedValue, _ := DecodeOrderPreservingVarUint64(value)
+		decodedValue, _, err := DecodeOrderPreservingVarUint64(value)
+		assert.NoError(t, err, "Error via calling DecodeOrderPreservingVarUint64")
 		if decodedValue != uint64(i) {
 			t.Fatalf("Value not same after decoding. Original value = [%d], decode value = [%d]", i, decodedValue)
 		}
@@ -44,11 +49,35 @@ func TestDecodingAppendedValues(t *testing.T) {
 
 	len := 0
 	value := uint64(0)
+	var err error
 	for i := 0; i < 1000; i++ {
 		appendedValues = appendedValues[len:]
-		value, len = DecodeOrderPreservingVarUint64(appendedValues)
+		value, len, err = DecodeOrderPreservingVarUint64(appendedValues)
+		assert.NoError(t, err, "Error via calling DecodeOrderPreservingVarUint64")
 		if value != uint64(i) {
 			t.Fatalf("expected value = [%d], decode value = [%d]", i, value)
 		}
 	}
 }
+
+func TestDecodingBadInputBytes(t *testing.T) {
+	// error case when num consumed bytes > 1
+	sizeBytes := proto.EncodeVarint(uint64(1000))
+	_, _, err := DecodeOrderPreservingVarUint64(sizeBytes)
+	assert.Equal(t, fmt.Sprintf("number of consumed bytes from DecodeVarint is invalid, expected 1, but got %d", len(sizeBytes)), err.Error())
+
+	// error case when decoding invalid bytes - trim off last byte
+	invalidSizeBytes := sizeBytes[0 : len(sizeBytes)-1]
+	_, _, err = DecodeOrderPreservingVarUint64(invalidSizeBytes)
+	assert.Equal(t, "number of consumed bytes from DecodeVarint is invalid, expected 1, but got 0", err.Error())
+
+	// error case when size is more than available bytes
+	inputBytes := proto.EncodeVarint(uint64(8))
+	_, _, err = DecodeOrderPreservingVarUint64(inputBytes)
+	assert.Equal(t, "decoded size (8) from DecodeVarint is more than available bytes (0)", err.Error())
+
+	// error case when size is greater than 8
+	bigSizeBytes := proto.EncodeVarint(uint64(12))
+	_, _, err = DecodeOrderPreservingVarUint64(bigSizeBytes)
+	assert.Equal(t, "decoded size from DecodeVarint is invalid, expected <=8, but got 12", err.Error())
+}

core/ledger/kvledger/history/historydb/historyleveldb/fakes/historyleveldb_logger.go

@@ -0,0 +1,26 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package fakes
+
+// HistoryleveldbLogger is a wrapper to fakes.Logger so that we can overwrite Warnf.
+type HistoryleveldbLogger struct {
+	*HistorydbLogger
+}
+
+// Warnf copies byte array in arg2 and then delegates to fakes.Logger.
+// We have to make the copy because goleveldb iterator
+// reuses the byte array for keys and values for the Next() call.
+func (hl *HistoryleveldbLogger) Warnf(arg1 string, arg2 ...interface{}) {
+	for i := 0; i < len(arg2); i++ {
+		if b, ok := arg2[i].([]byte); ok {
+			bCopy := make([]byte, len(b))
+			copy(bCopy, b)
+			arg2[i] = bCopy
+		}
+	}
+	hl.HistorydbLogger.Warnf(arg1, arg2...)
+}

core/ledger/kvledger/history/historydb/historyleveldb/historyleveldb.go

@@ -20,11 +20,21 @@ import (
 	putils "github.com/hyperledger/fabric/protos/utils"
 )
 
-var logger = flogging.MustGetLogger("historyleveldb")
+var logger historydbLogger = flogging.MustGetLogger("historyleveldb")
 
 var savePointKey = []byte{0x00}
 var emptyValue = []byte{}
 
+//go:generate counterfeiter -o fakes/historydb_logger.go -fake-name HistorydbLogger . historydbLogger
+
+// historydbLogger defines the interface for historyleveldb logging. The purpose is to allow unit tests to use a fake logger.
+type historydbLogger interface {
+	Debugf(template string, args ...interface{})
+	Errorf(template string, args ...interface{})
+	Infof(template string, args ...interface{})
+	Warnf(template string, args ...interface{})
+}
+
 // HistoryDBProvider implements interface HistoryDBProvider
 type HistoryDBProvider struct {
 	dbProvider *leveldbhelper.Provider