[FAB-6287] filter out private data in state transfer

This change set makes the coordinator filter out private data the requesting
peer receives along with the block.

It also removes GetBlocks() from all the APIs all the way to the committer
Since the method that is used now instead is GetPvtDataAndBlockByNum

Change-Id: Ie2d33abdd10f5e715db775bf3e599e8419ddddfe
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

core/ledger/ledger_interface.go

@@ -174,7 +174,7 @@ type TxPvtData struct {
 	WriteSet   *rwset.TxPvtReadWriteSet
 }
 
-// BlockAndPvtData encapsultes the block and a map that contains the tuples <seqInBlock, *TxPvtData>
+// BlockAndPvtData encapsulates the block and a map that contains the tuples <seqInBlock, *TxPvtData>
 // The map is expected to contain the entries only for the transactions that has associated pvt data
 type BlockAndPvtData struct {
 	Block        *common.Block

gossip/privdata/coordinator.go

@@ -57,10 +57,7 @@ type Coordinator interface {
 	// the order of private data in slice of PvtDataCollections doesn't implies the order of
 	// transactions in the block related to these private data, to get the correct placement
 	// need to read TxPvtData.SeqInBlock field
-	GetPvtDataAndBlockByNum(seqNum uint64) (*common.Block, util.PvtDataCollections, error)
-
-	// GetBlockByNum returns block and related to the block private data
-	GetBlockByNum(seqNum uint64) (*common.Block, error)
+	GetPvtDataAndBlockByNum(seqNum uint64, peerAuth common.SignedData) (*common.Block, util.PvtDataCollections, error)
 
 	// Get recent block sequence number
 	LedgerHeight() (uint64, error)
@@ -404,35 +401,27 @@ func (k *rwSetKey) toTxPvtReadWriteSet(rws []byte) *rwset.TxPvtReadWriteSet {
 	}
 }
 
-func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets map[rwSetKey][]byte) (rwSetKeysByTxIDs, error) {
-	if block.Metadata == nil || len(block.Metadata.Metadata) <= int(common.BlockMetadataIndex_TRANSACTIONS_FILTER) {
-		return nil, errors.New("Block.Metadata is nil or Block.Metadata lacks a Tx filter bitmap")
-	}
-	txsFilter := txValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
-	if len(txsFilter) != len(block.Data.Data) {
-		return nil, errors.Errorf("Block data size(%d) is different from Tx filter size(%d)", len(block.Data.Data), len(txsFilter))
-	}
+type blockData [][]byte
 
-	privateRWsetsInBlock := make(map[rwSetKey]struct{})
-	missing := make(rwSetKeysByTxIDs)
-	for seqInBlock, envBytes := range block.Data.Data {
+func (data blockData) forEach(txsFilter txValidationFlags, consumer func(seqInBlock uint64, chdr *common.ChannelHeader, txRWSet *rwsetutil.TxRwSet)) error {
+	for seqInBlock, envBytes := range data {
 		if txsFilter[seqInBlock] != uint8(peer.TxValidationCode_VALID) {
 			logger.Debug("Skipping Tx", seqInBlock, "because it's invalid. Status is", txsFilter[seqInBlock])
 			continue
 		}
 		env, err := utils.GetEnvelopeFromBlock(envBytes)
 		if err != nil {
-			return nil, err
+			return err
 		}
 
 		payload, err := utils.GetPayload(env)
 		if err != nil {
-			return nil, err
+			return err
 		}
 
 		chdr, err := utils.UnmarshalChannelHeader(payload.Header.ChannelHeader)
 		if err != nil {
-			return nil, err
+			return err
 		}
 
 		if chdr.Type != int32(common.HeaderType_ENDORSER_TRANSACTION) {
@@ -450,15 +439,32 @@ func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets ma
 			logger.Warning("Failed obtaining TxRwSet from ChaincodeAction's results", err)
 			continue
 		}
+		consumer(uint64(seqInBlock), chdr, txRWSet)
+	}
+	return nil
+}
 
+func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets map[rwSetKey][]byte) (rwSetKeysByTxIDs, error) {
+	if block.Metadata == nil || len(block.Metadata.Metadata) <= int(common.BlockMetadataIndex_TRANSACTIONS_FILTER) {
+		return nil, errors.New("Block.Metadata is nil or Block.Metadata lacks a Tx filter bitmap")
+	}
+	txsFilter := txValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+	if len(txsFilter) != len(block.Data.Data) {
+		return nil, errors.Errorf("Block data size(%d) is different from Tx filter size(%d)", len(block.Data.Data), len(txsFilter))
+	}
+
+	privateRWsetsInBlock := make(map[rwSetKey]struct{})
+	missing := make(rwSetKeysByTxIDs)
+	data := blockData(block.Data.Data)
+	err := data.forEach(txsFilter, func(seqInBlock uint64, chdr *common.ChannelHeader, txRWSet *rwsetutil.TxRwSet) {
 		for _, ns := range txRWSet.NsRwSets {
 			for _, hashed := range ns.CollHashedRwSets {
 				if !c.isEligible(chdr, ns.NameSpace, hashed.CollectionName) {
 					continue
 				}
 				key := rwSetKey{
 					txID:       chdr.TxId,
-					seqInBlock: uint64(seqInBlock),
+					seqInBlock: seqInBlock,
 					hash:       hex.EncodeToString(hashed.PvtRwSetHash),
 					namespace:  ns.NameSpace,
 					collection: hashed.CollectionName,
@@ -467,14 +473,16 @@ func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets ma
 				if _, exists := ownedRWsets[key]; !exists {
 					txAndSeq := txAndSeqInBlock{
 						txID:       chdr.TxId,
-						seqInBlock: uint64(seqInBlock),
+						seqInBlock: seqInBlock,
 					}
 					missing[txAndSeq] = append(missing[txAndSeq], key)
 				}
 			} // for all hashed RW sets
 		} // for all RW sets
-	} // for all transactions in block
-
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
 	// In the end, iterate over the ownedRWsets, and if the key doesn't exist in
 	// the privateRWsetsInBlock - delete it from the ownedRWsets
 	for k := range ownedRWsets {
@@ -512,30 +520,93 @@ func (c *coordinator) isEligible(chdr *common.ChannelHeader, namespace string, c
 	return eligible
 }
 
+type seqAndDataModel struct {
+	seq       uint64
+	dataModel rwset.TxReadWriteSet_DataModel
+}
+
+// map from seqAndDataModel to:
+//     maap from namespace to []*rwset.CollectionPvtReadWriteSet
+type aggregatedCollections map[seqAndDataModel]map[string][]*rwset.CollectionPvtReadWriteSet
+
+func (ac aggregatedCollections) addCollection(seqInBlock uint64, dm rwset.TxReadWriteSet_DataModel, namespace string, col *rwset.CollectionPvtReadWriteSet) {
+	seq := seqAndDataModel{
+		dataModel: dm,
+		seq:       seqInBlock,
+	}
+	if _, exists := ac[seq]; !exists {
+		ac[seq] = make(map[string][]*rwset.CollectionPvtReadWriteSet)
+	}
+	ac[seq][namespace] = append(ac[seq][namespace], col)
+}
+
+func (ac aggregatedCollections) asPrivateData() []*ledger.TxPvtData {
+	var data []*ledger.TxPvtData
+	for seq, ns := range ac {
+		txPrivateData := &ledger.TxPvtData{
+			SeqInBlock: seq.seq,
+			WriteSet: &rwset.TxPvtReadWriteSet{
+				DataModel: seq.dataModel,
+			},
+		}
+		for namespaceName, cols := range ns {
+			txPrivateData.WriteSet.NsPvtRwset = append(txPrivateData.WriteSet.NsPvtRwset, &rwset.NsPvtReadWriteSet{
+				Namespace:          namespaceName,
+				CollectionPvtRwset: cols,
+			})
+		}
+		data = append(data, txPrivateData)
+	}
+	return data
+}
+
 // GetPvtDataAndBlockByNum get block by number and returns also all related private data
 // the order of private data in slice of PvtDataCollections doesn't implies the order of
 // transactions in the block related to these private data, to get the correct placement
 // need to read TxPvtData.SeqInBlock field
-func (c *coordinator) GetPvtDataAndBlockByNum(seqNum uint64) (*common.Block, util.PvtDataCollections, error) {
+func (c *coordinator) GetPvtDataAndBlockByNum(seqNum uint64, peerAuthInfo common.SignedData) (*common.Block, util.PvtDataCollections, error) {
 	blockAndPvtData, err := c.Committer.GetPvtDataAndBlockByNum(seqNum)
 	if err != nil {
 		return nil, nil, fmt.Errorf("cannot retrieve block number %d, due to %s", seqNum, err)
 	}
 
-	var blockPvtData util.PvtDataCollections
+	seqs2Namespaces := aggregatedCollections(make(map[seqAndDataModel]map[string][]*rwset.CollectionPvtReadWriteSet))
+	data := blockData(blockAndPvtData.Block.Data.Data)
+	err = data.forEach(make(txValidationFlags, len(data)), func(seqInBlock uint64, chdr *common.ChannelHeader, txRWSet *rwsetutil.TxRwSet) {
+		item, exists := blockAndPvtData.BlockPvtData[seqInBlock]
+		if !exists {
+			return
+		}
 
-	for _, item := range blockAndPvtData.BlockPvtData {
-		blockPvtData = append(blockPvtData, item)
+		for _, ns := range item.WriteSet.NsPvtRwset {
+			for _, col := range ns.CollectionPvtRwset {
+				cc := rwset.CollectionCriteria{
+					Channel:    chdr.ChannelId,
+					TxId:       chdr.TxId,
+					Namespace:  ns.Namespace,
+					Collection: col.CollectionName,
+				}
+				sp := c.PolicyStore.CollectionPolicy(cc)
+				if sp == nil {
+					logger.Warning("Failed obtaining policy for", cc)
+					continue
+				}
+				isAuthorized := c.PolicyParser.Parse(sp)
+				if isAuthorized == nil {
+					logger.Warning("Failed obtaining filter for", cc)
+					continue
+				}
+				if !isAuthorized(peerAuthInfo) {
+					logger.Debug("Skipping", cc, "because peer isn't authorized")
+					continue
+				}
+				seqs2Namespaces.addCollection(seqInBlock, item.WriteSet.DataModel, ns.Namespace, col)
+			}
+		}
+	})
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
 	}
 
-	return blockAndPvtData.Block, blockPvtData, nil
-}
-
-// GetBlockByNum returns block by sequence number
-func (c *coordinator) GetBlockByNum(seqNum uint64) (*common.Block, error) {
-	blocks := c.GetBlocks([]uint64{seqNum})
-	if len(blocks) == 0 {
-		return nil, fmt.Errorf("cannot retrieve block number %d", seqNum)
-	}
-	return blocks[0], nil
+	return blockAndPvtData.Block, seqs2Namespaces.asPrivateData(), nil
 }

gossip/privdata/coordinator_test.go

@@ -222,19 +222,19 @@ func (f *fetcherMock) fetch(req *proto.RemotePvtDataRequest) ([]*proto.PvtDataEl
 	return nil, args.Get(1).(error)
 }
 
-func createPolicyStore(selfSignedData common.SignedData) *policyStore {
+func createPolicyStore(expectedSignedData common.SignedData) *policyStore {
 	return &policyStore{
-		selfSignedData: selfSignedData,
-		policies:       make(map[serializedPolicy]rwset.CollectionCriteria),
-		store:          make(map[rwset.CollectionCriteria]serializedPolicy),
+		expectedSignedData: expectedSignedData,
+		policies:           make(map[serializedPolicy]rwset.CollectionCriteria),
+		store:              make(map[rwset.CollectionCriteria]serializedPolicy),
 	}
 }
 
 type policyStore struct {
-	selfSignedData common.SignedData
-	acceptsAll     bool
-	store          map[rwset.CollectionCriteria]serializedPolicy
-	policies       map[serializedPolicy]rwset.CollectionCriteria
+	expectedSignedData common.SignedData
+	acceptsAll         bool
+	store              map[rwset.CollectionCriteria]serializedPolicy
+	policies           map[serializedPolicy]rwset.CollectionCriteria
 }
 
 func (ps *policyStore) thatAcceptsAll() *policyStore {
@@ -282,7 +282,7 @@ func (*policyParser) Parse(serializedPol privdata.SerializedPolicy) privdata.Fil
 	sp := serializedPol.(*serializedPolicy)
 	return func(sd common.SignedData) bool {
 		that, _ := asn1.Marshal(sd)
-		this, _ := asn1.Marshal(sp.ps.selfSignedData)
+		this, _ := asn1.Marshal(sp.ps.expectedSignedData)
 		if hex.EncodeToString(that) != hex.EncodeToString(this) {
 			panic("Self signed data passed isn't equal to expected")
 		}
@@ -810,56 +810,63 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 }
 
 func TestCoordinatorGetBlocks(t *testing.T) {
+	sd := common.SignedData{
+		Identity:  []byte{0, 1, 2},
+		Signature: []byte{3, 4, 5},
+		Data:      []byte{6, 7, 8},
+	}
+	ps := createPolicyStore(sd).thatAcceptsAll()
+	pp := &policyParser{}
 	committer := &committerMock{}
 	store := &mockTransientStore{t: t}
 	fetcher := &fetcherMock{t: t}
 	coordinator := NewCoordinator(Support{
+		PolicyStore:    ps,
+		PolicyParser:   pp,
 		Committer:      committer,
 		Fetcher:        fetcher,
 		TransientStore: store,
 		Validator:      &validatorMock{},
-	}, common.SignedData{})
+	}, sd)
 
-	// Bad path: block is not returned
-	committer.On("GetBlocks", mock.Anything).Return([]*common.Block{})
-	block, err := coordinator.GetBlockByNum(1)
-	assert.Contains(t, err.Error(), "cannot retrieve block number 1")
-	assert.Error(t, err)
-	assert.Nil(t, block)
+	hash := util2.ComputeSHA256([]byte("rws-pre-image"))
+	bf := &blockFactory{
+		channelID: "test",
+	}
+	block := bf.AddTxn("tx1", "ns1", hash, "c1", "c2").AddTxn("tx2", "ns2", hash, "c1").create()
 
-	// Green path: Block is returned
-	committer.Mock = mock.Mock{}
-	committer.On("GetBlocks", mock.Anything).Return([]*common.Block{
-		{
-			Header: &common.BlockHeader{
-				Number: 1,
-			},
-		},
+	// Green path - block and private data is returned, but the requester isn't eligible for all the private data,
+	// but only to a subset of it.
+	ps = createPolicyStore(sd).thatAccepts(rwset.CollectionCriteria{
+		Namespace:  "ns1",
+		Collection: "c2",
+		TxId:       "tx1",
+		Channel:    "test",
 	})
-	block, err = coordinator.GetBlockByNum(1)
-	assert.NoError(t, err)
-	assert.Equal(t, uint64(1), block.Header.Number)
-
+	committer.Mock = mock.Mock{}
 	committer.On("GetPvtDataAndBlockByNum", mock.Anything).Return(&ledger.BlockAndPvtData{
-		Block: block,
-		BlockPvtData: map[uint64]*ledger.TxPvtData{
-			0: {
-				SeqInBlock: 0,
-			},
-		},
+		Block:        block,
+		BlockPvtData: expectedCommittedPrivateData1,
 	}, nil)
-
-	// Green path - block and private data is returned
-	block2, privData, err := coordinator.GetPvtDataAndBlockByNum(1)
+	coordinator = NewCoordinator(Support{
+		PolicyStore:    ps,
+		PolicyParser:   pp,
+		Committer:      committer,
+		Fetcher:        fetcher,
+		TransientStore: store,
+		Validator:      &validatorMock{},
+	}, sd)
+	expectedPrivData := (&pvtDataFactory{}).addRWSet().addNSRWSet("ns1", "c2").create()
+	block2, returnedPrivateData, err := coordinator.GetPvtDataAndBlockByNum(1, sd)
 	assert.NoError(t, err)
 	assert.Equal(t, block, block2)
-	assert.Equal(t, util.PvtDataCollections{{SeqInBlock: 0}}, privData)
+	assert.Equal(t, expectedPrivData, []*ledger.TxPvtData(returnedPrivateData))
 
 	// Bad path - error occurs when trying to retrieve the block and private data
 	committer.Mock = mock.Mock{}
 	committer.On("GetPvtDataAndBlockByNum", mock.Anything).Return(nil, errors.New("uh oh"))
-	block2, privData, err = coordinator.GetPvtDataAndBlockByNum(1)
+	block2, returnedPrivateData, err = coordinator.GetPvtDataAndBlockByNum(1, sd)
 	assert.Nil(t, block2)
-	assert.Empty(t, privData)
+	assert.Empty(t, returnedPrivateData)
 	assert.Error(t, err)
 }

gossip/service/gossip_service.go

@@ -50,8 +50,6 @@ type GossipService interface {
 	NewConfigEventer() ConfigProcessor
 	// InitializeChannel allocates the state provider and should be invoked once per channel per execution
 	InitializeChannel(chainID string, endpoints []string, support Support)
-	// GetBlock returns block for given chain
-	GetBlock(chainID string, index uint64) *common.Block
 	// AddPayload appends message payload to for given chain
 	AddPayload(chainID string, payload *gproto.Payload) error
 }
@@ -308,13 +306,6 @@ func (g *gossipServiceImpl) configUpdated(config Config) {
 	g.JoinChan(jcm, gossipCommon.ChainID(config.ChainID()))
 }
 
-// GetBlock returns block for given chain
-func (g *gossipServiceImpl) GetBlock(chainID string, index uint64) *common.Block {
-	g.lock.RLock()
-	defer g.lock.RUnlock()
-	return g.chains[chainID].GetBlock(index)
-}
-
 // AddPayload appends message payload to for given chain
 func (g *gossipServiceImpl) AddPayload(chainID string, payload *gproto.Payload) error {
 	g.lock.RLock()