Handle Nil Signature in Pending Queue (#7097)

* add fix and test

* Update beacon-chain/sync/pending_attestations_queue_test.go

Co-authored-by: Victor Farazdagi <simple.square@gmail.com>

Co-authored-by: terence tsao <terence@prysmaticlabs.com>
Co-authored-by: Victor Farazdagi <simple.square@gmail.com>

beacon-chain/sync/pending_attestations_queue.go

@@ -96,6 +96,11 @@ func (s *Service) processPendingAtts(ctx context.Context) error {
 					}
 					numberOfAttsRecovered.Inc()
 
+					// Verify signed aggregate has a valid signature.
+					if _, err := bls.SignatureFromBytes(signedAtt.Signature); err != nil {
+						continue
+					}
+
 					// Broadcasting the signed attestation again once a node is able to process it.
 					if err := s.p2p.Broadcast(ctx, signedAtt); err != nil {
 						log.WithError(err).Debug("Failed to broadcast")

beacon-chain/sync/pending_attestations_queue_test.go

@@ -102,6 +102,57 @@ func TestProcessPendingAtts_HasBlockSaveUnAggregatedAtt(t *testing.T) {
 	require.LogsContain(t, hook, "Verified and saved pending attestations to pool")
 }
 
+func TestProcessPendingAtts_NoBroadcastWithBadSignature(t *testing.T) {
+	db, _ := dbtest.SetupDB(t)
+	p1 := p2ptest.NewTestP2P(t)
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{NewStateMgmt: true})
+	defer resetCfg()
+
+	r := &Service{
+		p2p:                  p1,
+		db:                   db,
+		chain:                &mock.ChainService{Genesis: roughtime.Now()},
+		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		attPool:              attestations.NewPool(),
+		stateSummaryCache:    cache.NewStateSummaryCache(),
+	}
+
+	a := &ethpb.AggregateAttestationAndProof{
+		Aggregate: &ethpb.Attestation{
+			Signature:       bls.RandKey().Sign([]byte("foo")).Marshal(),
+			AggregationBits: bitfield.Bitlist{0x02},
+			Data: &ethpb.AttestationData{
+				Target:          &ethpb.Checkpoint{Root: make([]byte, 32)},
+				Source:          &ethpb.Checkpoint{Root: make([]byte, 32)},
+				BeaconBlockRoot: make([]byte, 32),
+			},
+		},
+		SelectionProof: make([]byte, 96),
+	}
+
+	b := testutil.NewBeaconBlock()
+	r32, err := stateutil.BlockRoot(b.Block)
+	require.NoError(t, err)
+	s := testutil.NewBeaconState()
+	require.NoError(t, r.db.SaveBlock(context.Background(), b))
+	require.NoError(t, r.db.SaveState(context.Background(), s, r32))
+
+	r.blkRootToPendingAtts[r32] = []*ethpb.SignedAggregateAttestationAndProof{{Message: a, Signature: make([]byte, 96)}}
+	require.NoError(t, r.processPendingAtts(context.Background()))
+
+	assert.Equal(t, false, p1.BroadcastCalled, "Broadcasted bad aggregate")
+	// Clear pool.
+	err = r.attPool.DeleteUnaggregatedAttestation(a.Aggregate)
+	require.NoError(t, err)
+
+	r.blkRootToPendingAtts[r32] = []*ethpb.SignedAggregateAttestationAndProof{{Message: a, Signature: make([]byte, 96)}}
+	// Make the signature a zero sig
+	r.blkRootToPendingAtts[r32][0].Signature[0] = 0xC0
+	require.NoError(t, r.processPendingAtts(context.Background()))
+
+	assert.Equal(t, true, p1.BroadcastCalled, "Could not broadcast the good aggregate")
+}
+
 func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
 	hook := logTest.NewGlobal()
 	db, _ := dbtest.SetupDB(t)