Add PurgedKeyAuditLogging config option

Add peer config option to log purged private data keys.
Useful to audit that private data has been purged from a peer,
however some users may want to disable the audit logging to prevent
private data keys from appearing in peer logs, while preserving the
other pvtdatastorage log messages.

Signed-off-by: David Enyeart <enyeart@us.ibm.com>

core/ledger/ledger_interface.go

@@ -118,6 +118,8 @@ type PrivateDataConfig struct {
 	// from other peers. A chance for eligible deprioritized missing data
 	// would be given after every DeprioritizedDataReconcilerInterval
 	DeprioritizedDataReconcilerInterval time.Duration
+	// PurgedKeyAuditLogging specifies whether to log private data keys purged from private data store (INFO level) when explicitly purged via chaincode
+	PurgedKeyAuditLogging bool
 }
 
 // HistoryDBConfig is a structure used to configure the transaction history database.

core/ledger/pvtdatastorage/store.go

@@ -47,12 +47,13 @@ type PrivateDataConfig struct {
 
 // Store manages the permanent storage of private write sets for a ledger
 type Store struct {
-	db              *leveldbhelper.DBHandle
-	ledgerid        string
-	btlPolicy       pvtdatapolicy.BTLPolicy
-	batchesInterval int
-	maxBatchSize    int
-	purgeInterval   uint64
+	db                    *leveldbhelper.DBHandle
+	ledgerid              string
+	btlPolicy             pvtdatapolicy.BTLPolicy
+	batchesInterval       int
+	maxBatchSize          int
+	purgeInterval         uint64
+	purgedKeyAuditLogging bool
 
 	isEmpty            bool
 	lastCommittedBlock uint64
@@ -227,6 +228,7 @@ func (p *Provider) OpenStore(ledgerid string) (*Store, error) {
 		batchesInterval:                     p.pvtData.BatchesInterval,
 		maxBatchSize:                        p.pvtData.MaxBatchSize,
 		purgeInterval:                       uint64(p.pvtData.PurgeInterval),
+		purgedKeyAuditLogging:               p.pvtData.PurgedKeyAuditLogging,
 		deprioritizedDataReconcilerInterval: p.pvtData.DeprioritizedDataReconcilerInterval,
 		accessDeprioMissingDataAfter:        time.Now().Add(p.pvtData.DeprioritizedDataReconcilerInterval),
 		collElgProcSync: &collElgProcSync{
@@ -869,7 +871,7 @@ func (s *Store) deleteDataMarkedForPurge() error {
 	maxBatchSize := 4 * 1024 * 1024 // 4Mb
 	purgeMarkerCounter := 0
 	hashedIndexCounter := 0
-	p := newPurgeUpdatesProcessor(s.db, maxBatchSize)
+	p := newPurgeUpdatesProcessor(s.ledgerid, s.db, s.purgedKeyAuditLogging, maxBatchSize)
 	pStart, pEnd := rangeScanKeysForPurgeMarkers()
 
 	// get the purge markers that need to be processed at this block height
@@ -911,12 +913,18 @@ func (s *Store) deleteDataMarkedForPurge() error {
 		p.addProcessedPurgeMarkerForDeletion(encPurgeMarkerKey)
 		purgeMarkerCounter++
 	}
+
+	// commit the private data purges and index deletions to the private data store
+	err = p.commitPendingChanges()
+	if err != nil {
+		return err
+	}
+
 	if purgeMarkerCounter > 0 {
-		logger.Infow("Processed private data purges from private data storage", "channel", s.ledgerid, "numKeysPurged", purgeMarkerCounter, "numPrivateDataStoreRecordsPurged", hashedIndexCounter)
+		logger.Infow("Purged private data from private data storage", "channel", s.ledgerid, "numKeysPurged", purgeMarkerCounter, "numPrivateDataStoreRecordsPurged", hashedIndexCounter)
 	}
 
-	// commit the private data purges and index deletions to the private data store
-	return p.commitPendingChanges()
+	return nil
 }
 
 func (s *Store) retrieveExpiryEntries(minBlkNum, maxBlkNum uint64) ([]*expiryEntry, error) {
@@ -1159,23 +1167,27 @@ func (c *collElgProcSync) waitForDone() {
 }
 
 type purgeUpdatesProcessor struct {
+	ledgerid     string
 	db           *leveldbhelper.DBHandle
+	batch        *leveldbhelper.UpdateBatch
 	maxBatchSize int
 
 	pvtWrites map[string]*rwsetutil.CollPvtRwSet
-	batch     *leveldbhelper.UpdateBatch
 
-	currentSize int
+	currentSize           int
+	purgedKeyAuditLogging bool
 }
 
 // newPurgeUpdatesProcessor is used for processing the purge markers - i.e., delete the private data versions that are marked for purge from
 // the pvtdata store.
-func newPurgeUpdatesProcessor(db *leveldbhelper.DBHandle, maxBatchSize int) *purgeUpdatesProcessor {
+func newPurgeUpdatesProcessor(ledgerid string, db *leveldbhelper.DBHandle, purgedKeyAuditLogging bool, maxBatchSize int) *purgeUpdatesProcessor {
 	return &purgeUpdatesProcessor{
-		db:           db,
-		maxBatchSize: maxBatchSize,
-		pvtWrites:    map[string]*rwsetutil.CollPvtRwSet{},
-		batch:        db.NewUpdateBatch(),
+		ledgerid:              ledgerid,
+		db:                    db,
+		purgedKeyAuditLogging: purgedKeyAuditLogging,
+		maxBatchSize:          maxBatchSize,
+		pvtWrites:             map[string]*rwsetutil.CollPvtRwSet{},
+		batch:                 db.NewUpdateBatch(),
 	}
 }
 
@@ -1218,6 +1230,16 @@ func (p *purgeUpdatesProcessor) process(hashedIndexKey, hashedIndexVal []byte) e
 			break
 		}
 	}
+
+	// If configured, log the purged key for audit purpose
+	if p.purgedKeyAuditLogging {
+		decodedDataKey, err := decodeDatakey(dataKey)
+		if err != nil {
+			return err
+		}
+		logger.Infow("Purging private data from private data storage", "channel", p.ledgerid, "chaincode", decodedDataKey.nsCollBlk.ns, "collection", decodedDataKey.nsCollBlk.coll, "key", string(hashedIndexVal), "blockNum", decodedDataKey.nsCollBlk.blkNum, "tranNum", decodedDataKey.txNum)
+	}
+
 	p.batch.Delete(hashedIndexKey)
 	if p.currentSize+p.batch.Size() > p.maxBatchSize {
 		if err := p.commitPendingChanges(); err != nil {

internal/peer/node/config.go

@@ -41,6 +41,10 @@ func ledgerConfig() *ledger.Config {
 	if viper.IsSet("ledger.pvtdataStore.deprioritizedDataReconcilerInterval") {
 		deprioritizedDataReconcilerInterval = viper.GetDuration("ledger.pvtdataStore.deprioritizedDataReconcilerInterval")
 	}
+	purgedKeyAuditLogging := true
+	if viper.IsSet("ledger.pvtdataStore.purgedKeyAuditLogging") {
+		purgedKeyAuditLogging = viper.GetBool("ledger.pvtdataStore.purgedKeyAuditLogging")
+	}
 
 	fsPath := coreconfig.GetPath("peer.fileSystemPath")
 	ledgersDataRootDir := filepath.Join(fsPath, "ledgersData")
@@ -59,6 +63,7 @@ func ledgerConfig() *ledger.Config {
 			BatchesInterval:                     collElgProcDbBatchesInterval,
 			PurgeInterval:                       purgeInterval,
 			DeprioritizedDataReconcilerInterval: deprioritizedDataReconcilerInterval,
+			PurgedKeyAuditLogging:               purgedKeyAuditLogging,
 		},
 		HistoryDBConfig: &ledger.HistoryDBConfig{
 			Enabled: viper.GetBool("ledger.history.enableHistoryDatabase"),

internal/peer/node/config_test.go

@@ -39,6 +39,7 @@ func TestLedgerConfig(t *testing.T) {
 					BatchesInterval:                     1000,
 					PurgeInterval:                       100,
 					DeprioritizedDataReconcilerInterval: 60 * time.Minute,
+					PurgedKeyAuditLogging:               true,
 				},
 				HistoryDBConfig: &ledger.HistoryDBConfig{
 					Enabled: false,
@@ -85,6 +86,7 @@ func TestLedgerConfig(t *testing.T) {
 					BatchesInterval:                     1000,
 					PurgeInterval:                       100,
 					DeprioritizedDataReconcilerInterval: 60 * time.Minute,
+					PurgedKeyAuditLogging:               true,
 				},
 				HistoryDBConfig: &ledger.HistoryDBConfig{
 					Enabled: false,
@@ -112,6 +114,7 @@ func TestLedgerConfig(t *testing.T) {
 				"ledger.pvtdataStore.collElgProcMaxDbBatchSize":           50000,
 				"ledger.pvtdataStore.collElgProcDbBatchesInterval":        10000,
 				"ledger.pvtdataStore.purgeInterval":                       1000,
+				"ledger.pvtdataStore.purgedKeyAuditLogging":               false,
 				"ledger.pvtdataStore.deprioritizedDataReconcilerInterval": "180m",
 				"ledger.history.enableHistoryDatabase":                    true,
 				"ledger.snapshots.rootDir":                                "/peerfs/customLocationForsnapshots",
@@ -139,6 +142,7 @@ func TestLedgerConfig(t *testing.T) {
 					BatchesInterval:                     10000,
 					PurgeInterval:                       1000,
 					DeprioritizedDataReconcilerInterval: 180 * time.Minute,
+					PurgedKeyAuditLogging:               false,
 				},
 				HistoryDBConfig: &ledger.HistoryDBConfig{
 					Enabled: true,

sampleconfig/core.yaml

@@ -732,6 +732,8 @@ ledger:
     # Private data is purged from the peer's private data store based on
     # the collection property blockToLive or an explicit chaincode call to PurgePrivateData().
     purgeInterval: 100
+    # Whether to log private data keys purged from private data store (INFO level) when explicitly purged via chaincode
+    purgedKeyAuditLogging: true
 
   snapshots:
     # Path on the file system where peer will store ledger snapshots