Hardening aggregated attestation queue check (#7826)

beacon-chain/sync/pending_attestations_queue.go

@@ -75,6 +75,7 @@ func (s *Service) processPendingAtts(ctx context.Context) error {
 						if err := s.attPool.SaveAggregatedAttestation(att.Aggregate); err != nil {
 							return err
 						}
+						s.setAggregatorIndexEpochSeen(att.Aggregate.Data.Target.Epoch, att.AggregatorIndex)
 
 						// Broadcasting the signed attestation again once a node is able to process it.
 						if err := s.p2p.Broadcast(ctx, signedAtt); err != nil {

beacon-chain/sync/pending_attestations_queue_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/p2p/enr"
+	lru "github.com/hashicorp/golang-lru"
 	"github.com/libp2p/go-libp2p-core/network"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/go-bitfield"
@@ -173,7 +174,7 @@ func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
 		Data: &ethpb.AttestationData{
 			BeaconBlockRoot: root[:],
 			Source:          &ethpb.Checkpoint{Epoch: 0, Root: bytesutil.PadTo([]byte("hello-world"), 32)},
-			Target:          &ethpb.Checkpoint{Epoch: 0, Root: bytesutil.PadTo([]byte("hello-world"), 32)},
+			Target:          &ethpb.Checkpoint{Epoch: 0, Root: root[:]},
 		},
 		AggregationBits: aggBits,
 	}
@@ -207,6 +208,8 @@ func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
 
 	require.NoError(t, beaconState.SetGenesisTime(uint64(time.Now().Unix())))
 
+	c, err := lru.New(10)
+	require.NoError(t, err)
 	r := &Service{
 		p2p: p1,
 		db:  db,
@@ -219,6 +222,7 @@ func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
 		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
 		attPool:              attestations.NewPool(),
 		stateSummaryCache:    cache.NewStateSummaryCache(),
+		seenAttestationCache: c,
 	}
 
 	sb = testutil.NewBeaconBlock()

beacon-chain/sync/validate_aggregate_proof.go

@@ -80,15 +80,6 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms
 		return pubsub.ValidationIgnore
 	}
 
-	// Verify attestation target root is consistent with the head root.
-	// This verification is not in the spec, however we guard against it as it opens us up
-	// to weird edge cases during verification. The attestation technically could be used to add value to a block,
-	// but it's invalid in the spirit of the protocol. Here we choose safety over profit.
-	if err := s.chain.VerifyLmdFfgConsistency(ctx, m.Message.Aggregate); err != nil {
-		traceutil.AnnotateError(span, err)
-		return pubsub.ValidationReject
-	}
-
 	validationRes := s.validateAggregatedAtt(ctx, m)
 	if validationRes != pubsub.ValidationAccept {
 		return validationRes
@@ -105,6 +96,16 @@ func (s *Service) validateAggregatedAtt(ctx context.Context, signed *ethpb.Signe
 	ctx, span := trace.StartSpan(ctx, "sync.validateAggregatedAtt")
 	defer span.End()
 
+	// Verify attestation target root is consistent with the head root.
+	// This verification is not in the spec, however we guard against it as it opens us up
+	// to weird edge cases during verification. The attestation technically could be used to add value to a block,
+	// but it's invalid in the spirit of the protocol. Here we choose safety over profit.
+	if err := s.chain.VerifyLmdFfgConsistency(ctx, signed.Message.Aggregate); err != nil {
+		fmt.Println(err)
+		traceutil.AnnotateError(span, err)
+		return pubsub.ValidationReject
+	}
+
 	attSlot := signed.Message.Aggregate.Data.Slot
 
 	bs, err := s.chain.AttestationPreState(ctx, signed.Message.Aggregate)