Skip to content

Commit

Permalink
Hardening unaggregated attestation queue check (#7834)
Browse files Browse the repository at this point in the history 
* Add more checks and tests

* Move VerifyLmdFfgConsistency

* Move VerifyFinalizedConsistency

* Move VerifyFinalizedConsistency higher

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
Co-authored-by: Nishant Das <nishdas93@gmail.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>
  • Loading branch information
@terencechain @rauljordan
@nisdas @prylabs-bulldozer
4 people committed Nov 17, 2020
1 parent d3f8599 commit 7c54cfe
Show file tree
Hide file tree
Showing 3 changed files with 190 additions and 68 deletions.
32 changes: 19 additions & 13 deletions beacon-chain/sync/pending_attestations_queue.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ import (
pubsub "github.com/libp2p/go-libp2p-pubsub"
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
"github.com/prysmaticlabs/prysm/shared/bls"
"github.com/prysmaticlabs/prysm/shared/bytesutil"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/prysmaticlabs/prysm/shared/rand"
Expand Down Expand Up @@ -83,23 +82,30 @@ func (s *Service) processPendingAtts(ctx context.Context) error {
}
}
} else {
// Save the pending unaggregated attestation to the pool if the BLS signature is
// valid.
if _, err := bls.SignatureFromBytes(att.Aggregate.Signature); err != nil {
continue
// This is an important validation before retrieving attestation pre state to defend against
// attestation's target intentionally reference checkpoint that's long ago.
// Verify current finalized checkpoint is an ancestor of the block defined by the attestation's beacon block root.
if err := s.chain.VerifyFinalizedConsistency(ctx, att.Aggregate.Data.BeaconBlockRoot); err != nil {
return err
}
if err := s.attPool.SaveUnaggregatedAttestation(att.Aggregate); err != nil {
if err := s.chain.VerifyLmdFfgConsistency(ctx, att.Aggregate); err != nil {
return err
}

// Verify signed aggregate has a valid signature.
if _, err := bls.SignatureFromBytes(signedAtt.Signature); err != nil {
continue
preState, err := s.chain.AttestationPreState(ctx, att.Aggregate)
if err != nil {
return err
}
valid := s.validateUnaggregatedAttWithState(ctx, att.Aggregate, preState)
if valid == pubsub.ValidationAccept {
if err := s.attPool.SaveUnaggregatedAttestation(att.Aggregate); err != nil {
return err
}
s.setSeenCommitteeIndicesSlot(att.Aggregate.Data.Slot, att.Aggregate.Data.CommitteeIndex, att.Aggregate.AggregationBits)

// 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")
// 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")
}
}
}
}
Expand Down
147 changes: 119 additions & 28 deletions beacon-chain/sync/pending_attestations_queue_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,45 +60,83 @@ func TestProcessPendingAtts_HasBlockSaveUnAggregatedAtt(t *testing.T) {
hook := logTest.NewGlobal()
db, _ := dbtest.SetupDB(t)
p1 := p2ptest.NewTestP2P(t)
validators := uint64(256)
testutil.ResetCache()
beaconState, privKeys := testutil.DeterministicGenesisState(t, validators)

sb := testutil.NewBeaconBlock()
require.NoError(t, db.SaveBlock(context.Background(), sb))
root, err := sb.Block.HashTreeRoot()
require.NoError(t, err)

aggBits := bitfield.NewBitlist(8)
aggBits.SetBitAt(1, true)
att := &ethpb.Attestation{
Data: &ethpb.AttestationData{
BeaconBlockRoot: root[:],
Source: &ethpb.Checkpoint{Epoch: 0, Root: bytesutil.PadTo([]byte("hello-world"), 32)},
Target: &ethpb.Checkpoint{Epoch: 0, Root: root[:]},
},
AggregationBits: aggBits,
}

committee, err := helpers.BeaconCommitteeFromState(beaconState, att.Data.Slot, att.Data.CommitteeIndex)
assert.NoError(t, err)
attestingIndices := attestationutil.AttestingIndices(att.AggregationBits, committee)
assert.NoError(t, err)
attesterDomain, err := helpers.Domain(beaconState.Fork(), 0, params.BeaconConfig().DomainBeaconAttester, beaconState.GenesisValidatorRoot())
require.NoError(t, err)
hashTreeRoot, err := helpers.ComputeSigningRoot(att.Data, attesterDomain)
assert.NoError(t, err)
for _, i := range attestingIndices {
att.Signature = privKeys[i].Sign(hashTreeRoot[:]).Marshal()
}

// Arbitrary aggregator index for testing purposes.
aggregatorIndex := committee[0]
sig, err := helpers.ComputeDomainAndSign(beaconState, 0, att.Data.Slot, params.BeaconConfig().DomainSelectionProof, privKeys[aggregatorIndex])
require.NoError(t, err)
aggregateAndProof := &ethpb.AggregateAttestationAndProof{
SelectionProof: sig,
Aggregate: att,
AggregatorIndex: aggregatorIndex,
}
aggreSig, err := helpers.ComputeDomainAndSign(beaconState, 0, aggregateAndProof, params.BeaconConfig().DomainAggregateAndProof, privKeys[aggregatorIndex])
require.NoError(t, err)

require.NoError(t, beaconState.SetGenesisTime(uint64(time.Now().Unix())))

c, err := lru.New(10)
require.NoError(t, err)
r := &Service{
p2p: p1,
db: db,
chain: &mock.ChainService{Genesis: timeutils.Now()},
p2p: p1,
db: db,
chain: &mock.ChainService{Genesis: time.Now(),
State: beaconState,
FinalizedCheckPoint: &ethpb.Checkpoint{
Root: aggregateAndProof.Aggregate.Data.BeaconBlockRoot,
Epoch: 0,
}},
blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
attPool: attestations.NewPool(),
stateSummaryCache: cache.NewStateSummaryCache(),
seenAttestationCache: c,
}

priv, err := bls.RandKey()
require.NoError(t, err)
a := &ethpb.AggregateAttestationAndProof{
Aggregate: &ethpb.Attestation{
Signature: priv.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 := b.Block.HashTreeRoot()
sb = testutil.NewBeaconBlock()
r32, err := sb.Block.HashTreeRoot()
require.NoError(t, err)
require.NoError(t, r.db.SaveBlock(context.Background(), sb))
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)}}
r.blkRootToPendingAtts[r32] = []*ethpb.SignedAggregateAttestationAndProof{{Message: aggregateAndProof, Signature: aggreSig}}
require.NoError(t, r.processPendingAtts(context.Background()))

atts, err := r.attPool.UnaggregatedAttestations()
require.NoError(t, err)
assert.Equal(t, 1, len(atts), "Did not save unaggregated att")
assert.DeepEqual(t, a.Aggregate, atts[0], "Incorrect saved att")
assert.DeepEqual(t, att, atts[0], "Incorrect saved att")
assert.Equal(t, 0, len(r.attPool.AggregatedAttestations()), "Did save aggregated att")
require.LogsContain(t, hook, "Verified and saved pending attestations to pool")
}
Expand All @@ -110,7 +148,7 @@ func TestProcessPendingAtts_NoBroadcastWithBadSignature(t *testing.T) {
r := &Service{
p2p: p1,
db: db,
chain: &mock.ChainService{Genesis: timeutils.Now()},
chain: &mock.ChainService{Genesis: timeutils.Now(), FinalizedCheckPoint: &ethpb.Checkpoint{Root: make([]byte, 32)}},
blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
attPool: attestations.NewPool(),
stateSummaryCache: cache.NewStateSummaryCache(),
Expand Down Expand Up @@ -146,9 +184,62 @@ func TestProcessPendingAtts_NoBroadcastWithBadSignature(t *testing.T) {
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
validators := uint64(256)
testutil.ResetCache()
s, privKeys := testutil.DeterministicGenesisState(t, validators)
aggBits := bitfield.NewBitlist(8)
aggBits.SetBitAt(1, true)
att := &ethpb.Attestation{
Data: &ethpb.AttestationData{
BeaconBlockRoot: r32[:],
Source: &ethpb.Checkpoint{Epoch: 0, Root: bytesutil.PadTo([]byte("hello-world"), 32)},
Target: &ethpb.Checkpoint{Epoch: 0, Root: r32[:]},
},
AggregationBits: aggBits,
}
committee, err := helpers.BeaconCommitteeFromState(s, att.Data.Slot, att.Data.CommitteeIndex)
assert.NoError(t, err)
attestingIndices := attestationutil.AttestingIndices(att.AggregationBits, committee)
assert.NoError(t, err)
attesterDomain, err := helpers.Domain(s.Fork(), 0, params.BeaconConfig().DomainBeaconAttester, s.GenesisValidatorRoot())
require.NoError(t, err)
hashTreeRoot, err := helpers.ComputeSigningRoot(att.Data, attesterDomain)
assert.NoError(t, err)
for _, i := range attestingIndices {
att.Signature = privKeys[i].Sign(hashTreeRoot[:]).Marshal()
}

// Arbitrary aggregator index for testing purposes.
aggregatorIndex := committee[0]
sig, err := helpers.ComputeDomainAndSign(s, 0, att.Data.Slot, params.BeaconConfig().DomainSelectionProof, privKeys[aggregatorIndex])
require.NoError(t, err)
aggregateAndProof := &ethpb.AggregateAttestationAndProof{
SelectionProof: sig,
Aggregate: att,
AggregatorIndex: aggregatorIndex,
}
aggreSig, err := helpers.ComputeDomainAndSign(s, 0, aggregateAndProof, params.BeaconConfig().DomainAggregateAndProof, privKeys[aggregatorIndex])
require.NoError(t, err)

require.NoError(t, s.SetGenesisTime(uint64(time.Now().Unix())))
c, err := lru.New(10)
require.NoError(t, err)
r = &Service{
p2p: p1,
db: db,
chain: &mock.ChainService{Genesis: time.Now(),
State: s,
FinalizedCheckPoint: &ethpb.Checkpoint{
Root: aggregateAndProof.Aggregate.Data.BeaconBlockRoot,
Epoch: 0,
}},
blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
attPool: attestations.NewPool(),
stateSummaryCache: cache.NewStateSummaryCache(),
seenAttestationCache: c,
}

r.blkRootToPendingAtts[r32] = []*ethpb.SignedAggregateAttestationAndProof{{Message: aggregateAndProof, Signature: aggreSig}}
require.NoError(t, r.processPendingAtts(context.Background()))

assert.Equal(t, true, p1.BroadcastCalled, "Could not broadcast the good aggregate")
Expand Down Expand Up @@ -239,7 +330,7 @@ func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
assert.DeepEqual(t, att, r.attPool.AggregatedAttestations()[0], "Incorrect saved att")
atts, err := r.attPool.UnaggregatedAttestations()
require.NoError(t, err)
assert.Equal(t, 0, len(atts), "Did save unaggregated att")
assert.Equal(t, 0, len(atts), "Did save aggregated att")
require.LogsContain(t, hook, "Verified and saved pending attestations to pool")
}

Expand Down
79 changes: 52 additions & 27 deletions beacon-chain/sync/validate_beacon_attestation.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
"github.com/prysmaticlabs/prysm/beacon-chain/state"
"github.com/prysmaticlabs/prysm/shared/bytesutil"
"github.com/prysmaticlabs/prysm/shared/traceutil"
"go.opencensus.io/trace"
Expand Down Expand Up @@ -79,6 +80,7 @@ func (s *Service) validateCommitteeIndexBeaconAttestation(ctx context.Context, p
if s.hasSeenCommitteeIndicesSlot(att.Data.Slot, att.Data.CommitteeIndex, att.AggregationBits) {
return pubsub.ValidationIgnore
}

// Reject an attestation if it references an invalid block.
if s.hasBadBlock(bytesutil.ToBytes32(att.Data.BeaconBlockRoot)) ||
s.hasBadBlock(bytesutil.ToBytes32(att.Data.Target.Root)) ||
Expand All @@ -94,17 +96,13 @@ func (s *Service) validateCommitteeIndexBeaconAttestation(ctx context.Context, p
return pubsub.ValidationIgnore
}

if err := s.chain.VerifyLmdFfgConsistency(ctx, att); err != nil {
if err := s.chain.VerifyFinalizedConsistency(ctx, att.Data.BeaconBlockRoot); err != nil {
traceutil.AnnotateError(span, err)
return pubsub.ValidationReject
}

// The attestation's committee index (attestation.data.index) is for the correct subnet.
digest, err := s.forkDigest()
if err != nil {
log.WithError(err).Error("Failed to compute fork digest")
if err := s.chain.VerifyLmdFfgConsistency(ctx, att); err != nil {
traceutil.AnnotateError(span, err)
return pubsub.ValidationIgnore
return pubsub.ValidationReject
}

preState, err := s.chain.AttestationPreState(ctx, att)
Expand All @@ -113,57 +111,84 @@ func (s *Service) validateCommitteeIndexBeaconAttestation(ctx context.Context, p
traceutil.AnnotateError(span, err)
return pubsub.ValidationIgnore
}
valCount, err := helpers.ActiveValidatorCount(preState, helpers.SlotToEpoch(att.Data.Slot))

validationRes := s.validateUnaggregatedAttTopic(ctx, att, preState, *originalTopic)
if validationRes != pubsub.ValidationAccept {
return validationRes
}

validationRes = s.validateUnaggregatedAttWithState(ctx, att, preState)
if validationRes != pubsub.ValidationAccept {
return validationRes
}

s.setSeenCommitteeIndicesSlot(att.Data.Slot, att.Data.CommitteeIndex, att.AggregationBits)

msg.ValidatorData = att

return pubsub.ValidationAccept
}

// This validates beacon unaggregated attestation has correct topic string.
func (s *Service) validateUnaggregatedAttTopic(ctx context.Context, a *eth.Attestation, bs *state.BeaconState, t string) pubsub.ValidationResult {
ctx, span := trace.StartSpan(ctx, "sync.validateUnaggregatedAttTopic")
defer span.End()

valCount, err := helpers.ActiveValidatorCount(bs, helpers.SlotToEpoch(a.Data.Slot))
if err != nil {
log.WithError(err).Error("Could not retrieve active validator count")
traceutil.AnnotateError(span, err)
return pubsub.ValidationIgnore
}

count := helpers.SlotCommitteeCount(valCount)
if att.Data.CommitteeIndex > count {
if a.Data.CommitteeIndex > count {
return pubsub.ValidationReject
}

subnet := helpers.ComputeSubnetForAttestation(valCount, att)
if !strings.HasPrefix(*originalTopic, fmt.Sprintf(format, digest, subnet)) {
subnet := helpers.ComputeSubnetForAttestation(valCount, a)
format := p2p.GossipTypeMapping[reflect.TypeOf(&eth.Attestation{})]
digest, err := s.forkDigest()
if err != nil {
log.WithError(err).Error("Failed to compute fork digest")
traceutil.AnnotateError(span, err)
return pubsub.ValidationIgnore
}
if !strings.HasPrefix(t, fmt.Sprintf(format, digest, subnet)) {
return pubsub.ValidationReject
}

committee, err := helpers.BeaconCommitteeFromState(preState, att.Data.Slot, att.Data.CommitteeIndex)
return pubsub.ValidationAccept
}

// This validates beacon unaggregated attestation using the given state, the validation consists of bitfield length and count consistency
// and signature verification.
func (s *Service) validateUnaggregatedAttWithState(ctx context.Context, a *eth.Attestation, bs *state.BeaconState) pubsub.ValidationResult {
ctx, span := trace.StartSpan(ctx, "sync.validateUnaggregatedAttWithState")
defer span.End()

committee, err := helpers.BeaconCommitteeFromState(bs, a.Data.Slot, a.Data.CommitteeIndex)
if err != nil {
traceutil.AnnotateError(span, err)
return pubsub.ValidationIgnore
}

// Verify number of aggregation bits matches the committee size.
if err := helpers.VerifyBitfieldLength(att.AggregationBits, uint64(len(committee))); err != nil {
if err := helpers.VerifyBitfieldLength(a.AggregationBits, uint64(len(committee))); err != nil {
return pubsub.ValidationReject
}

// Attestation must be unaggregated and the bit index must exist in the range of committee indices.
// Note: eth2 spec suggests (len(get_attesting_indices(state, attestation.data, attestation.aggregation_bits)) == 1)
// however this validation can be achieved without use of get_attesting_indices which is an O(n) lookup.
if att.AggregationBits.Count() != 1 || att.AggregationBits.BitIndices()[0] >= len(committee) {
if a.AggregationBits.Count() != 1 || a.AggregationBits.BitIndices()[0] >= len(committee) {
return pubsub.ValidationReject
}

if err := blocks.VerifyAttestationSignature(ctx, preState, att); err != nil {
if err := blocks.VerifyAttestationSignature(ctx, bs, a); err != nil {
log.WithError(err).Error("Could not verify attestation")
traceutil.AnnotateError(span, err)
return pubsub.ValidationReject
}

// Verify current finalized checkpoint is an ancestor of the block defined by the attestation's beacon block root.
if err := s.chain.VerifyFinalizedConsistency(ctx, att.Data.BeaconBlockRoot); err != nil {
traceutil.AnnotateError(span, err)
return pubsub.ValidationReject
}

s.setSeenCommitteeIndicesSlot(att.Data.Slot, att.Data.CommitteeIndex, att.AggregationBits)

msg.ValidatorData = att

return pubsub.ValidationAccept
}

Expand Down

0 comments on commit 7c54cfe

Please sign in to comment.