Filter Canonical Attester for RPC (#2551)

* exclusive of finalized block

* add filter to only include canonical attestation

* comments

* grammer

* gaz

* typo

* fixed existing tests

* added test for IsAttCanonical

* add nil blocks test

beacon-chain/operations/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
     deps = [
         "//beacon-chain/db:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
+        "//shared/bytesutil:go_default_library",
         "//shared/event:go_default_library",
         "//shared/hashutil:go_default_library",
         "//shared/messagehandler:go_default_library",

beacon-chain/operations/service.go

@@ -2,6 +2,7 @@
 package operations
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"sort"
@@ -9,6 +10,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	"github.com/prysmaticlabs/prysm/shared/bytesutil"
 	"github.com/prysmaticlabs/prysm/shared/event"
 	"github.com/prysmaticlabs/prysm/shared/hashutil"
 	handler "github.com/prysmaticlabs/prysm/shared/messagehandler"
@@ -208,6 +210,33 @@ func (s *Service) HandleAttestations(ctx context.Context, message proto.Message)
 	return nil
 }
 
+// IsAttCanonical returns true if the input attestation is voting on the canonical chain, false
+// otherwise. The steps to verify are:
+//	1.) retrieve the voted block
+//	2.) retrieve the canonical block by using voted block's slot number
+//	3.) return true if voted block root and the canonical block root are the same
+func (s *Service) IsAttCanonical(ctx context.Context, att *pb.Attestation) (bool, error) {
+	votedBlk, err := s.beaconDB.Block(bytesutil.ToBytes32(att.Data.BeaconBlockRootHash32))
+	if err != nil {
+		return false, fmt.Errorf("could not hash block: %v", err)
+	}
+	if votedBlk == nil {
+		return false, nil
+	}
+	canonicalBlk, err := s.beaconDB.CanonicalBlockBySlot(ctx, votedBlk.Slot)
+	if err != nil {
+		return false, fmt.Errorf("could not hash block: %v", err)
+	}
+	if canonicalBlk == nil {
+		return false, nil
+	}
+	canonicalRoot, err := hashutil.HashBeaconBlock(canonicalBlk)
+	if err != nil {
+		return false, fmt.Errorf("could not hash block: %v", err)
+	}
+	return bytes.Equal(att.Data.BeaconBlockRootHash32, canonicalRoot[:]), nil
+}
+
 // removeOperations removes the processed operations from operation pool and DB.
 func (s *Service) removeOperations() {
 	incomingBlockSub := s.incomingProcessedBlockFeed.Subscribe(s.incomingProcessedBlock)

beacon-chain/operations/service_test.go

@@ -326,3 +326,75 @@ func TestReceiveBlkRemoveOps_Ok(t *testing.T) {
 		t.Errorf("Attestation pool should be empty but got a length of %d", len(atts))
 	}
 }
+
+func TestIsCanonical_CanGetCanonical(t *testing.T) {
+	db := internal.SetupDB(t)
+	defer internal.TeardownDB(t, db)
+	s := NewOpsPoolService(context.Background(), &Config{BeaconDB: db})
+
+	cb1 := &pb.BeaconBlock{Slot: 999, ParentRootHash32: []byte{'A'}}
+	if err := s.beaconDB.SaveBlock(cb1); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.beaconDB.UpdateChainHead(context.Background(), cb1, &pb.BeaconState{}); err != nil {
+		t.Fatal(err)
+	}
+	r1, err := hashutil.HashBeaconBlock(cb1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	att1 := &pb.Attestation{Data: &pb.AttestationData{BeaconBlockRootHash32: r1[:]}}
+	canonical, err := s.IsAttCanonical(context.Background(), att1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !canonical {
+		t.Error("Attestation should be canonical")
+	}
+
+	cb2 := &pb.BeaconBlock{Slot: 999, ParentRootHash32: []byte{'B'}}
+	if err := s.beaconDB.SaveBlock(cb2); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.beaconDB.UpdateChainHead(context.Background(), cb2, &pb.BeaconState{}); err != nil {
+		t.Fatal(err)
+	}
+	canonical, err = s.IsAttCanonical(context.Background(), att1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if canonical {
+		t.Error("Attestation should not be canonical")
+	}
+}
+
+func TestIsCanonical_NilBlocks(t *testing.T) {
+	db := internal.SetupDB(t)
+	defer internal.TeardownDB(t, db)
+	s := NewOpsPoolService(context.Background(), &Config{BeaconDB: db})
+
+	canonical, err := s.IsAttCanonical(context.Background(), &pb.Attestation{Data: &pb.AttestationData{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if canonical {
+		t.Error("Attestation shouldn't be canonical")
+	}
+
+	cb1 := &pb.BeaconBlock{Slot: 999, ParentRootHash32: []byte{'A'}}
+	if err := s.beaconDB.SaveBlock(cb1); err != nil {
+		t.Fatal(err)
+	}
+	r1, err := hashutil.HashBeaconBlock(cb1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	att1 := &pb.Attestation{Data: &pb.AttestationData{BeaconBlockRootHash32: r1[:]}}
+	canonical, err = s.IsAttCanonical(context.Background(), att1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if canonical {
+		t.Error("Attestation shouldn't be canonical")
+	}
+}

beacon-chain/rpc/proposer_server.go

@@ -155,6 +155,21 @@ func (ps *ProposerServer) PendingAttestations(ctx context.Context, req *pb.Pendi
 			}
 			continue
 		}
+
+		if featureconfig.FeatureConfig().EnableCanonicalAttestationFilter {
+			canonical, err := ps.operationService.IsAttCanonical(ctx, att)
+			if err != nil {
+				// Delete attestation that failed to verify as canonical.
+				if err := ps.beaconDB.DeleteAttestation(att); err != nil {
+					return nil, fmt.Errorf("could not delete failed attestation: %v", err)
+				}
+				return nil, fmt.Errorf("could not verify canonical attestation: %v", err)
+			}
+			// Skip the attestation if it's not canonical.
+			if !canonical {
+				continue
+			}
+		}
 		validAtts = append(validAtts, att)
 	}
 

beacon-chain/rpc/service.go

@@ -45,6 +45,7 @@ type chainService interface {
 
 type operationService interface {
 	PendingAttestations(ctx context.Context) ([]*pbp2p.Attestation, error)
+	IsAttCanonical(ctx context.Context, att *pbp2p.Attestation) (bool, error)
 	HandleAttestations(context.Context, proto.Message) error
 	IncomingAttFeed() *event.Feed
 }

beacon-chain/rpc/service_test.go

@@ -50,6 +50,10 @@ func (ms *mockOperationService) HandleAttestations(_ context.Context, _ proto.Me
 	return nil
 }
 
+func (ms *mockOperationService) IsAttCanonical(_ context.Context, att *pb.Attestation) (bool, error) {
+	return true, nil
+}
+
 func (ms *mockOperationService) PendingAttestations(_ context.Context) ([]*pb.Attestation, error) {
 	if ms.pendingAttestations != nil {
 		return ms.pendingAttestations, nil

shared/featureconfig/config.go

@@ -25,14 +25,15 @@ var log = logrus.WithField("prefix", "flags")
 
 // FeatureFlagConfig is a struct to represent what features the client will perform on runtime.
 type FeatureFlagConfig struct {
-	VerifyAttestationSigs         bool // VerifyAttestationSigs declares if the client will verify attestations.
-	EnableComputeStateRoot        bool // EnableComputeStateRoot implementation on server side.
-	EnableCrosslinks              bool // EnableCrosslinks in epoch processing.
-	EnableCheckBlockStateRoot     bool // EnableCheckBlockStateRoot in block processing.
-	DisableHistoricalStatePruning bool // DisableHistoricalStatePruning when updating finalized states.
-	DisableGossipSub              bool // DisableGossipSub in p2p messaging.
-	EnableCommitteesCache         bool // EnableCommitteesCache for state transition.
-	CacheTreeHash                 bool // CacheTreeHash determent whether tree hashes will be cached.
+	VerifyAttestationSigs            bool // VerifyAttestationSigs declares if the client will verify attestations.
+	EnableComputeStateRoot           bool // EnableComputeStateRoot implementation on server side.
+	EnableCrosslinks                 bool // EnableCrosslinks in epoch processing.
+	EnableCheckBlockStateRoot        bool // EnableCheckBlockStateRoot in block processing.
+	EnableCanonicalAttestationFilter bool // EnableCanonicalAttestationFilter for RPC server.
+	DisableHistoricalStatePruning    bool // DisableHistoricalStatePruning when updating finalized states.
+	DisableGossipSub                 bool // DisableGossipSub in p2p messaging.
+	EnableCommitteesCache            bool // EnableCommitteesCache for state transition.
+	CacheTreeHash                    bool // CacheTreeHash determent whether tree hashes will be cached.
 }
 
 var featureConfig *FeatureFlagConfig
@@ -82,7 +83,10 @@ func ConfigureBeaconFeatures(ctx *cli.Context) {
 		log.Info("Disabled gossipsub, using floodsub")
 		cfg.DisableGossipSub = true
 	}
-
+	if ctx.GlobalBool(EnableCanonicalAttestationFilter.Name) {
+		log.Info("Enabled canonical attestation filter")
+		cfg.EnableCanonicalAttestationFilter = true
+	}
 	InitFeatureConfig(cfg)
 }
 

shared/featureconfig/flags.go

@@ -35,6 +35,11 @@ var (
 		Name:  "enable-check-block-state-root",
 		Usage: "Enable check block state root in block processing, default is disabled.",
 	}
+	// EnableCanonicalAttestationFilter filters and sends canonical attestation to RPC requests.
+	EnableCanonicalAttestationFilter = cli.BoolFlag{
+		Name:  "enable-canonical-attestation-filter",
+		Usage: "Enable filtering and sending canonical attestations to RPC request, default is disabled.",
+	}
 	// DisableHistoricalStatePruningFlag allows the database to keep old historical states.
 	DisableHistoricalStatePruningFlag = cli.BoolFlag{
 		Name:  "disable-historical-state-pruning",
@@ -57,6 +62,7 @@ var BeaconChainFlags = []cli.Flag{
 	EnableComputeStateRootFlag,
 	EnableCrosslinksFlag,
 	EnableCheckBlockStateRootFlag,
+	EnableCanonicalAttestationFilter,
 	DisableHistoricalStatePruningFlag,
 	DisableGossipSubFlag,
 	CacheTreeHashFlag,