Validate aggregate and proof subscriber

beacon-chain/sync/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "subscriber_beacon_attestation.go",
         "subscriber_beacon_blocks.go",
         "subscriber_handlers.go",
+        "validate_aggregate_proof.go",
         "validate_attester_slashing.go",
         "validate_beacon_attestation.go",
         "validate_beacon_blocks.go",
@@ -78,6 +79,7 @@ go_test(
         "rpc_test.go",
         "subscriber_beacon_blocks_test.go",
         "subscriber_test.go",
+        "validate_aggregate_proof_test.go",
         "validate_attester_slashing_test.go",
         "validate_beacon_attestation_test.go",
         "validate_beacon_blocks_test.go",
@@ -107,6 +109,7 @@ go_test(
         "@com_github_libp2p_go_libp2p_core//network:go_default_library",
         "@com_github_libp2p_go_libp2p_core//protocol:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
+        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_prysmaticlabs_go_ssz//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",

beacon-chain/sync/validate_aggregate_proof.go

@@ -0,0 +1,142 @@
+package sync
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/pkg/errors"
+	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
+	"github.com/prysmaticlabs/go-ssz"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
+	"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	"github.com/prysmaticlabs/prysm/shared/bls"
+	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+	"github.com/prysmaticlabs/prysm/shared/params"
+	"github.com/prysmaticlabs/prysm/shared/roughtime"
+	"go.opencensus.io/trace"
+)
+
+// validateAggregateAndProof verifies the aggregated signature and the selection proof is valid before forwarding to the
+// network and downstream services.
+func (r *RegularSync) validateAggregateAndProof(ctx context.Context, msg proto.Message, p p2p.Broadcaster, fromSelf bool) (bool, error) {
+	ctx, span := trace.StartSpan(ctx, "sync.validateAggregateAndProof")
+	defer span.End()
+
+	// To process the following it requires the recent blocks to be present in the database, so we'll skip
+	// validating or processing aggregated attestations until fully synced.
+	if r.initialSync.Syncing() {
+		return false, nil
+	}
+
+	m, ok := msg.(*pb.AggregateAndProof)
+	if !ok {
+		return false, nil
+	}
+
+	attSlot := m.Aggregate.Data.Slot
+
+	// Verify aggregate attestation has not already been seen via aggregate gossip, within a block, or through the creation locally.
+	// TODO(3835): Blocked by operation pool redesign
+
+	// Verify the block being voted for passes validation. The block should have passed validation if it's in the DB.
+	if !r.db.HasBlock(ctx, bytesutil.ToBytes32(m.Aggregate.Data.BeaconBlockRoot)) {
+		return false, errPointsToBlockNotInDatabase
+	}
+
+	s, err := r.chain.HeadState(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	// Verify attestation slot is within the last ATTESTATION_PROPAGATION_SLOT_RANGE slots.
+	currentSlot := (uint64(roughtime.Now().Unix()) - s.GenesisTime) / params.BeaconConfig().SecondsPerSlot
+	if attSlot > currentSlot || currentSlot > attSlot+params.BeaconConfig().AttestationPropagationSlotRange {
+		return false, fmt.Errorf("attestation slot out of range %d <= %d <= %d",
+			attSlot, currentSlot, attSlot+params.BeaconConfig().AttestationPropagationSlotRange)
+	}
+
+	if attSlot > s.Slot {
+		s, err = state.ProcessSlots(ctx, s, attSlot)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	// Verify validator index is within the aggregate's committee.
+	if err := validateIndexInCommittee(ctx, s, m.Aggregate, m.AggregatorIndex); err != nil {
+		return false, errors.Wrapf(err, "Could not validate index in committee")
+	}
+
+	// Verify selection proof reflects to the right validator and signature is valid.
+	if err := validateSelection(ctx, s, m.Aggregate.Data, m.AggregatorIndex, m.SelectionProof); err != nil {
+		return false, errors.Wrapf(err, "Could not validate selection for validator %d", m.AggregatorIndex)
+	}
+
+	// Verify aggregated attestation has a valid signature.
+	if err := blocks.VerifyAttestation(ctx, s, m.Aggregate); err != nil {
+		return false, err
+	}
+
+	return true, nil
+}
+
+// This validates the aggregator's index in state is within the attesting indices of the attestation.
+func validateIndexInCommittee(ctx context.Context, s *pb.BeaconState, a *ethpb.Attestation, validatorIndex uint64) error {
+	_, span := trace.StartSpan(ctx, "sync..validateIndexInCommittee")
+	defer span.End()
+
+	attestingIndices, err := helpers.AttestingIndices(s, a.Data, a.AggregationBits)
+	if err != nil {
+		return err
+	}
+	var withinCommittee bool
+	for _, i := range attestingIndices {
+		if validatorIndex == i {
+			withinCommittee = true
+			break
+		}
+	}
+	if !withinCommittee {
+		return fmt.Errorf("validator index %d is not within the committee: %v",
+			validatorIndex, attestingIndices)
+	}
+	return nil
+}
+
+// This validates selection proof by validating it's from the correct validator index of the slot and selection
+// proof is a valid signature.
+func validateSelection(ctx context.Context, s *pb.BeaconState, data *ethpb.AttestationData, validatorIndex uint64, proof []byte) error {
+	_, span := trace.StartSpan(ctx, "sync.validateSelection")
+	defer span.End()
+
+	slotSig, err := bls.SignatureFromBytes(proof)
+	if err != nil {
+		return err
+	}
+	aggregator, err := helpers.IsAggregator(s, data.Slot, data.CommitteeIndex, slotSig)
+	if err != nil {
+		return err
+	}
+	if !aggregator {
+		return fmt.Errorf("validator is not an aggregator for slot %d", data.Slot)
+	}
+
+	domain := helpers.Domain(s.Fork, helpers.SlotToEpoch(data.Slot), params.BeaconConfig().DomainBeaconAttester)
+	slotMsg, err := ssz.HashTreeRoot(data.Slot)
+	if err != nil {
+		return err
+	}
+	pubKey, err := bls.PublicKeyFromBytes(s.Validators[validatorIndex].PublicKey)
+	if err != nil {
+		return err
+	}
+	if !slotSig.Verify(slotMsg[:], pubKey, domain) {
+		return errors.New("could not validate slot signature")
+	}
+
+	return nil
+}

beacon-chain/sync/validate_aggregate_proof_test.go

@@ -0,0 +1,213 @@
+package sync
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
+	"github.com/prysmaticlabs/go-bitfield"
+	"github.com/prysmaticlabs/go-ssz"
+	mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
+	dbtest "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	p2ptest "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing"
+	mockSync "github.com/prysmaticlabs/prysm/beacon-chain/sync/initial-sync/testing"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	"github.com/prysmaticlabs/prysm/shared/bls"
+	"github.com/prysmaticlabs/prysm/shared/params"
+	"github.com/prysmaticlabs/prysm/shared/testutil"
+)
+
+func TestVerifyIndexInCommittee_CanVerify(t *testing.T) {
+	ctx := context.Background()
+	params.UseMinimalConfig()
+	defer params.UseMainnetConfig()
+
+	validators := uint64(64)
+	deposits, _, _ := testutil.SetupInitialDeposits(t, validators)
+	s, err := state.GenesisBeaconState(deposits, uint64(0), &ethpb.Eth1Data{BlockHash: make([]byte, 32)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.Slot = params.BeaconConfig().SlotsPerEpoch
+
+	bf := []byte{0xff}
+	att := &ethpb.Attestation{Data: &ethpb.AttestationData{
+		Target: &ethpb.Checkpoint{Epoch: 0}},
+		AggregationBits: bf}
+
+	indices, err := helpers.AttestingIndices(s, att.Data, att.AggregationBits)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := validateIndexInCommittee(ctx, s, att, indices[0]); err != nil {
+		t.Fatal(err)
+	}
+
+	wanted := "validator index 1000 is not within the committee"
+	if err := validateIndexInCommittee(ctx, s, att, 1000); !strings.Contains(err.Error(), wanted) {
+		t.Error("Did not receive wanted error")
+	}
+}
+
+func TestVerifySelection_NotAnAggregator(t *testing.T) {
+	ctx := context.Background()
+	params.UseMinimalConfig()
+	defer params.UseMainnetConfig()
+	deposits, _, privKeys := testutil.SetupInitialDeposits(t, 2048)
+	beaconState, err := state.GenesisBeaconState(deposits, uint64(0), &ethpb.Eth1Data{BlockHash: make([]byte, 32)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := privKeys[0].Sign([]byte{}, 0)
+	data := &ethpb.AttestationData{}
+
+	wanted := "validator is not an aggregator for slot"
+	if err := validateSelection(ctx, beaconState, data, 0, sig.Marshal()); !strings.Contains(err.Error(), wanted) {
+		t.Error("Did not receive wanted error")
+	}
+}
+
+func TestVerifySelection_BadSignature(t *testing.T) {
+	ctx := context.Background()
+	deposits, _, privKeys := testutil.SetupInitialDeposits(t, 256)
+	beaconState, err := state.GenesisBeaconState(deposits, uint64(0), &ethpb.Eth1Data{BlockHash: make([]byte, 32)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := privKeys[0].Sign([]byte{}, 0)
+	data := &ethpb.AttestationData{}
+
+	wanted := "could not validate slot signature"
+	if err := validateSelection(ctx, beaconState, data, 0, sig.Marshal()); !strings.Contains(err.Error(), wanted) {
+		t.Error("Did not receive wanted error")
+	}
+}
+
+func TestVerifySelection_CanVerify(t *testing.T) {
+	ctx := context.Background()
+	deposits, _, privKeys := testutil.SetupInitialDeposits(t, 256)
+	beaconState, err := state.GenesisBeaconState(deposits, uint64(0), &ethpb.Eth1Data{BlockHash: make([]byte, 32)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	data := &ethpb.AttestationData{}
+	slotRoot, err := ssz.HashTreeRoot(data.Slot)
+	if err != nil {
+		t.Fatal(err)
+	}
+	domain := helpers.Domain(beaconState.Fork, 0, params.BeaconConfig().DomainBeaconAttester)
+	sig := privKeys[0].Sign(slotRoot[:], domain)
+
+	if err := validateSelection(ctx, beaconState, data, 0, sig.Marshal()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestValidateAggregateAndProof_NoBlock(t *testing.T) {
+	db := dbtest.SetupDB(t)
+	defer dbtest.TeardownDB(t, db)
+
+	att := &ethpb.Attestation{
+		Data: &ethpb.AttestationData{
+			Source: &ethpb.Checkpoint{Epoch: 0, Root: []byte("hello-world")},
+			Target: &ethpb.Checkpoint{Epoch: 0, Root: []byte("hello-world")},
+		},
+	}
+
+	aggregateAndProof := &pb.AggregateAndProof{
+		SelectionProof:  []byte{'A'},
+		Aggregate:       att,
+		AggregatorIndex: 0,
+	}
+
+	r := &RegularSync{
+		db:          db,
+		initialSync: &mockSync.Sync{IsSyncing: false},
+	}
+
+	wanted := "attestation points to a block which is not in the database"
+	if _, err := r.validateAggregateAndProof(context.Background(), aggregateAndProof, &p2ptest.MockBroadcaster{}, false); !strings.Contains(err.Error(), wanted) {
+		t.Error("Did not receive wanted error")
+	}
+}
+
+func TestValidateAggregateAndProof_CanValidate(t *testing.T) {
+	db := dbtest.SetupDB(t)
+	defer dbtest.TeardownDB(t, db)
+
+	deposits, _, privKeys := testutil.SetupInitialDeposits(t, 256)
+	beaconState, err := state.GenesisBeaconState(deposits, uint64(0), &ethpb.Eth1Data{BlockHash: make([]byte, 32)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	b := &ethpb.BeaconBlock{}
+	db.SaveBlock(context.Background(), b)
+	root, _ := ssz.SigningRoot(b)
+
+	aggBits := bitfield.NewBitlist(3)
+	aggBits.SetBitAt(0, true)
+	att := &ethpb.Attestation{
+		Data: &ethpb.AttestationData{
+			BeaconBlockRoot: root[:],
+			Source:          &ethpb.Checkpoint{Epoch: 0, Root: []byte("hello-world")},
+			Target:          &ethpb.Checkpoint{Epoch: 0, Root: []byte("hello-world")},
+		},
+		AggregationBits: aggBits,
+	}
+
+	attestingIndices, err := helpers.AttestingIndices(beaconState, att.Data, att.AggregationBits)
+	if err != nil {
+		t.Error(err)
+	}
+	hashTreeRoot, err := ssz.HashTreeRoot(att.Data)
+	if err != nil {
+		t.Error(err)
+	}
+	domain := helpers.Domain(beaconState.Fork, 0, params.BeaconConfig().DomainBeaconAttester)
+	sigs := make([]*bls.Signature, len(attestingIndices))
+	for i, indice := range attestingIndices {
+		sig := privKeys[indice].Sign(hashTreeRoot[:], domain)
+		sigs[i] = sig
+	}
+	att.Signature = bls.AggregateSignatures(sigs).Marshal()[:]
+
+	slotRoot, err := ssz.HashTreeRoot(att.Data.Slot)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := privKeys[2].Sign(slotRoot[:], domain)
+	aggregateAndProof := &pb.AggregateAndProof{
+		SelectionProof:  sig.Marshal(),
+		Aggregate:       att,
+		AggregatorIndex: 2,
+	}
+
+	beaconState.GenesisTime = uint64(time.Now().Unix())
+	r := &RegularSync{
+		db:          db,
+		initialSync: &mockSync.Sync{IsSyncing: false},
+		chain: &mock.ChainService{Genesis: time.Now(),
+			State: beaconState,
+			FinalizedCheckPoint: &ethpb.Checkpoint{
+				Epoch: 0,
+			}},
+	}
+
+	validated, err := r.validateAggregateAndProof(context.Background(), aggregateAndProof, &p2ptest.MockBroadcaster{}, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !validated {
+		t.Fatal("Validated status is false")
+	}
+}

shared/params/config.go

@@ -51,6 +51,7 @@ type BeaconChainConfig struct {
 	MinEpochsToInactivityPenalty     uint64 `yaml:"MIN_EPOCHS_TO_INACTIVITY_PENALTY"`    // MinEpochsToInactivityPenalty defines the minimum amount of epochs since finality to begin penalizing inactivity.
 	Eth1FollowDistance               uint64 // Eth1FollowDistance is the number of eth1.0 blocks to wait before considering a new deposit for voting. This only applies after the chain as been started.
 	SafeSlotsToUpdateJustified       uint64 // SafeSlotsToUpdateJustified is the minimal slots needed to update justified check point.
+	AttestationPropagationSlotRange  uint64 // AttestationPropagationSlotRange is the maximum number of slots during which an attestation can be propagated.
 
 	// State list lengths
 	EpochsPerHistoricalVector uint64 `yaml:"EPOCHS_PER_HISTORICAL_VECTOR"` // EpochsPerHistoricalVector defines max length in epoch to store old historical stats in beacon state.
@@ -147,6 +148,7 @@ var defaultBeaconConfig = &BeaconChainConfig{
 	MinEpochsToInactivityPenalty:     4,
 	Eth1FollowDistance:               1024,
 	SafeSlotsToUpdateJustified:       8,
+	AttestationPropagationSlotRange:  32,
 
 	// State list length constants.
 	EpochsPerHistoricalVector: 65536,