Clean up validation

eth2/beacon/operations/attestation_pool.py

@@ -12,7 +12,9 @@
 from .pool import OperationPool
 
 
-def is_valid_slot(attestation: Attestation, current_slot: Slot, config: Eth2Config) -> bool:
+def is_valid_slot(
+    attestation: Attestation, current_slot: Slot, config: Eth2Config
+) -> bool:
     try:
         validate_attestation_slot(
             attestation.data.slot,
@@ -28,29 +30,28 @@ def is_valid_slot(attestation: Attestation, current_slot: Slot, config: Eth2Conf
 
 class AttestationPool(OperationPool[Attestation]):
     def get_valid_attestation_by_current_slot(
-        self,
-        slot: Slot,
-        config: Eth2Config
+        self, slot: Slot, config: Eth2Config
     ) -> Tuple[Attestation, ...]:
         return tuple(
             filter(
                 lambda attestation: is_valid_slot(attestation, slot, config),
-                self._pool_storage.values()
+                self._pool_storage.values(),
             )
         )
 
     def get_acceptable_attestations(
         self,
         slot: Slot,
         committee_index: CommitteeIndex,
-        beacon_block_root: SigningRoot
+        beacon_block_root: SigningRoot,
     ) -> Tuple[Attestation, ...]:
         return tuple(
             filter(
-                lambda attestation:
-                beacon_block_root == attestation.data.beacon_block_root and
-                slot == attestation.data.slot and
-                committee_index == attestation.data.index,
-                self._pool_storage.values()
+                lambda attestation: (
+                    beacon_block_root == attestation.data.beacon_block_root
+                    and slot == attestation.data.slot
+                    and committee_index == attestation.data.index
+                ),
+                self._pool_storage.values(),
             )
         )

eth2/beacon/tools/builder/aggregator.py

@@ -1,11 +1,14 @@
 from typing import Sequence
 
 from eth_typing import BLSSignature
+from eth_utils import ValidationError
 from ssz import get_hash_tree_root, uint64
 
 from eth2._utils.bls import bls
 from eth2._utils.hash import hash_eth2
+from eth2.beacon.attestation_helpers import validate_indexed_attestation_aggregate_signature
 from eth2.beacon.committee_helpers import get_beacon_committee
+from eth2.beacon.epoch_processing_helpers import get_attesting_indices, get_indexed_attestation
 from eth2.beacon.helpers import compute_epoch_at_slot, get_domain
 from eth2.beacon.signature_domain import SignatureDomain
 from eth2.beacon.types.aggregate_and_proof import AggregateAndProof
@@ -76,13 +79,70 @@ def get_aggregate_from_valid_committee_attestations(
     ]
     aggregation_bits = tuple(map(any, zip(*all_aggregation_bits)))
 
+    assert len(attestations) > 0
+
     return Attestation(
         data=attestations[0].data,
         aggregation_bits=Bitfield(aggregation_bits),
         signature=aggregate_signature,
     )
 
 
+#
+# Validation
+#
+
+
+def validate_aggregate_and_proof(
+    state: BeaconState,
+    aggregate_and_proof: AggregateAndProof,
+    attestation_propagation_slot_range: int,
+    config: CommitteeConfig,
+) -> None:
+    """
+    Validate aggregate_and_proof
+
+    Reference: https://github.com/ethereum/eth2.0-specs/blob/v09x/specs/networking/p2p-interface.md#global-topics  # noqa: E501
+    """
+    if (
+        aggregate_and_proof.aggregate.data.slot + attestation_propagation_slot_range < state.slot
+        or aggregate_and_proof.aggregate.data.slot > state.slot
+    ):
+        raise ValidationError(
+            "aggregate_and_proof.aggregate.data.slot should be within the last"
+            " {attestation_propagation_slot_range} slots. Got"
+            f" aggregate_and_proof.aggregate.data.slot={aggregate_and_proof.aggregate.data.slot},"
+            f" current slot={state.slot}"
+        )
+
+    attesting_indices = get_attesting_indices(
+        state,
+        aggregate_and_proof.aggregate.data,
+        aggregate_and_proof.aggregate.aggregation_bits,
+        config,
+    )
+    if aggregate_and_proof.index not in attesting_indices:
+        raise ValidationError(
+            f"The aggregator index ({aggregate_and_proof.index}) is not within"
+            f" the aggregate's committee {attesting_indices}"
+        )
+
+    if not is_aggregator(
+        state,
+        aggregate_and_proof.aggregate.data.slot,
+        aggregate_and_proof.aggregate.data.index,
+        aggregate_and_proof.selection_proof,
+        config,
+    ):
+        raise ValidationError(
+            f"The given validator {aggregate_and_proof.index} is not selected aggregator"
+        )
+
+    validate_aggregator_proof(state, aggregate_and_proof, config)
+
+    validate_aggregate_signature(state, aggregate_and_proof.aggregate, config)
+
+
 def validate_aggregator_proof(
     state: BeaconState, aggregate_and_proof: AggregateAndProof, config: CommitteeConfig
 ) -> None:
@@ -102,3 +162,16 @@ def validate_aggregator_proof(
         signature=aggregate_and_proof.selection_proof,
         domain=domain,
     )
+
+
+def validate_aggregate_signature(
+    state: BeaconState,
+    attestation: Attestation,
+    config: CommitteeConfig
+) -> None:
+    indexed_attestation = get_indexed_attestation(state, attestation, config)
+    validate_indexed_attestation_aggregate_signature(
+        state,
+        indexed_attestation,
+        config.SLOTS_PER_EPOCH,
+    )

eth2/beacon/tools/factories.py

@@ -30,7 +30,12 @@ def _create(
     ) -> BaseBeaconChain:
         override_lengths(cls.config)
 
-        keymap = mk_keymap_of_size(cls.num_validators)
+        if "num_validators" in kwargs:
+            num_validators = kwargs["num_validators"]
+        else:
+            num_validators = cls.num_validators
+
+        keymap = mk_keymap_of_size(num_validators)
 
         genesis_state, genesis_block = create_mock_genesis(
             config=cls.config,

tests/components/eth2/beacon/test_validator.py

@@ -65,22 +65,32 @@ async def broadcast_attestation(self, attestation):
     async def broadcast_attestation_to_subnet(self, attestation, subnet_id):
         pass
 
+    async def broadcast_beacon_aggregate_and_proof(self, aggregate_and_proof):
+        pass
+
+
+async def get_validator(event_loop, event_bus, indices, num_validators=None) -> Validator:
+    if num_validators is not None:
+        chain = BeaconChainFactory(num_validators=num_validators)
+    else:
+        chain = BeaconChainFactory()
 
-async def get_validator(event_loop, event_bus, indices) -> Validator:
-    chain = BeaconChainFactory()
     validator_privkeys = {
         index: mk_key_pair_from_seed_index(index)[1]
         for index in indices
     }
 
+    # Mock attestation pool
+    attestatation_pool = set()
+
     def get_ready_attestations_fn(slot):
-        return ()
+        return tuple(attestatation_pool)
 
     def get_aggregatable_attestations_fn(slot, committee_index):
-        return ()
+        return tuple(attestatation_pool)
 
     def import_attestation_fn(attestation):
-        return ()
+        attestatation_pool.add(attestation)
 
     v = Validator(
         chain=chain,
@@ -352,7 +362,7 @@ async def test_validator_get_committee_assigment(event_loop, event_bus):
 
 
 @pytest.mark.asyncio
-async def test_validator_attest(event_loop, event_bus, monkeypatch):
+async def test_validator_attest(event_loop, event_bus):
     alice_indices = [i for i in range(NUM_VALIDATORS)]
     alice = await get_validator(event_loop=event_loop, event_bus=event_bus, indices=alice_indices)
     head = alice.chain.get_canonical_head()
@@ -382,6 +392,53 @@ async def test_validator_attest(event_loop, event_bus, monkeypatch):
     )
 
 
+@pytest.mark.asyncio
+async def test_validator_aggregate(event_loop, event_bus):
+    num_validators = 50
+    alice_indices = [i for i in range(num_validators)]
+    alice = await get_validator(
+        event_loop=event_loop,
+        event_bus=event_bus,
+        indices=alice_indices,
+        num_validators=num_validators,
+    )
+    alice.skip_block(
+        slot=alice.chain.get_canonical_head().slot + 100,
+        state=alice.chain.get_head_state(),
+        state_machine=alice.chain.get_state_machine(),
+    )
+    state_machine = alice.chain.get_state_machine()
+    state = alice.chain.get_head_state()
+    head = alice.chain.get_canonical_head()
+
+    epoch = compute_epoch_at_slot(state.slot, state_machine.config.SLOTS_PER_EPOCH)
+    assignment = alice._get_local_current_epoch_assignment(alice_indices[0], epoch)
+
+    attested_attsetation = await alice.attest(assignment.slot)
+    assert len(attested_attsetation) >= 1
+
+    aggregate_and_proofs = await alice.aggregate(assignment.slot)
+    assert len(aggregate_and_proofs) >= 1
+    for aggregate_and_proof in aggregate_and_proofs:
+        attestation = aggregate_and_proof.aggregate
+        assert attestation.data.slot == assignment.slot
+        assert attestation.data.beacon_block_root == head.signing_root
+        assert attestation.data.index == assignment.committee_index
+
+        # Advance the state and validate the attestation
+        config = state_machine.config
+        future_state = state_machine.state_transition.apply_state_transition(
+            state,
+            future_slot=assignment.slot + config.MIN_ATTESTATION_INCLUSION_DELAY,
+        )
+        validate_attestation(
+            future_state,
+            attestation,
+            config,
+        )
+        # break
+
+
 @pytest.mark.asyncio
 async def test_validator_include_ready_attestations(event_loop, event_bus, monkeypatch):
     # Alice controls all validators

trinity/components/eth2/beacon/validator.py

@@ -377,8 +377,8 @@ def _get_local_attesters_at_assignment(
         """
         for validator_index, (_, assignment) in self.local_validator_epoch_assignment.items():
             if (
-                assignment.slot == target_assignment.slot and
-                assignment.committee_index == target_assignment.committee_index
+                assignment.slot == target_assignment.slot
+                and assignment.committee_index == target_assignment.committee_index
             ):
                 yield validator_index
 
@@ -475,7 +475,7 @@ async def aggregate(
         config = state_machine.config
 
         attesting_committee_assignments_at_slot = self._get_attesting_assignments_at_slot(slot)
-
+        # 1. For each committee_assignment at the given slot
         for committee_assignment in attesting_committee_assignments_at_slot:
             committee_index = committee_assignment.committee_index
 
@@ -487,7 +487,9 @@ async def aggregate(
             }
 
             selected_proofs: Dict[ValidatorIndex, BLSSignature] = {}
+            # 2. For each attester
             for validator_index, privkey in attesting_validator_privkeys.items():
+                # Check if the vallidator is one of the aggregators
                 signature = slot_signature(
                     state, slot, privkey, CommitteeConfig(config),
                 )
@@ -498,24 +500,24 @@ async def aggregate(
                     signature,
                     CommitteeConfig(config),
                 )
-
                 if is_aggregator_result:
                     self.logger.debug(
                         f"validator ({validator_index}) is aggregator of"
                         f" committee_index={committee_index} at slot={slot}"
                     )
                     selected_proofs[validator_index] = signature
+                else:
+                    continue
 
-            for validator_index, selected_proof in selected_proofs.items():
                 aggregates = self._get_aggregates(slot, committee_index, config)
+                # 3. For each aggregate
+                # (it's possible with same CommitteeIndex and different AttesatationData)
                 for aggregate in aggregates:
                     aggregate_and_proof = AggregateAndProof(
                         index=validator_index,
                         aggregate=aggregate,
-                        selection_proof=selected_proof,
+                        selection_proof=selected_proofs[validator_index],
                     )
-
-                    # Import attestation to pool and then broadcast it
                     self.import_attestation(aggregate_and_proof.aggregate)
                     await self.p2p_node.broadcast_beacon_aggregate_and_proof(aggregate_and_proof)
                     aggregate_and_proofs += (aggregate_and_proof,)