gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-22 06:14:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Validator client refactor (#618)

Browse Source 
* Update to spec v0.9.0

* Update to v0.9.1

* Bump spec tags for v0.9.1

* Formatting, fix CI failures

* Resolve accidental KeyPair merge conflict

* Document new BeaconState functions

* Add `validator` changes from `validator-to-rest`

* Add initial (failing) REST api tests

* Fix signature parsing

* Add more tests

* Refactor http router

* Add working tests for publish beacon block

* Add validator duties tests

* Move account_manager under `lighthouse` binary

* Unify logfile handling in `environment` crate.

* Fix incorrect cache drops in `advance_caches`

* Update fork choice for v0.9.1

* Add `deposit_contract` crate

* Add progress on validator onboarding

* Add unfinished attesation code

* Update account manager CLI

* Write eth1 data file as hex string

* Integrate ValidatorDirectory with validator_client

* Move ValidatorDirectory into validator_client

* Clean up some FIXMEs

* Add beacon_chain_sim

* Fix a few docs/logs

* Expand `beacon_chain_sim`

* Fix spec for `beacon_chain_sim

* More testing for api

* Start work on attestation endpoint

* Reject empty attestations

* Allow attestations to genesis block

* Add working tests for `rest_api` validator endpoint

* Remove grpc from beacon_node

* Start heavy refactor of validator client

- Block production is working

* Prune old validator client files

* Start works on attestation service

* Add attestation service to validator client

* Use full pubkey for validator directories

* Add validator duties post endpoint

* Use par_iter for keypair generation

* Use bulk duties request in validator client

* Add version http endpoint tests

* Add interop keys and startup wait

* Ensure a prompt exit

* Add duties pruning

* Fix compile error in beacon node tests

* Add github workflow

* Modify rust.yaml

* Modify gitlab actions

* Add to CI file

* Add sudo to CI npm install

* Move cargo fmt to own job in tests

* Fix cargo fmt in CI

* Add rustup update before cargo fmt

* Change name of CI job

* Make other CI jobs require cargo fmt

* Add CI badge

* Remove gitlab and travis files

* Add different http timeout for debug

* Update docker file, use makefile in CI

* Use make in the dockerfile, skip the test

* Use the makefile for debug GI test

* Update book

* Tidy grpc and misc things

* Apply discv5 fixes

* Address other minor issues

* Fix warnings

* Attempt fix for addr parsing

* Tidy validator config, CLIs

* Tidy comments

* Tidy signing, reduce ForkService duplication

* Fail if skipping too many slots

* Set default recent genesis time to 0

* Add custom http timeout to validator

* Fix compile bug in node_test_rig

* Remove old bootstrap flag from val CLI

* Update docs

* Tidy val client

* Change val client log levels

* Add comments, more validity checks

* Fix compile error, add comments

* Undo changes to eth2-libp2p/src

* Reduce duplication of keypair generation

* Add more logging for validator duties

* Fix beacon_chain_sim, nitpicks

* Fix compile error, minor nits

* Address Michael's comments
This commit is contained in:
Paul Hauner
2019-11-25 15:48:24 +11:00
committed by GitHub
parent 3ca63cfa83
commit 78d82d9193
100 changed files with 4571 additions and 4032 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
validator_client/src/attestation_producer/beacon_node_attestation.rs
View File

@@ -1,23 +0,0 @@
//TODO: generalise these enums to the crate
use crate::block_producer::{BeaconNodeError, PublishOutcome};
use types::{Attestation, AttestationData, CommitteeIndex, EthSpec, Slot};
/// Defines the methods required to produce and publish attestations on a Beacon Node. Abstracts the
/// actual beacon node.
pub trait BeaconNodeAttestation: Send + Sync {
/// Request that the node produces the required attestation data.
///
fn produce_attestation_data(
&self,
slot: Slot,
index: CommitteeIndex,
) -> Result<AttestationData, BeaconNodeError>;
/// Request that the node publishes a attestation.
///
/// Returns `true` if the publish was successful.
fn publish_attestation<T: EthSpec>(
&self,
attestation: Attestation<T>,
) -> Result<PublishOutcome, BeaconNodeError>;
}

57
validator_client/src/attestation_producer/grpc.rs
View File

@@ -1,57 +0,0 @@
use super::beacon_node_attestation::BeaconNodeAttestation;
use crate::block_producer::{BeaconNodeError, PublishOutcome};
use protos::services_grpc::AttestationServiceClient;
use ssz::{Decode, Encode};
use protos::services::{
Attestation as GrpcAttestation, ProduceAttestationDataRequest, PublishAttestationRequest,
};
use types::{Attestation, AttestationData, CommitteeIndex, EthSpec, Slot};
impl BeaconNodeAttestation for AttestationServiceClient {
fn produce_attestation_data(
&self,
slot: Slot,
index: CommitteeIndex,
) -> Result<AttestationData, BeaconNodeError> {
let mut req = ProduceAttestationDataRequest::new();
req.set_slot(slot.as_u64());
req.set_shard(index);
let reply = self
.produce_attestation_data(&req)
.map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
let attestation_data =
AttestationData::from_ssz_bytes(reply.get_attestation_data().get_ssz())
.map_err(|_| BeaconNodeError::DecodeFailure)?;
Ok(attestation_data)
}
fn publish_attestation<T: EthSpec>(
&self,
attestation: Attestation<T>,
) -> Result<PublishOutcome, BeaconNodeError> {
let mut req = PublishAttestationRequest::new();
let ssz = attestation.as_ssz_bytes();
let mut grpc_attestation = GrpcAttestation::new();
grpc_attestation.set_ssz(ssz);
req.set_attestation(grpc_attestation);
let reply = self
.publish_attestation(&req)
.map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
if reply.get_success() {
Ok(PublishOutcome::Valid)
} else {
// TODO: distinguish between different errors
Ok(PublishOutcome::InvalidAttestation(
"Publish failed".to_string(),
))
}
}
}

166
validator_client/src/attestation_producer/mod.rs
View File

@@ -1,166 +0,0 @@
mod beacon_node_attestation;
mod grpc;
use std::sync::Arc;
use types::{ChainSpec, Domain, EthSpec, Fork};
//TODO: Move these higher up in the crate
use super::block_producer::{BeaconNodeError, PublishOutcome, ValidatorEvent};
use crate::signer::Signer;
use beacon_node_attestation::BeaconNodeAttestation;
use core::marker::PhantomData;
use slog::{error, info, warn};
use tree_hash::TreeHash;
use types::{AggregateSignature, Attestation, AttestationData, AttestationDuty, BitList};
//TODO: Group these errors at a crate level
#[derive(Debug, PartialEq)]
pub enum Error {
BeaconNodeError(BeaconNodeError),
}
impl From<BeaconNodeError> for Error {
fn from(e: BeaconNodeError) -> Error {
Error::BeaconNodeError(e)
}
}
/// This struct contains the logic for requesting and signing beacon attestations for a validator. The
/// validator can abstractly sign via the Signer trait object.
pub struct AttestationProducer<'a, B: BeaconNodeAttestation, S: Signer, E: EthSpec> {
/// The current fork.
pub fork: Fork,
/// The attestation duty to perform.
pub duty: AttestationDuty,
/// The current epoch.
pub spec: Arc<ChainSpec>,
/// The beacon node to connect to.
pub beacon_node: Arc<B>,
/// The signer to sign the block.
pub signer: &'a S,
/// Used for calculating epoch.
pub slots_per_epoch: u64,
/// Mere vessel for E.
pub _phantom: PhantomData<E>,
}
impl<'a, B: BeaconNodeAttestation, S: Signer, E: EthSpec> AttestationProducer<'a, B, S, E> {
/// Handle outputs and results from attestation production.
pub fn handle_produce_attestation(&mut self, log: slog::Logger) {
match self.produce_attestation() {
Ok(ValidatorEvent::AttestationProduced(slot)) => info!(
log,
"Attestation produced";
"validator" => format!("{}", self.signer),
"slot" => slot,
),
Err(e) => error!(log, "Attestation production error"; "Error" => format!("{:?}", e)),
Ok(ValidatorEvent::SignerRejection(_slot)) => {
error!(log, "Attestation production error"; "Error" => "Signer could not sign the attestation".to_string())
}
Ok(ValidatorEvent::IndexedAttestationNotProduced(_slot)) => {
error!(log, "Attestation production error"; "Error" => "Rejected the attestation as it could have been slashed".to_string())
}
Ok(ValidatorEvent::PublishAttestationFailed) => {
error!(log, "Attestation production error"; "Error" => "Beacon node was unable to publish an attestation".to_string())
}
Ok(ValidatorEvent::InvalidAttestation) => {
error!(log, "Attestation production error"; "Error" => "The signed attestation was invalid".to_string())
}
Ok(v) => {
warn!(log, "Unknown result for attestation production"; "Error" => format!("{:?}",v))
}
}
}
/// Produce an attestation, sign it and send it back
///
/// Assumes that an attestation is required at this slot (does not check the duties).
///
/// Ensures the message is not slashable.
///
/// !!! UNSAFE !!!
///
/// The slash-protection code is not yet implemented. There is zero protection against
/// slashing.
pub fn produce_attestation(&mut self) -> Result<ValidatorEvent, Error> {
let epoch = self.duty.slot.epoch(self.slots_per_epoch);
let attestation = self
.beacon_node
.produce_attestation_data(self.duty.slot, self.duty.index)?;
if self.safe_to_produce(&attestation) {
let domain = self
.spec
.get_domain(epoch, Domain::BeaconAttester, &self.fork);
if let Some(attestation) = self.sign_attestation(attestation, self.duty, domain) {
match self.beacon_node.publish_attestation(attestation) {
Ok(PublishOutcome::InvalidAttestation(_string)) => {
Ok(ValidatorEvent::InvalidAttestation)
}
Ok(PublishOutcome::Valid) => {
Ok(ValidatorEvent::AttestationProduced(self.duty.slot))
}
Err(_) | Ok(_) => Ok(ValidatorEvent::PublishAttestationFailed),
}
} else {
Ok(ValidatorEvent::SignerRejection(self.duty.slot))
}
} else {
Ok(ValidatorEvent::IndexedAttestationNotProduced(
self.duty.slot,
))
}
}
/// Consumes an attestation, returning the attestation signed by the validators private key.
///
/// Important: this function will not check to ensure the attestation is not slashable. This must be
/// done upstream.
fn sign_attestation(
&mut self,
attestation: AttestationData,
duties: AttestationDuty,
domain: u64,
) -> Option<Attestation<E>> {
self.store_produce(&attestation);
// build the aggregate signature
let aggregate_signature = {
let message = attestation.tree_hash_root();
let sig = self.signer.sign_message(&message, domain)?;
let mut agg_sig = AggregateSignature::new();
agg_sig.add(&sig);
agg_sig
};
let mut aggregation_bits = BitList::with_capacity(duties.committee_len).ok()?;
aggregation_bits.set(duties.committee_position, true).ok()?;
Some(Attestation {
aggregation_bits,
data: attestation,
signature: aggregate_signature,
})
}
/// Returns `true` if signing an attestation is safe (non-slashable).
///
/// !!! UNSAFE !!!
///
/// Important: this function is presently stubbed-out. It provides ZERO SAFETY.
fn safe_to_produce(&self, _attestation: &AttestationData) -> bool {
//TODO: Implement slash protection
true
}
/// Record that an attestation was produced so that slashable votes may not be made in the future.
///
/// !!! UNSAFE !!!
///
/// Important: this function is presently stubbed-out. It provides ZERO SAFETY.
fn store_produce(&mut self, _attestation: &AttestationData) {
// TODO: Implement slash protection
}
}

314
validator_client/src/attestation_service.rs Normal file
View File

@@ -0,0 +1,314 @@
use crate::{duties_service::DutiesService, validator_store::ValidatorStore};
use environment::RuntimeContext;
use exit_future::Signal;
use futures::{Future, Stream};
use remote_beacon_node::{PublishStatus, RemoteBeaconNode, ValidatorDuty};
use slog::{crit, info, trace};
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{ChainSpec, CommitteeIndex, EthSpec, Slot};
/// Builds an `AttestationService`.
pub struct AttestationServiceBuilder<T, E: EthSpec> {
duties_service: Option<DutiesService<T, E>>,
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<T>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
impl<T: SlotClock + 'static, E: EthSpec> AttestationServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
duties_service: None,
validator_store: None,
slot_clock: None,
beacon_node: None,
context: None,
}
}
pub fn duties_service(mut self, service: DutiesService<T, E>) -> Self {
self.duties_service = Some(service);
self
}
pub fn validator_store(mut self, store: ValidatorStore<T, E>) -> Self {
self.validator_store = Some(store);
self
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(slot_clock);
self
}
pub fn beacon_node(mut self, beacon_node: RemoteBeaconNode<E>) -> Self {
self.beacon_node = Some(beacon_node);
self
}
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
pub fn build(self) -> Result<AttestationService<T, E>, String> {
Ok(AttestationService {
inner: Arc::new(Inner {
duties_service: self
.duties_service
.ok_or_else(|| "Cannot build AttestationService without duties_service")?,
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build AttestationService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build AttestationService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build AttestationService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build AttestationService without runtime_context")?,
}),
})
}
}
/// Helper to minimise `Arc` usage.
pub struct Inner<T, E: EthSpec> {
duties_service: DutiesService<T, E>,
validator_store: ValidatorStore<T, E>,
slot_clock: T,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
}
/// Attempts to produce attestations for all known validators 1/3rd of the way through each slot.
///
/// If any validators are on the same committee, a single attestation will be downloaded and
/// returned to the beacon node. This attestation will have a signature from each of the
/// validators.
pub struct AttestationService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for AttestationService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for AttestationService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
/// Starts the service which periodically produces attestations.
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let context = &self.context;
let log = context.log.clone();
let duration_to_next_slot = self
.slot_clock
.duration_to_next_slot()
.ok_or_else(|| "Unable to determine duration to next slot".to_string())?;
let interval = {
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
Interval::new(
Instant::now() + duration_to_next_slot + slot_duration / 3,
slot_duration,
)
};
let (exit_signal, exit_fut) = exit_future::signal();
let service = self.clone();
let log_1 = log.clone();
let log_2 = log.clone();
let log_3 = log.clone();
context.executor.spawn(
exit_fut
.until(
interval
.map_err(move |e| {
crit! {
log_1,
"Timer thread failed";
"error" => format!("{}", e)
}
})
.for_each(move |_| {
if let Err(e) = service.spawn_attestation_tasks() {
crit!(
log_2,
"Failed to spawn attestation tasks";
"error" => e
)
} else {
trace!(
log_2,
"Spawned attestation tasks";
)
}
Ok(())
})
// Prevent any errors from escaping and stopping the interval.
.then(|_| Ok(())),
)
.map(move |_| info!(log_3, "Shutdown complete")),
);
Ok(exit_signal)
}
/// For each each required attestation, spawn a new task that downloads, signs and uploads the
/// attestation to the beacon node.
fn spawn_attestation_tasks(&self) -> Result<(), String> {
let service = self.clone();
let slot = service
.slot_clock
.now()
.ok_or_else(|| "Failed to read slot clock".to_string())?;
let mut committee_indices: HashMap<CommitteeIndex, Vec<ValidatorDuty>> = HashMap::new();
service
.duties_service
.attesters(slot)
.into_iter()
.for_each(|duty| {
if let Some(committee_index) = duty.attestation_committee_index {
let validator_duties =
committee_indices.entry(committee_index).or_insert(vec![]);
validator_duties.push(duty);
}
});
committee_indices
.into_iter()
.for_each(|(committee_index, validator_duties)| {
// Spawn a separate task for each attestation.
service.context.executor.spawn(self.clone().do_attestation(
slot,
committee_index,
validator_duties,
));
});
Ok(())
}
/// For a given `committee_index`, download the attestation, have it signed by all validators
/// in `validator_duties` then upload it.
fn do_attestation(
&self,
slot: Slot,
committee_index: CommitteeIndex,
validator_duties: Vec<ValidatorDuty>,
) -> impl Future<Item = (), Error = ()> {
let service_1 = self.clone();
let service_2 = self.clone();
let log_1 = self.context.log.clone();
let log_2 = self.context.log.clone();
self.beacon_node
.http
.validator()
.produce_attestation(slot, committee_index)
.map_err(|e| format!("Failed to produce attestation: {:?}", e))
.map(move |attestation| {
validator_duties
.iter()
.fold(attestation, |mut attestation, duty| {
let log = service_1.context.log.clone();
if let Some((
duty_slot,
duty_committee_index,
validator_committee_position,
)) = attestation_duties(duty)
{
if duty_slot == slot && duty_committee_index == committee_index {
if service_1
.validator_store
.sign_attestation(
&duty.validator_pubkey,
validator_committee_position,
&mut attestation,
)
.is_none()
{
crit!(log, "Failed to sign attestation");
}
} else {
crit!(log, "Inconsistent validator duties during signing");
}
} else {
crit!(log, "Missing validator duties when signing");
}
attestation
})
})
.and_then(move |attestation| {
service_2
.beacon_node
.http
.validator()
.publish_attestation(attestation.clone())
.map(|publish_status| (attestation, publish_status))
.map_err(|e| format!("Failed to publish attestation: {:?}", e))
})
.map(move |(attestation, publish_status)| match publish_status {
PublishStatus::Valid => info!(
log_1,
"Successfully published attestation";
"signatures" => attestation.aggregation_bits.num_set_bits(),
"head_block" => format!("{}", attestation.data.beacon_block_root),
"committee_index" => attestation.data.index,
"slot" => attestation.data.slot.as_u64(),
),
PublishStatus::Invalid(msg) => crit!(
log_1,
"Published attestation was invalid";
"message" => msg,
"committee_index" => attestation.data.index,
"slot" => attestation.data.slot.as_u64(),
),
PublishStatus::Unknown => {
crit!(log_1, "Unknown condition when publishing attestation")
}
})
.map_err(move |e| {
crit!(
log_2,
"Error during attestation production";
"error" => e
)
})
}
}
fn attestation_duties(duty: &ValidatorDuty) -> Option<(Slot, CommitteeIndex, usize)> {
Some((
duty.attestation_slot?,
duty.attestation_committee_index?,
duty.attestation_committee_position?,
))
}

34
validator_client/src/block_producer/beacon_node_block.rs
View File

@@ -1,34 +0,0 @@
use types::{BeaconBlock, EthSpec, Signature, Slot};
#[derive(Debug, PartialEq, Clone)]
pub enum BeaconNodeError {
RemoteFailure(String),
DecodeFailure,
}
#[derive(Debug, PartialEq, Clone)]
pub enum PublishOutcome {
Valid,
InvalidBlock(String),
InvalidAttestation(String),
}
/// Defines the methods required to produce and publish blocks on a Beacon Node. Abstracts the
/// actual beacon node.
pub trait BeaconNodeBlock: Send + Sync {
/// Request that the node produces a block.
///
/// Returns Ok(None) if the Beacon Node is unable to produce at the given slot.
fn produce_beacon_block<T: EthSpec>(
&self,
slot: Slot,
randao_reveal: &Signature,
) -> Result<Option<BeaconBlock<T>>, BeaconNodeError>;
/// Request that the node publishes a block.
///
/// Returns `true` if the publish was successful.
fn publish_beacon_block<T: EthSpec>(
&self,
block: BeaconBlock<T>,
) -> Result<PublishOutcome, BeaconNodeError>;
}

87
validator_client/src/block_producer/grpc.rs
View File

@@ -1,87 +0,0 @@
use super::beacon_node_block::*;
use protos::services::{
BeaconBlock as GrpcBeaconBlock, ProduceBeaconBlockRequest, PublishBeaconBlockRequest,
};
use protos::services_grpc::BeaconBlockServiceClient;
use ssz::{Decode, Encode};
use std::sync::Arc;
use types::{BeaconBlock, EthSpec, Signature, Slot};
//TODO: Remove this new type. Do not need to wrap
/// A newtype designed to wrap the gRPC-generated service so the `BeaconNode` trait may be
/// implemented upon it.
pub struct BeaconBlockGrpcClient {
client: Arc<BeaconBlockServiceClient>,
}
impl BeaconBlockGrpcClient {
pub fn new(client: Arc<BeaconBlockServiceClient>) -> Self {
Self { client }
}
}
impl BeaconNodeBlock for BeaconBlockGrpcClient {
/// Request a Beacon Node (BN) to produce a new block at the supplied slot.
///
/// Returns `None` if it is not possible to produce at the supplied slot. For example, if the
/// BN is unable to find a parent block.
fn produce_beacon_block<T: EthSpec>(
&self,
slot: Slot,
randao_reveal: &Signature,
) -> Result<Option<BeaconBlock<T>>, BeaconNodeError> {
// request a beacon block from the node
let mut req = ProduceBeaconBlockRequest::new();
req.set_slot(slot.as_u64());
req.set_randao_reveal(randao_reveal.as_ssz_bytes());
//TODO: Determine if we want an explicit timeout
let reply = self
.client
.produce_beacon_block(&req)
.map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
// format the reply
if reply.has_block() {
let block = reply.get_block();
let ssz = block.get_ssz();
let block =
BeaconBlock::from_ssz_bytes(&ssz).map_err(|_| BeaconNodeError::DecodeFailure)?;
Ok(Some(block))
} else {
Ok(None)
}
}
/// Request a Beacon Node (BN) to publish a block.
///
/// Generally, this will be called after a `produce_beacon_block` call with a block that has
/// been completed (signed) by the validator client.
fn publish_beacon_block<T: EthSpec>(
&self,
block: BeaconBlock<T>,
) -> Result<PublishOutcome, BeaconNodeError> {
let mut req = PublishBeaconBlockRequest::new();
let ssz = block.as_ssz_bytes();
let mut grpc_block = GrpcBeaconBlock::new();
grpc_block.set_ssz(ssz);
req.set_block(grpc_block);
let reply = self
.client
.publish_beacon_block(&req)
.map_err(|err| BeaconNodeError::RemoteFailure(format!("{:?}", err)))?;
if reply.get_success() {
Ok(PublishOutcome::Valid)
} else {
// TODO: distinguish between different errors
Ok(PublishOutcome::InvalidBlock("Publish failed".to_string()))
}
}
}

259
validator_client/src/block_producer/mod.rs
View File

@@ -1,259 +0,0 @@
mod beacon_node_block;
mod grpc;
use self::beacon_node_block::BeaconNodeBlock;
pub use self::beacon_node_block::{BeaconNodeError, PublishOutcome};
pub use self::grpc::BeaconBlockGrpcClient;
use crate::signer::Signer;
use core::marker::PhantomData;
use slog::{error, info, trace, warn};
use std::sync::Arc;
use tree_hash::{SignedRoot, TreeHash};
use types::{BeaconBlock, ChainSpec, Domain, EthSpec, Fork, Slot};
#[derive(Debug, PartialEq)]
pub enum Error {
BeaconNodeError(BeaconNodeError),
}
#[derive(Debug, PartialEq)]
pub enum ValidatorEvent {
/// A new block was produced.
BlockProduced(Slot),
/// A new attestation was produced.
AttestationProduced(Slot),
/// A block was not produced as it would have been slashable.
SlashableBlockNotProduced(Slot),
/// An attestation was not produced as it would have been slashable.
IndexedAttestationNotProduced(Slot),
/// The Beacon Node was unable to produce a block at that slot.
BeaconNodeUnableToProduceBlock(Slot),
/// The signer failed to sign the message.
SignerRejection(Slot),
/// Publishing an attestation failed.
PublishAttestationFailed,
/// Beacon node rejected the attestation.
InvalidAttestation,
}
/// This struct contains the logic for requesting and signing beacon blocks for a validator. The
/// validator can abstractly sign via the Signer trait object.
pub struct BlockProducer<'a, B: BeaconNodeBlock, S: Signer, E: EthSpec> {
/// The current fork.
pub fork: Fork,
/// The current slot to produce a block for.
pub slot: Slot,
/// The current epoch.
pub spec: Arc<ChainSpec>,
/// The beacon node to connect to.
pub beacon_node: Arc<B>,
/// The signer to sign the block.
pub signer: &'a S,
/// Used for calculating epoch.
pub slots_per_epoch: u64,
/// Mere vessel for E.
pub _phantom: PhantomData<E>,
/// The logger, for logging
pub log: slog::Logger,
}
impl<'a, B: BeaconNodeBlock, S: Signer, E: EthSpec> BlockProducer<'a, B, S, E> {
/// Handle outputs and results from block production.
pub fn handle_produce_block(&mut self) {
match self.produce_block() {
Ok(ValidatorEvent::BlockProduced(slot)) => info!(
self.log,
"Block produced";
"validator" => format!("{}", self.signer),
"slot" => slot,
),
Err(e) => error!(self.log, "Block production error"; "Error" => format!("{:?}", e)),
Ok(ValidatorEvent::SignerRejection(_slot)) => {
error!(self.log, "Block production error"; "Error" => "Signer Could not sign the block".to_string())
}
Ok(ValidatorEvent::SlashableBlockNotProduced(_slot)) => {
error!(self.log, "Block production error"; "Error" => "Rejected the block as it could have been slashed".to_string())
}
Ok(ValidatorEvent::BeaconNodeUnableToProduceBlock(_slot)) => {
error!(self.log, "Block production error"; "Error" => "Beacon node was unable to produce a block".to_string())
}
Ok(v) => {
warn!(self.log, "Unknown result for block production"; "Error" => format!("{:?}",v))
}
}
}
/// Produce a block at some slot.
///
/// Assumes that a block is required at this slot (does not check the duties).
///
/// Ensures the message is not slashable.
///
/// !!! UNSAFE !!!
///
/// The slash-protection code is not yet implemented. There is zero protection against
/// slashing.
pub fn produce_block(&mut self) -> Result<ValidatorEvent, Error> {
let epoch = self.slot.epoch(self.slots_per_epoch);
trace!(self.log, "Producing block"; "epoch" => epoch);
let message = epoch.tree_hash_root();
let randao_reveal = match self.signer.sign_message(
&message,
self.spec.get_domain(epoch, Domain::Randao, &self.fork),
) {
None => {
warn!(self.log, "Signing rejected"; "message" => format!("{:?}", message));
return Ok(ValidatorEvent::SignerRejection(self.slot));
}
Some(signature) => signature,
};
if let Some(block) = self
.beacon_node
.produce_beacon_block(self.slot, &randao_reveal)?
{
if self.safe_to_produce(&block) {
let slot = block.slot;
let domain = self
.spec
.get_domain(epoch, Domain::BeaconProposer, &self.fork);
if let Some(block) = self.sign_block(block, domain) {
self.beacon_node.publish_beacon_block(block)?;
Ok(ValidatorEvent::BlockProduced(slot))
} else {
Ok(ValidatorEvent::SignerRejection(self.slot))
}
} else {
Ok(ValidatorEvent::SlashableBlockNotProduced(self.slot))
}
} else {
Ok(ValidatorEvent::BeaconNodeUnableToProduceBlock(self.slot))
}
}
/// Consumes a block, returning that block signed by the validators private key.
///
/// Important: this function will not check to ensure the block is not slashable. This must be
/// done upstream.
fn sign_block(&mut self, mut block: BeaconBlock<E>, domain: u64) -> Option<BeaconBlock<E>> {
self.store_produce(&block);
match self.signer.sign_message(&block.signed_root()[..], domain) {
None => None,
Some(signature) => {
block.signature = signature;
Some(block)
}
}
}
/// Returns `true` if signing a block is safe (non-slashable).
///
/// !!! UNSAFE !!!
///
/// Important: this function is presently stubbed-out. It provides ZERO SAFETY.
fn safe_to_produce(&self, _block: &BeaconBlock<E>) -> bool {
// TODO: ensure the producer doesn't produce slashable blocks.
// https://github.com/sigp/lighthouse/issues/160
true
}
/// Record that a block was produced so that slashable votes may not be made in the future.
///
/// !!! UNSAFE !!!
///
/// Important: this function is presently stubbed-out. It provides ZERO SAFETY.
fn store_produce(&mut self, _block: &BeaconBlock<E>) {
// TODO: record this block production to prevent future slashings.
// https://github.com/sigp/lighthouse/issues/160
}
}
impl From<BeaconNodeError> for Error {
fn from(e: BeaconNodeError) -> Error {
Error::BeaconNodeError(e)
}
}
/* Old tests - Re-work for new logic
#[cfg(test)]
mod tests {
use super::test_utils::{EpochMap, LocalSigner, SimulatedBeaconNode};
use super::*;
use slot_clock::TestingSlotClock;
use types::{
test_utils::{SeedableRng, TestRandom, XorShiftRng},
Keypair,
};
// TODO: implement more thorough testing.
// https://github.com/sigp/lighthouse/issues/160
//
// These tests should serve as a good example for future tests.
#[test]
pub fn polling() {
let mut rng = XorShiftRng::from_seed([42; 16]);
let spec = Arc::new(ChainSpec::mainnet());
let slot_clock = Arc::new(TestingSlotClock::new(0));
let beacon_node = Arc::new(SimulatedBeaconNode::default());
let signer = Arc::new(LocalSigner::new(Keypair::random()));
let mut epoch_map = EpochMap::new(T::slots_per_epoch());
let produce_slot = Slot::new(100);
let produce_epoch = produce_slot.epoch(T::slots_per_epoch());
epoch_map.map.insert(produce_epoch, produce_slot);
let epoch_map = Arc::new(epoch_map);
let mut block_proposer = BlockProducer::new(
spec.clone(),
epoch_map.clone(),
slot_clock.clone(),
beacon_node.clone(),
signer.clone(),
);
// Configure responses from the BeaconNode.
beacon_node.set_next_produce_result(Ok(Some(BeaconBlock::random_for_test(&mut rng))));
beacon_node.set_next_publish_result(Ok(PublishOutcome::ValidBlock));
// One slot before production slot...
slot_clock.set_slot(produce_slot.as_u64() - 1);
assert_eq!(
block_proposer.poll(),
Ok(PollOutcome::BlockProductionNotRequired(produce_slot - 1))
);
// On the produce slot...
slot_clock.set_slot(produce_slot.as_u64());
assert_eq!(
block_proposer.poll(),
Ok(PollOutcome::BlockProduced(produce_slot.into()))
);
// Trying the same produce slot again...
slot_clock.set_slot(produce_slot.as_u64());
assert_eq!(
block_proposer.poll(),
Ok(PollOutcome::SlotAlreadyProcessed(produce_slot))
);
// One slot after the produce slot...
slot_clock.set_slot(produce_slot.as_u64() + 1);
assert_eq!(
block_proposer.poll(),
Ok(PollOutcome::BlockProductionNotRequired(produce_slot + 1))
);
// In an epoch without known duties...
let slot = (produce_epoch.as_u64() + 1) * T::slots_per_epoch();
slot_clock.set_slot(slot);
assert_eq!(
block_proposer.poll(),
Ok(PollOutcome::ProducerDutiesUnknown(Slot::new(slot)))
);
}
}
*/

268
validator_client/src/block_service.rs Normal file
View File

@@ -0,0 +1,268 @@
use crate::{duties_service::DutiesService, validator_store::ValidatorStore};
use environment::RuntimeContext;
use exit_future::Signal;
use futures::{stream, Future, IntoFuture, Stream};
use remote_beacon_node::{PublishStatus, RemoteBeaconNode};
use slog::{crit, error, info, trace};
use slot_clock::SlotClock;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{ChainSpec, EthSpec};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
/// Builds a `BlockService`.
pub struct BlockServiceBuilder<T, E: EthSpec> {
duties_service: Option<DutiesService<T, E>>,
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<Arc<T>>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
duties_service: None,
validator_store: None,
slot_clock: None,
beacon_node: None,
context: None,
}
}
pub fn duties_service(mut self, service: DutiesService<T, E>) -> Self {
self.duties_service = Some(service);
self
}
pub fn validator_store(mut self, store: ValidatorStore<T, E>) -> Self {
self.validator_store = Some(store);
self
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(Arc::new(slot_clock));
self
}
pub fn beacon_node(mut self, beacon_node: RemoteBeaconNode<E>) -> Self {
self.beacon_node = Some(beacon_node);
self
}
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
pub fn build(self) -> Result<BlockService<T, E>, String> {
Ok(BlockService {
inner: Arc::new(Inner {
duties_service: self
.duties_service
.ok_or_else(|| "Cannot build BlockService without duties_service")?,
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build BlockService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build BlockService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build BlockService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build BlockService without runtime_context")?,
}),
})
}
}
/// Helper to minimise `Arc` usage.
pub struct Inner<T, E: EthSpec> {
duties_service: DutiesService<T, E>,
validator_store: ValidatorStore<T, E>,
slot_clock: Arc<T>,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
}
/// Attempts to produce attestations for any block producer(s) at the start of the epoch.
pub struct BlockService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for BlockService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for BlockService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
/// Starts the service that periodically attempts to produce blocks.
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let log = self.context.log.clone();
let duration_to_next_slot = self
.slot_clock
.duration_to_next_slot()
.ok_or_else(|| "Unable to determine duration to next slot".to_string())?;
let interval = {
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
Interval::new(
Instant::now() + duration_to_next_slot + TIME_DELAY_FROM_SLOT,
slot_duration,
)
};
let (exit_signal, exit_fut) = exit_future::signal();
let service = self.clone();
let log_1 = log.clone();
let log_2 = log.clone();
self.context.executor.spawn(
exit_fut
.until(
interval
.map_err(move |e| {
crit! {
log_1,
"Timer thread failed";
"error" => format!("{}", e)
}
})
.for_each(move |_| service.clone().do_update())
// Prevent any errors from escaping and stopping the interval.
.then(|_| Ok(())),
)
.map(move |_| info!(log_2, "Shutdown complete")),
);
Ok(exit_signal)
}
/// Attempt to produce a block for any block producers in the `ValidatorStore`.
fn do_update(self) -> impl Future<Item = (), Error = ()> {
let service = self.clone();
let log_1 = self.context.log.clone();
let log_2 = self.context.log.clone();
self.slot_clock
.now()
.ok_or_else(move || {
crit!(log_1, "Duties manager failed to read slot clock");
})
.into_future()
.and_then(move |slot| {
let iter = service.duties_service.block_producers(slot).into_iter();
if iter.len() == 0 {
trace!(
log_2,
"No local block proposers for this slot";
"slot" => slot.as_u64()
)
} else if iter.len() > 1 {
error!(
log_2,
"Multiple block proposers for this slot";
"action" => "producing blocks for all proposers",
"num_proposers" => iter.len(),
"slot" => slot.as_u64(),
)
}
stream::unfold(iter, move |mut block_producers| {
let log_1 = service.context.log.clone();
let log_2 = service.context.log.clone();
let service_1 = service.clone();
let service_2 = service.clone();
let service_3 = service.clone();
block_producers.next().map(move |validator_pubkey| {
service_1
.validator_store
.randao_reveal(&validator_pubkey, slot.epoch(E::slots_per_epoch()))
.ok_or_else(|| "Unable to produce randao reveal".to_string())
.into_future()
.and_then(move |randao_reveal| {
service_1
.beacon_node
.http
.validator()
.produce_block(slot, randao_reveal)
.map_err(|e| {
format!(
"Error from beacon node when producing block: {:?}",
e
)
})
})
.and_then(move |block| {
service_2
.validator_store
.sign_block(&validator_pubkey, block)
.ok_or_else(|| "Unable to sign block".to_string())
})
.and_then(move |block| {
service_3
.beacon_node
.http
.validator()
.publish_block(block.clone())
.map(|publish_status| (block, publish_status))
.map_err(|e| {
format!(
"Error from beacon node when publishing block: {:?}",
e
)
})
})
.map(move |(block, publish_status)| match publish_status {
PublishStatus::Valid => info!(
log_1,
"Successfully published block";
"deposits" => block.body.deposits.len(),
"attestations" => block.body.attestations.len(),
"slot" => block.slot.as_u64(),
),
PublishStatus::Invalid(msg) => crit!(
log_1,
"Published block was invalid";
"message" => msg,
"slot" => block.slot.as_u64(),
),
PublishStatus::Unknown => {
crit!(log_1, "Unknown condition when publishing block")
}
})
.map_err(move |e| {
crit!(
log_2,
"Error whilst producing block";
"message" => e
)
})
.then(|_| Ok(((), block_producers)))
})
})
.collect()
.map(|_| ())
})
}
}

50
validator_client/src/cli.rs
View File

@@ -1,44 +1,16 @@
use crate::config::{DEFAULT_SERVER, DEFAULT_SERVER_GRPC_PORT, DEFAULT_SERVER_HTTP_PORT};
use crate::config::DEFAULT_HTTP_SERVER;
use clap::{App, Arg, SubCommand};
pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
App::new("Validator Client")
.visible_aliases(&["v", "vc", "validator", "validator_client"])
.version("0.0.1")
.author("Sigma Prime <contact@sigmaprime.io>")
.about("Eth 2.0 Validator Client")
.arg(
Arg::with_name("datadir")
.long("datadir")
.short("d")
.value_name("DIR")
.help("Data directory for keys and databases.")
.takes_value(true),
)
App::new("validator_client")
.visible_aliases(&["v", "vc", "validator"])
.about("Ethereum 2.0 Validator Client")
.arg(
Arg::with_name("server")
.long("server")
.value_name("NETWORK_ADDRESS")
.help("Address to connect to BeaconNode.")
.default_value(DEFAULT_SERVER)
.takes_value(true),
)
.arg(
Arg::with_name("server-grpc-port")
.long("server-grpc-port")
.short("g")
.value_name("PORT")
.help("Port to use for gRPC API connection to the server.")
.default_value(DEFAULT_SERVER_GRPC_PORT)
.takes_value(true),
)
.arg(
Arg::with_name("server-http-port")
.long("server-http-port")
.short("h")
.value_name("PORT")
.help("Port to use for HTTP API connection to the server.")
.default_value(DEFAULT_SERVER_HTTP_PORT)
.default_value(&DEFAULT_HTTP_SERVER)
.takes_value(true),
)
/*
@@ -49,12 +21,6 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.subcommand(SubCommand::with_name("testnet")
.about("Starts a testnet validator using INSECURE, predicatable private keys, based off the canonical \
validator index. ONLY USE FOR TESTING PURPOSES!")
.arg(
Arg::with_name("bootstrap")
.short("b")
.long("bootstrap")
.help("Connect to the RPC server to download the eth2_config via the HTTP API.")
)
.subcommand(SubCommand::with_name("insecure")
.about("Uses the standard, predicatable `interop` keygen method to produce a range \
of predicatable private keys and starts performing their validator duties.")
@@ -62,10 +28,10 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.value_name("VALIDATOR_INDEX")
.required(true)
.help("The first validator public key to be generated for this client."))
.arg(Arg::with_name("validator_count")
.value_name("COUNT")
.arg(Arg::with_name("last_validator")
.value_name("VALIDATOR_INDEX")
.required(true)
.help("The number of validators."))
.help("The end of the range of keys to generate. This index is not generated."))
)
.subcommand(SubCommand::with_name("interop-yaml")
.about("Loads plain-text secret keys from YAML files. Expects the interop format defined

254
validator_client/src/config.rs
View File

@@ -1,29 +1,16 @@
use bincode;
use bls::Keypair;
use clap::ArgMatches;
use serde_derive::{Deserialize, Serialize};
use slog::{error, warn};
use std::fs::{self, File};
use std::io::{Error, ErrorKind};
use std::ops::Range;
use std::path::PathBuf;
use types::{
test_utils::{generate_deterministic_keypair, load_keypairs_from_yaml},
EthSpec, MainnetEthSpec,
};
pub const DEFAULT_SERVER: &str = "localhost";
pub const DEFAULT_SERVER_GRPC_PORT: &str = "5051";
pub const DEFAULT_SERVER_HTTP_PORT: &str = "5052";
pub const DEFAULT_HTTP_SERVER: &str = "http://localhost:5052/";
/// Specifies a method for obtaining validator keypairs.
#[derive(Clone)]
pub enum KeySource {
/// Load the keypairs from disk.
Disk,
/// Generate the keypairs (insecure, generates predictable keys).
TestingKeypairRange(Range<usize>),
/// Load testing keypairs from YAML
YamlKeypairs(PathBuf),
InsecureKeypairs(Vec<usize>),
}
impl Default for KeySource {
@@ -37,205 +24,78 @@ impl Default for KeySource {
pub struct Config {
/// The data directory, which stores all validator databases
pub data_dir: PathBuf,
/// The source for loading keypairs
/// Specifies how the validator client should load keypairs.
#[serde(skip)]
pub key_source: KeySource,
/// The path where the logs will be outputted
pub log_file: PathBuf,
/// The server at which the Beacon Node can be contacted
pub server: String,
/// The gRPC port on the server
pub server_grpc_port: u16,
/// The HTTP port on the server, for the REST API.
pub server_http_port: u16,
/// The number of slots per epoch.
pub slots_per_epoch: u64,
/// The http endpoint of the beacon node API.
///
/// Should be similar to `http://localhost:8080`
pub http_server: String,
}
const DEFAULT_PRIVATE_KEY_FILENAME: &str = "private.key";
impl Default for Config {
/// Build a new configuration from defaults.
fn default() -> Self {
Self {
data_dir: PathBuf::from(".lighthouse-validator"),
data_dir: PathBuf::from(".lighthouse/validators"),
key_source: <_>::default(),
log_file: PathBuf::from(""),
server: DEFAULT_SERVER.into(),
server_grpc_port: DEFAULT_SERVER_GRPC_PORT
.parse::<u16>()
.expect("gRPC port constant should be valid"),
server_http_port: DEFAULT_SERVER_GRPC_PORT
.parse::<u16>()
.expect("HTTP port constant should be valid"),
slots_per_epoch: MainnetEthSpec::slots_per_epoch(),
http_server: DEFAULT_HTTP_SERVER.to_string(),
}
}
}
impl Config {
/// Returns the full path for the client data directory (not just the name of the directory).
pub fn full_data_dir(&self) -> Option<PathBuf> {
dirs::home_dir().map(|path| path.join(&self.data_dir))
}
/// Returns a `Default` implementation of `Self` with some parameters modified by the supplied
/// `cli_args`.
pub fn from_cli(cli_args: &ArgMatches) -> Result<Config, String> {
let mut config = Config::default();
/// Creates the data directory (and any non-existing parent directories).
pub fn create_data_dir(&self) -> Option<PathBuf> {
let path = dirs::home_dir()?.join(&self.data_dir);
fs::create_dir_all(&path).ok()?;
Some(path)
}
/// Apply the following arguments to `self`, replacing values if they are specified in `args`.
///
/// Returns an error if arguments are obviously invalid. May succeed even if some values are
/// invalid.
pub fn apply_cli_args(
&mut self,
args: &ArgMatches,
_log: &slog::Logger,
) -> Result<(), &'static str> {
if let Some(datadir) = args.value_of("datadir") {
self.data_dir = PathBuf::from(datadir);
};
if let Some(srv) = args.value_of("server") {
self.server = srv.to_string();
};
Ok(())
}
/// Reads a single keypair from the given `path`.
///
/// `path` should be the path to a directory containing a private key. The file name of `path`
/// must align with the public key loaded from it, otherwise an error is returned.
///
/// An error will be returned if `path` is a file (not a directory).
fn read_keypair_file(&self, path: PathBuf) -> Result<Keypair, String> {
if !path.is_dir() {
return Err("Is not a directory".into());
if let Some(server) = cli_args.value_of("server") {
config.http_server = server.to_string();
}
let key_filename: PathBuf = path.join(DEFAULT_PRIVATE_KEY_FILENAME);
if !key_filename.is_file() {
return Err(format!(
"Private key is not a file: {:?}",
key_filename.to_str()
));
}
let mut key_file = File::open(key_filename.clone())
.map_err(|e| format!("Unable to open private key file: {}", e))?;
let key: Keypair = bincode::deserialize_from(&mut key_file)
.map_err(|e| format!("Unable to deserialize private key: {:?}", e))?;
let ki = key.identifier();
if ki
!= path
.file_name()
.ok_or_else(|| "Invalid path".to_string())?
.to_string_lossy()
{
Err(format!(
"The validator key ({:?}) did not match the directory filename {:?}.",
ki,
path.to_str()
))
} else {
Ok(key)
}
}
pub fn fetch_keys_from_disk(&self, log: &slog::Logger) -> Result<Vec<Keypair>, String> {
Ok(
fs::read_dir(&self.full_data_dir().expect("Data dir must exist"))
.map_err(|e| format!("Failed to read datadir: {:?}", e))?
.filter_map(|validator_dir| {
let path = validator_dir.ok()?.path();
if path.is_dir() {
match self.read_keypair_file(path.clone()) {
Ok(keypair) => Some(keypair),
Err(e) => {
error!(
log,
"Failed to parse a validator keypair";
"error" => e,
"path" => path.to_str(),
);
None
}
}
} else {
None
}
})
.collect(),
)
}
pub fn fetch_testing_keypairs(
&self,
range: std::ops::Range<usize>,
) -> Result<Vec<Keypair>, String> {
Ok(range.map(generate_deterministic_keypair).collect())
}
/// Loads the keypairs according to `self.key_source`. Will return one or more keypairs, or an
/// error.
#[allow(dead_code)]
pub fn fetch_keys(&self, log: &slog::Logger) -> Result<Vec<Keypair>, String> {
let keypairs = match &self.key_source {
KeySource::Disk => self.fetch_keys_from_disk(log)?,
KeySource::TestingKeypairRange(range) => {
warn!(
log,
"Using insecure interop private keys";
"range" => format!("{:?}", range)
);
self.fetch_testing_keypairs(range.clone())?
let config = match cli_args.subcommand() {
("testnet", Some(sub_cli_args)) => {
if cli_args.is_present("eth2-config") && sub_cli_args.is_present("bootstrap") {
return Err(
"Cannot specify --eth2-config and --bootstrap as it may result \
in ambiguity."
.into(),
);
}
process_testnet_subcommand(sub_cli_args, config)
}
KeySource::YamlKeypairs(path) => {
warn!(
log,
"Private keys are stored insecurely (plain text). Testing use only."
);
_ => return Err("You must use the testnet command. See '--help'.".into()),
}?;
load_keypairs_from_yaml(path.to_path_buf())?
}
};
// Check if it's an empty vector, and return none.
if keypairs.is_empty() {
Err(
"No validator keypairs were found, unable to proceed. To generate \
testing keypairs, see 'testnet range --help'."
.into(),
)
} else {
Ok(keypairs)
}
}
/// Saves a keypair to a file inside the appropriate validator directory. Returns the saved path filename.
#[allow(dead_code)]
pub fn save_key(&self, key: &Keypair) -> Result<PathBuf, Error> {
use std::os::unix::fs::PermissionsExt;
let validator_config_path = self.data_dir.join(key.identifier());
let key_path = validator_config_path.join(DEFAULT_PRIVATE_KEY_FILENAME);
fs::create_dir_all(&validator_config_path)?;
let mut key_file = File::create(&key_path)?;
let mut perm = key_file.metadata()?.permissions();
perm.set_mode((libc::S_IWUSR | libc::S_IRUSR) as u32);
key_file.set_permissions(perm)?;
bincode::serialize_into(&mut key_file, &key)
.map_err(|e| Error::new(ErrorKind::InvalidData, e))?;
Ok(key_path)
Ok(config)
}
}
/// Parses the `testnet` CLI subcommand, modifying the `config` based upon the parameters in
/// `cli_args`.
fn process_testnet_subcommand(cli_args: &ArgMatches, mut config: Config) -> Result<Config, String> {
config.key_source = match cli_args.subcommand() {
("insecure", Some(sub_cli_args)) => {
let first = sub_cli_args
.value_of("first_validator")
.ok_or_else(|| "No first validator supplied")?
.parse::<usize>()
.map_err(|e| format!("Unable to parse first validator: {:?}", e))?;
let last = sub_cli_args
.value_of("last_validator")
.ok_or_else(|| "No last validator supplied")?
.parse::<usize>()
.map_err(|e| format!("Unable to parse last validator: {:?}", e))?;
if last < first {
return Err("Cannot supply a last validator less than the first".to_string());
}
KeySource::InsecureKeypairs((first..last).collect())
}
_ => KeySource::Disk,
};
Ok(config)
}

20
validator_client/src/duties/beacon_node_duties.rs
View File

@@ -1,20 +0,0 @@
use super::EpochDuties;
use types::{Epoch, PublicKey};
#[derive(Debug, PartialEq, Clone)]
pub enum BeaconNodeDutiesError {
RemoteFailure(String),
}
/// Defines the methods required to obtain a validators shuffling from a Beacon Node.
pub trait BeaconNodeDuties: Send + Sync {
/// Gets the duties for all validators.
///
/// Returns a vector of EpochDuties for each validator public key. The entry will be None for
/// validators that are not activated.
fn request_duties(
&self,
epoch: Epoch,
pub_keys: &[PublicKey],
) -> Result<EpochDuties, BeaconNodeDutiesError>;
}

133
validator_client/src/duties/epoch_duties.rs
View File

@@ -1,133 +0,0 @@
use std::collections::HashMap;
use std::fmt;
use std::ops::{Deref, DerefMut};
use types::{AttestationDuty, Epoch, PublicKey, Slot};
/// When work needs to be performed by a validator, this type is given back to the main service
/// which indicates all the information that required to process the work.
///
/// Note: This is calculated per slot, so a validator knows which slot is related to this struct.
#[derive(Debug, Clone)]
pub struct WorkInfo {
/// Validator needs to produce a block.
pub produce_block: bool,
/// Validator needs to produce an attestation. This supplies the required attestation data.
pub attestation_duty: Option<AttestationDuty>,
}
/// The information required for a validator to propose and attest during some epoch.
///
/// Generally obtained from a Beacon Node, this information contains the validators canonical index
/// (their sequence in the global validator induction process) and the "shuffling" for that index
/// for some epoch.
#[derive(Debug, PartialEq, Clone, Copy, Default)]
pub struct EpochDuty {
pub block_production_slot: Option<Slot>,
pub attestation_duty: AttestationDuty,
}
impl EpochDuty {
/// Returns `WorkInfo` if work needs to be done in the supplied `slot`
pub fn is_work_slot(&self, slot: Slot) -> Option<WorkInfo> {
// if validator is required to produce a slot return true
let produce_block = match self.block_production_slot {
Some(s) if s == slot => true,
_ => false,
};
// if the validator is required to attest to a index, create the data
let mut attestation_duty = None;
if self.attestation_duty.slot == slot {
attestation_duty = Some(self.attestation_duty)
}
if produce_block | attestation_duty.is_some() {
return Some(WorkInfo {
produce_block,
attestation_duty,
});
}
None
}
}
impl fmt::Display for EpochDuty {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
let mut display_block = String::from("None");
if let Some(block_slot) = self.block_production_slot {
display_block = block_slot.to_string();
}
write!(
f,
"produce block slot: {}, attestation slot: {}, attestation index: {}",
display_block, self.attestation_duty.slot, self.attestation_duty.index
)
}
}
/// Maps a list of keypairs (many validators) to an EpochDuty.
pub type EpochDuties = HashMap<PublicKey, Option<EpochDuty>>;
pub enum EpochDutiesMapError {
UnknownEpoch,
UnknownValidator,
}
/// Maps an `epoch` to some `EpochDuties` for a single validator.
pub struct EpochDutiesMap {
pub slots_per_epoch: u64,
pub map: HashMap<Epoch, EpochDuties>,
}
impl EpochDutiesMap {
pub fn new(slots_per_epoch: u64) -> Self {
Self {
slots_per_epoch,
map: HashMap::new(),
}
}
}
// Expose the hashmap methods
impl Deref for EpochDutiesMap {
type Target = HashMap<Epoch, EpochDuties>;
fn deref(&self) -> &Self::Target {
&self.map
}
}
impl DerefMut for EpochDutiesMap {
fn deref_mut(&mut self) -> &mut HashMap<Epoch, EpochDuties> {
&mut self.map
}
}
impl EpochDutiesMap {
/// Checks if the validator has work to do.
pub fn is_work_slot(
&self,
slot: Slot,
signer: &PublicKey,
) -> Result<Option<WorkInfo>, EpochDutiesMapError> {
let epoch = slot.epoch(self.slots_per_epoch);
let epoch_duties = self
.map
.get(&epoch)
.ok_or_else(|| EpochDutiesMapError::UnknownEpoch)?;
if let Some(epoch_duty) = epoch_duties.get(signer) {
if let Some(duty) = epoch_duty {
// Retrieves the duty for a validator at a given slot
Ok(duty.is_work_slot(slot))
} else {
// the validator isn't active
Ok(None)
}
} else {
// validator isn't known
Err(EpochDutiesMapError::UnknownValidator)
}
}
}
// TODO: add tests.

67
validator_client/src/duties/grpc.rs
View File

@@ -1,67 +0,0 @@
use super::beacon_node_duties::{BeaconNodeDuties, BeaconNodeDutiesError};
use super::epoch_duties::{EpochDuties, EpochDuty};
// to use if we manually specify a timeout
//use grpcio::CallOption;
use protos::services::{GetDutiesRequest, Validators};
use protos::services_grpc::ValidatorServiceClient;
use ssz::ssz_encode;
use std::collections::HashMap;
// use std::time::Duration;
use types::{AttestationDuty, Epoch, PublicKey, Slot};
impl BeaconNodeDuties for ValidatorServiceClient {
/// Requests all duties (block signing and committee attesting) from the Beacon Node (BN).
fn request_duties(
&self,
epoch: Epoch,
pub_keys: &[PublicKey],
) -> Result<EpochDuties, BeaconNodeDutiesError> {
// Get the required duties from all validators
// build the request
let mut req = GetDutiesRequest::new();
req.set_epoch(epoch.as_u64());
let mut validators = Validators::new();
validators.set_public_keys(pub_keys.iter().map(|v| ssz_encode(v)).collect());
req.set_validators(validators);
// set a timeout for requests
// let call_opt = CallOption::default().timeout(Duration::from_secs(2));
// send the request, get the duties reply
let reply = self
.get_validator_duties(&req)
.map_err(|err| BeaconNodeDutiesError::RemoteFailure(format!("{:?}", err)))?;
let mut epoch_duties: HashMap<PublicKey, Option<EpochDuty>> = HashMap::new();
for (index, validator_duty) in reply.get_active_validators().iter().enumerate() {
if !validator_duty.has_duty() {
// validator is inactive
epoch_duties.insert(pub_keys[index].clone(), None);
continue;
}
// active validator
let active_duty = validator_duty.get_duty();
let block_production_slot = {
if active_duty.has_block_production_slot() {
Some(Slot::from(active_duty.get_block_production_slot()))
} else {
None
}
};
let attestation_duty = AttestationDuty {
slot: Slot::from(active_duty.get_attestation_slot()),
index: active_duty.get_attestation_shard(),
committee_position: active_duty.get_committee_index() as usize,
committee_len: active_duty.get_committee_len() as usize,
};
let epoch_duty = EpochDuty {
block_production_slot,
attestation_duty,
};
epoch_duties.insert(pub_keys[index].clone(), Some(epoch_duty));
}
Ok(epoch_duties)
}
}

214
validator_client/src/duties/mod.rs
View File

@@ -1,214 +0,0 @@
mod beacon_node_duties;
mod epoch_duties;
mod grpc;
// TODO: reintroduce tests
//#[cfg(test)]
//mod test_node;
pub use self::beacon_node_duties::{BeaconNodeDuties, BeaconNodeDutiesError};
use self::epoch_duties::{EpochDuties, EpochDutiesMapError};
pub use self::epoch_duties::{EpochDutiesMap, WorkInfo};
use super::signer::Signer;
use futures::Async;
use parking_lot::RwLock;
use slog::{debug, error, info};
use std::fmt::Display;
use std::sync::Arc;
use types::{Epoch, PublicKey, Slot};
#[derive(Debug, PartialEq, Clone)]
pub enum UpdateOutcome {
/// The `EpochDuties` were not updated during this poll.
NoChange(Epoch),
/// The `EpochDuties` for the `epoch` were previously unknown, but obtained in the poll.
NewDuties(Epoch, EpochDuties),
/// New `EpochDuties` were obtained, different to those which were previously known. This is
/// likely to be the result of chain re-organisation.
DutiesChanged(Epoch, EpochDuties),
}
#[derive(Debug, PartialEq)]
pub enum Error {
DutiesMapPoisoned,
BeaconNodeDutiesError(BeaconNodeDutiesError),
UnknownEpoch,
UnknownValidator,
}
/// A polling state machine which ensures the latest `EpochDuties` are obtained from the Beacon
/// Node.
///
/// This keeps track of all validator keys and required voting slots.
pub struct DutiesManager<U: BeaconNodeDuties, S: Signer> {
pub duties_map: RwLock<EpochDutiesMap>,
/// A list of all signer objects known to the validator service.
pub signers: Arc<Vec<S>>,
pub beacon_node: Arc<U>,
}
impl<U: BeaconNodeDuties, S: Signer + Display> DutiesManager<U, S> {
/// Check the Beacon Node for `EpochDuties`.
///
/// be a wall-clock (e.g., system time, remote server time, etc.).
fn update(&self, epoch: Epoch) -> Result<UpdateOutcome, Error> {
let public_keys: Vec<PublicKey> = self.signers.iter().map(Signer::to_public).collect();
let duties = self.beacon_node.request_duties(epoch, &public_keys)?;
{
// If these duties were known, check to see if they're updates or identical.
if let Some(known_duties) = self.duties_map.read().get(&epoch) {
if *known_duties == duties {
return Ok(UpdateOutcome::NoChange(epoch));
}
}
}
if !self.duties_map.read().contains_key(&epoch) {
//TODO: Remove clone by removing duties from outcome
self.duties_map.write().insert(epoch, duties.clone());
return Ok(UpdateOutcome::NewDuties(epoch, duties));
}
// duties have changed
//TODO: Duties could be large here. Remove from display and avoid the clone.
self.duties_map.write().insert(epoch, duties.clone());
Ok(UpdateOutcome::DutiesChanged(epoch, duties))
}
/// A future wrapping around `update()`. This will perform logic based upon the update
/// process and complete once the update has completed.
pub fn run_update(&self, epoch: Epoch, log: slog::Logger) -> Result<Async<()>, ()> {
match self.update(epoch) {
Err(error) => error!(log, "Epoch duties poll error"; "error" => format!("{:?}", error)),
Ok(UpdateOutcome::NoChange(epoch)) => {
debug!(log, "No change in duties"; "epoch" => epoch)
}
Ok(UpdateOutcome::DutiesChanged(epoch, duties)) => {
info!(log, "Duties changed (potential re-org)"; "epoch" => epoch, "duties" => format!("{:?}", duties))
}
Ok(UpdateOutcome::NewDuties(epoch, duties)) => {
info!(log, "New duties obtained"; "epoch" => epoch);
print_duties(&log, duties);
}
};
Ok(Async::Ready(()))
}
/// Returns a list of (index, WorkInfo) indicating all the validators that have work to perform
/// this slot.
pub fn get_current_work(&self, slot: Slot) -> Option<Vec<(usize, WorkInfo)>> {
let mut current_work: Vec<(usize, WorkInfo)> = Vec::new();
// if the map is poisoned, return None
let duties = self.duties_map.read();
for (index, validator_signer) in self.signers.iter().enumerate() {
match duties.is_work_slot(slot, &validator_signer.to_public()) {
Ok(Some(work_type)) => current_work.push((index, work_type)),
Ok(None) => {} // No work for this validator
//TODO: This should really log an error, as we shouldn't end up with an err here.
Err(_) => {} // Unknown epoch or validator, no work
}
}
if current_work.is_empty() {
return None;
}
Some(current_work)
}
}
//TODO: Use error_chain to handle errors
impl From<BeaconNodeDutiesError> for Error {
fn from(e: BeaconNodeDutiesError) -> Error {
Error::BeaconNodeDutiesError(e)
}
}
//TODO: Use error_chain to handle errors
impl<T> From<std::sync::PoisonError<T>> for Error {
fn from(_e: std::sync::PoisonError<T>) -> Error {
Error::DutiesMapPoisoned
}
}
impl From<EpochDutiesMapError> for Error {
fn from(e: EpochDutiesMapError) -> Error {
match e {
EpochDutiesMapError::UnknownEpoch => Error::UnknownEpoch,
EpochDutiesMapError::UnknownValidator => Error::UnknownValidator,
}
}
}
fn print_duties(log: &slog::Logger, duties: EpochDuties) {
for (pk, duty) in duties.iter() {
if let Some(display_duty) = duty {
info!(log, "Validator: {}",pk; "Duty" => format!("{}",display_duty));
} else {
info!(log, "Validator: {}",pk; "Duty" => "None");
}
}
}
/* TODO: Modify tests for new Duties Manager form
#[cfg(test)]
mod tests {
use super::test_node::TestBeaconNode;
use super::*;
use bls::Keypair;
use slot_clock::TestingSlotClock;
use types::Slot;
// TODO: implement more thorough testing.
// https://github.com/sigp/lighthouse/issues/160
//
// These tests should serve as a good example for future tests.
#[test]
pub fn polling() {
let spec = Arc::new(ChainSpec::mainnet());
let duties_map = Arc::new(EpochDutiesMap::new(T::slots_per_epoch()));
let keypair = Keypair::random();
let slot_clock = Arc::new(TestingSlotClock::new(0));
let beacon_node = Arc::new(TestBeaconNode::default());
let manager = DutiesManager {
spec: spec.clone(),
pubkey: keypair.pk.clone(),
duties_map: duties_map.clone(),
slot_clock: slot_clock.clone(),
beacon_node: beacon_node.clone(),
};
// Configure response from the BeaconNode.
let duties = EpochDuties {
validator_index: 0,
block_production_slot: Some(Slot::new(10)),
};
beacon_node.set_next_shuffling_result(Ok(Some(duties)));
// Get the duties for the first time...
assert_eq!(
manager.poll(),
Ok(PollOutcome::NewDuties(Epoch::new(0), duties))
);
// Get the same duties again...
assert_eq!(manager.poll(), Ok(PollOutcome::NoChange(Epoch::new(0))));
// Return new duties.
let duties = EpochDuties {
validator_index: 0,
block_production_slot: Some(Slot::new(11)),
};
beacon_node.set_next_shuffling_result(Ok(Some(duties)));
assert_eq!(
manager.poll(),
Ok(PollOutcome::DutiesChanged(Epoch::new(0), duties))
);
// Return no duties.
beacon_node.set_next_shuffling_result(Ok(None));
assert_eq!(
manager.poll(),
Ok(PollOutcome::UnknownValidatorOrEpoch(Epoch::new(0)))
);
}
}
*/

32
validator_client/src/duties/test_node.rs
View File

@@ -1,32 +0,0 @@
use super::traits::{BeaconNode, BeaconNodeError};
use super::EpochDuties;
use bls::PublicKey;
use std::sync::RwLock;
use types::Epoch;
type ShufflingResult = Result<Option<EpochDuties>, BeaconNodeError>;
/// A test-only struct used to simulate a Beacon Node.
#[derive(Default)]
pub struct TestBeaconNode {
pub request_shuffling_input: RwLock<Option<(Epoch, PublicKey)>>,
pub request_shuffling_result: RwLock<Option<ShufflingResult>>,
}
impl TestBeaconNode {
/// Set the result to be returned when `request_shuffling` is called.
pub fn set_next_shuffling_result(&self, result: ShufflingResult) {
*self.request_shuffling_result.write().unwrap() = Some(result);
}
}
impl BeaconNode for TestBeaconNode {
/// Returns the value specified by the `set_next_shuffling_result`.
fn request_shuffling(&self, epoch: Epoch, public_key: &PublicKey) -> ShufflingResult {
*self.request_shuffling_input.write().unwrap() = Some((epoch, public_key.clone()));
match *self.request_shuffling_result.read().unwrap() {
Some(ref r) => r.clone(),
None => panic!("TestBeaconNode: produce_result == None"),
}
}
}

426
validator_client/src/duties_service.rs Normal file
View File

@@ -0,0 +1,426 @@
use crate::validator_store::ValidatorStore;
use environment::RuntimeContext;
use exit_future::Signal;
use futures::{Future, IntoFuture, Stream};
use parking_lot::RwLock;
use remote_beacon_node::{RemoteBeaconNode, ValidatorDuty};
use slog::{crit, error, info, trace, warn};
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{ChainSpec, Epoch, EthSpec, PublicKey, Slot};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
/// Remove any duties where the `duties_epoch < current_epoch - PRUNE_DEPTH`.
const PRUNE_DEPTH: u64 = 4;
type BaseHashMap = HashMap<PublicKey, HashMap<Epoch, ValidatorDuty>>;
/// The outcome of inserting some `ValidatorDuty` into the `DutiesStore`.
enum InsertOutcome {
/// These are the first duties received for this validator.
NewValidator,
/// The duties for this given epoch were previously unknown and have been stored.
NewEpoch,
/// The duties were identical to some already in the store.
Identical,
/// There were duties for this validator and epoch in the store that were different to the ones
/// provided. The existing duties were replaced.
Replaced,
/// The given duties were invalid.
Invalid,
}
#[derive(Default)]
pub struct DutiesStore {
store: RwLock<BaseHashMap>,
}
impl DutiesStore {
fn block_producers(&self, slot: Slot, slots_per_epoch: u64) -> Vec<PublicKey> {
self.store
.read()
.iter()
// As long as a `HashMap` iterator does not return duplicate keys, neither will this
// function.
.filter_map(|(_validator_pubkey, validator_map)| {
let epoch = slot.epoch(slots_per_epoch);
validator_map.get(&epoch).and_then(|duties| {
if duties.block_proposal_slot == Some(slot) {
Some(duties.validator_pubkey.clone())
} else {
None
}
})
})
.collect()
}
fn attesters(&self, slot: Slot, slots_per_epoch: u64) -> Vec<ValidatorDuty> {
self.store
.read()
.iter()
// As long as a `HashMap` iterator does not return duplicate keys, neither will this
// function.
.filter_map(|(_validator_pubkey, validator_map)| {
let epoch = slot.epoch(slots_per_epoch);
validator_map.get(&epoch).and_then(|duties| {
if duties.attestation_slot == Some(slot) {
Some(duties)
} else {
None
}
})
})
.cloned()
.collect()
}
fn insert(&self, epoch: Epoch, duties: ValidatorDuty, slots_per_epoch: u64) -> InsertOutcome {
let mut store = self.store.write();
if !duties_match_epoch(&duties, epoch, slots_per_epoch) {
return InsertOutcome::Invalid;
}
if let Some(validator_map) = store.get_mut(&duties.validator_pubkey) {
if let Some(known_duties) = validator_map.get_mut(&epoch) {
if *known_duties == duties {
InsertOutcome::Identical
} else {
*known_duties = duties;
InsertOutcome::Replaced
}
} else {
validator_map.insert(epoch, duties);
InsertOutcome::NewEpoch
}
} else {
let validator_pubkey = duties.validator_pubkey.clone();
let mut validator_map = HashMap::new();
validator_map.insert(epoch, duties);
store.insert(validator_pubkey, validator_map);
InsertOutcome::NewValidator
}
}
fn prune(&self, prior_to: Epoch) {
self.store
.write()
.retain(|_validator_pubkey, validator_map| {
validator_map.retain(|epoch, _duties| *epoch >= prior_to);
!validator_map.is_empty()
});
}
}
pub struct DutiesServiceBuilder<T, E: EthSpec> {
validator_store: Option<ValidatorStore<T, E>>,
slot_clock: Option<T>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
impl<T: SlotClock + 'static, E: EthSpec> DutiesServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
validator_store: None,
slot_clock: None,
beacon_node: None,
context: None,
}
}
pub fn validator_store(mut self, store: ValidatorStore<T, E>) -> Self {
self.validator_store = Some(store);
self
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(slot_clock);
self
}
pub fn beacon_node(mut self, beacon_node: RemoteBeaconNode<E>) -> Self {
self.beacon_node = Some(beacon_node);
self
}
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
pub fn build(self) -> Result<DutiesService<T, E>, String> {
Ok(DutiesService {
inner: Arc::new(Inner {
store: Arc::new(DutiesStore::default()),
validator_store: self
.validator_store
.ok_or_else(|| "Cannot build DutiesService without validator_store")?,
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build DutiesService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build DutiesService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build DutiesService without runtime_context")?,
}),
})
}
}
/// Helper to minimise `Arc` usage.
pub struct Inner<T, E: EthSpec> {
store: Arc<DutiesStore>,
validator_store: ValidatorStore<T, E>,
slot_clock: T,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
}
/// Maintains a store of the duties for all voting validators in the `validator_store`.
///
/// Polls the beacon node at the start of each epoch, collecting duties for the current and next
/// epoch.
pub struct DutiesService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for DutiesService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for DutiesService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
/// Returns the pubkeys of the validators which are assigned to propose in the given slot.
///
/// In normal cases, there should be 0 or 1 validators returned. In extreme cases (i.e., deep forking)
///
/// It is possible that multiple validators have an identical proposal slot, however that is
/// likely the result of heavy forking (lol) or inconsistent beacon node connections.
pub fn block_producers(&self, slot: Slot) -> Vec<PublicKey> {
self.store.block_producers(slot, E::slots_per_epoch())
}
/// Returns all `ValidatorDuty` for the given `slot`.
pub fn attesters(&self, slot: Slot) -> Vec<ValidatorDuty> {
self.store.attesters(slot, E::slots_per_epoch())
}
/// Start the service that periodically polls the beacon node for validator duties.
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let log = self.context.log.clone();
let duration_to_next_slot = self
.slot_clock
.duration_to_next_slot()
.ok_or_else(|| "Unable to determine duration to next slot".to_string())?;
let interval = {
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
Interval::new(
Instant::now() + duration_to_next_slot + TIME_DELAY_FROM_SLOT,
slot_duration,
)
};
let (exit_signal, exit_fut) = exit_future::signal();
let service = self.clone();
let log_1 = log.clone();
let log_2 = log.clone();
// Run an immediate update before starting the updater service.
self.context.executor.spawn(service.clone().do_update());
self.context.executor.spawn(
exit_fut
.until(
interval
.map_err(move |e| {
crit! {
log_1,
"Timer thread failed";
"error" => format!("{}", e)
}
})
.for_each(move |_| service.clone().do_update())
// Prevent any errors from escaping and stopping the interval.
.then(|_| Ok(())),
)
.map(move |_| info!(log_2, "Shutdown complete")),
);
Ok(exit_signal)
}
/// Attempt to download the duties of all managed validators for this epoch and the next.
fn do_update(&self) -> impl Future<Item = (), Error = ()> {
let service_1 = self.clone();
let service_2 = self.clone();
let service_3 = self.clone();
let log_1 = self.context.log.clone();
let log_2 = self.context.log.clone();
self.slot_clock
.now()
.ok_or_else(move || {
error!(log_1, "Duties manager failed to read slot clock");
})
.into_future()
.map(move |slot| {
let epoch = slot.epoch(E::slots_per_epoch());
if slot % E::slots_per_epoch() == 0 {
let prune_below = epoch - PRUNE_DEPTH;
trace!(
log_2,
"Pruning duties cache";
"pruning_below" => prune_below.as_u64(),
"current_epoch" => epoch.as_u64(),
);
service_1.store.prune(prune_below);
}
epoch
})
.and_then(move |epoch| {
let log = service_2.context.log.clone();
service_2.update_epoch(epoch).then(move |result| {
if let Err(e) = result {
error!(
log,
"Failed to get current epoch duties";
"http_error" => format!("{:?}", e)
);
}
let log = service_3.context.log.clone();
service_3.update_epoch(epoch + 1).map_err(move |e| {
error!(
log,
"Failed to get next epoch duties";
"http_error" => format!("{:?}", e)
);
})
})
})
.map(|_| ())
}
/// Attempt to download the duties of all managed validators for the given `epoch`.
fn update_epoch(self, epoch: Epoch) -> impl Future<Item = (), Error = String> {
let service_1 = self.clone();
let service_2 = self.clone();
let pubkeys = service_1.validator_store.voting_pubkeys();
service_1
.beacon_node
.http
.validator()
.get_duties_bulk(epoch, pubkeys.as_slice())
.map(move |all_duties| (epoch, all_duties))
.map_err(move |e| format!("Failed to get duties for epoch {}: {:?}", epoch, e))
.map(move |(epoch, all_duties)| {
let log = service_2.context.log.clone();
let mut new_validator = 0;
let mut new_epoch = 0;
let mut identical = 0;
let mut replaced = 0;
let mut invalid = 0;
all_duties.into_iter().for_each(|duties| {
match service_2
.store
.insert(epoch, duties.clone(), E::slots_per_epoch())
{
InsertOutcome::NewValidator => {
info!(
log,
"First duty assignment for validator";
"proposal_slot" => format!("{:?}", &duties.block_proposal_slot),
"attestation_slot" => format!("{:?}", &duties.attestation_slot),
"validator" => format!("{:?}", &duties.validator_pubkey)
);
new_validator += 1
}
InsertOutcome::NewEpoch => new_epoch += 1,
InsertOutcome::Identical => identical += 1,
InsertOutcome::Replaced => replaced += 1,
InsertOutcome::Invalid => invalid += 1,
};
});
if invalid > 0 {
error!(
log,
"Received invalid duties from beacon node";
"bad_duty_count" => invalid,
"info" => "Duties are from wrong epoch."
)
}
trace!(
log,
"Performed duties update";
"identical" => identical,
"new_epoch" => new_epoch,
"new_validator" => new_validator,
"replaced" => replaced,
"epoch" => format!("{}", epoch)
);
if replaced > 0 {
warn!(
log,
"Duties changed during routine update";
"info" => "Chain re-org likely occurred."
)
}
})
}
}
/// Returns `true` if the slots in the `duties` are from the given `epoch`
fn duties_match_epoch(duties: &ValidatorDuty, epoch: Epoch, slots_per_epoch: u64) -> bool {
if let Some(attestation_slot) = duties.attestation_slot {
if attestation_slot.epoch(slots_per_epoch) != epoch {
return false;
}
}
if let Some(block_proposal_slot) = duties.block_proposal_slot {
if block_proposal_slot.epoch(slots_per_epoch) != epoch {
return false;
}
}
true
}

12
validator_client/src/error.rs
View File

@@ -1,12 +0,0 @@
use error_chain::error_chain;
error_chain! {
links { }
errors {
SystemTimeError(t: String ) {
description("Error reading system time"),
display("SystemTimeError: '{}'", t)
}
}
}

173
validator_client/src/fork_service.rs Normal file
View File

@@ -0,0 +1,173 @@
use environment::RuntimeContext;
use exit_future::Signal;
use futures::{Future, Stream};
use parking_lot::RwLock;
use remote_beacon_node::RemoteBeaconNode;
use slog::{crit, info, trace};
use slot_clock::SlotClock;
use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{ChainSpec, EthSpec, Fork};
/// Delay this period of time after the slot starts. This allows the node to process the new slot.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(80);
/// Builds a `ForkService`.
pub struct ForkServiceBuilder<T, E: EthSpec> {
fork: Option<Fork>,
slot_clock: Option<T>,
beacon_node: Option<RemoteBeaconNode<E>>,
context: Option<RuntimeContext<E>>,
}
impl<T: SlotClock + 'static, E: EthSpec> ForkServiceBuilder<T, E> {
pub fn new() -> Self {
Self {
fork: None,
slot_clock: None,
beacon_node: None,
context: None,
}
}
pub fn slot_clock(mut self, slot_clock: T) -> Self {
self.slot_clock = Some(slot_clock);
self
}
pub fn beacon_node(mut self, beacon_node: RemoteBeaconNode<E>) -> Self {
self.beacon_node = Some(beacon_node);
self
}
pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
self.context = Some(context);
self
}
pub fn build(self) -> Result<ForkService<T, E>, String> {
Ok(ForkService {
inner: Arc::new(Inner {
fork: RwLock::new(self.fork),
slot_clock: self
.slot_clock
.ok_or_else(|| "Cannot build ForkService without slot_clock")?,
beacon_node: self
.beacon_node
.ok_or_else(|| "Cannot build ForkService without beacon_node")?,
context: self
.context
.ok_or_else(|| "Cannot build ForkService without runtime_context")?,
}),
})
}
}
/// Helper to minimise `Arc` usage.
pub struct Inner<T, E: EthSpec> {
fork: RwLock<Option<Fork>>,
beacon_node: RemoteBeaconNode<E>,
context: RuntimeContext<E>,
slot_clock: T,
}
/// Attempts to download the `Fork` struct from the beacon node at the start of each epoch.
pub struct ForkService<T, E: EthSpec> {
inner: Arc<Inner<T, E>>,
}
impl<T, E: EthSpec> Clone for ForkService<T, E> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl<T, E: EthSpec> Deref for ForkService<T, E> {
type Target = Inner<T, E>;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
impl<T: SlotClock + 'static, E: EthSpec> ForkService<T, E> {
/// Returns the last fork downloaded from the beacon node, if any.
pub fn fork(&self) -> Option<Fork> {
self.fork.read().clone()
}
/// Starts the service that periodically polls for the `Fork`.
pub fn start_update_service(&self, spec: &ChainSpec) -> Result<Signal, String> {
let log = self.context.log.clone();
let duration_to_next_epoch = self
.slot_clock
.duration_to_next_epoch(E::slots_per_epoch())
.ok_or_else(|| "Unable to determine duration to next epoch".to_string())?;
let interval = {
let slot_duration = Duration::from_millis(spec.milliseconds_per_slot);
Interval::new(
Instant::now() + duration_to_next_epoch + TIME_DELAY_FROM_SLOT,
slot_duration * E::slots_per_epoch() as u32,
)
};
let (exit_signal, exit_fut) = exit_future::signal();
let service = self.clone();
let log_1 = log.clone();
let log_2 = log.clone();
// Run an immediate update before starting the updater service.
self.context.executor.spawn(service.clone().do_update());
self.context.executor.spawn(
exit_fut
.until(
interval
.map_err(move |e| {
crit! {
log_1,
"Timer thread failed";
"error" => format!("{}", e)
}
})
.for_each(move |_| service.do_update())
// Prevent any errors from escaping and stopping the interval.
.then(|_| Ok(())),
)
.map(move |_| info!(log_2, "Shutdown complete")),
);
Ok(exit_signal)
}
/// Attempts to download the `Fork` from the server.
fn do_update(&self) -> impl Future<Item = (), Error = ()> {
let service_1 = self.clone();
let log_1 = service_1.context.log.clone();
let log_2 = service_1.context.log.clone();
self.inner
.beacon_node
.http
.beacon()
.get_fork()
.map(move |fork| *(service_1.fork.write()) = Some(fork))
.map(move |_| trace!(log_1, "Fork update success"))
.map_err(move |e| {
trace!(
log_2,
"Fork update failed";
"error" => format!("Error retrieving fork: {:?}", e)
)
})
// Returning an error will stop the interval. This is not desired, a single failure
// should not stop all future attempts.
.then(|_| Ok(()))
}
}

450
validator_client/src/lib.rs
View File

@@ -1,258 +1,266 @@
mod attestation_producer;
mod block_producer;
mod attestation_service;
mod block_service;
mod cli;
mod config;
mod duties;
mod error;
mod service;
mod signer;
mod duties_service;
mod fork_service;
mod validator_store;
pub mod validator_directory;
pub use cli::cli_app;
pub use config::Config;
pub use config::{Config, KeySource};
use attestation_service::{AttestationService, AttestationServiceBuilder};
use block_service::{BlockService, BlockServiceBuilder};
use clap::ArgMatches;
use config::{Config as ClientConfig, KeySource};
use duties_service::{DutiesService, DutiesServiceBuilder};
use environment::RuntimeContext;
use eth2_config::Eth2Config;
use exit_future::Signal;
use futures::Stream;
use lighthouse_bootstrap::Bootstrapper;
use parking_lot::RwLock;
use protos::services_grpc::ValidatorServiceClient;
use service::Service;
use slog::{error, info, warn, Logger};
use fork_service::{ForkService, ForkServiceBuilder};
use futures::{
future::{self, loop_fn, Loop},
Future, IntoFuture,
};
use remote_beacon_node::RemoteBeaconNode;
use slog::{error, info, Logger};
use slot_clock::SlotClock;
use std::path::PathBuf;
use std::sync::Arc;
use slot_clock::SystemTimeSlotClock;
use std::time::{Duration, Instant};
use tokio::timer::Interval;
use types::{EthSpec, Keypair};
use tokio::timer::Delay;
use types::EthSpec;
use validator_store::ValidatorStore;
/// A fixed amount of time after a slot to perform operations. This gives the node time to complete
/// per-slot processes.
const TIME_DELAY_FROM_SLOT: Duration = Duration::from_millis(100);
/// The interval between attempts to contact the beacon node during startup.
const RETRY_DELAY: Duration = Duration::from_secs(2);
/// The global timeout for HTTP requests to the beacon node.
const HTTP_TIMEOUT: Duration = Duration::from_secs(12);
#[derive(Clone)]
pub struct ProductionValidatorClient<T: EthSpec> {
context: RuntimeContext<T>,
service: Arc<Service<ValidatorServiceClient, Keypair, T>>,
exit_signals: Arc<RwLock<Vec<Signal>>>,
duties_service: DutiesService<SystemTimeSlotClock, T>,
fork_service: ForkService<SystemTimeSlotClock, T>,
block_service: BlockService<SystemTimeSlotClock, T>,
attestation_service: AttestationService<SystemTimeSlotClock, T>,
exit_signals: Vec<Signal>,
}
impl<T: EthSpec> ProductionValidatorClient<T> {
/// Instantiates the validator client, _without_ starting the timers to trigger block
/// and attestation production.
pub fn new_from_cli(context: RuntimeContext<T>, matches: &ArgMatches) -> Result<Self, String> {
let mut log = context.log.clone();
let (client_config, eth2_config) = get_configs(&matches, &mut log)
.map_err(|e| format!("Unable to initialize config: {}", e))?;
info!(
log,
"Starting validator client";
"datadir" => client_config.full_data_dir().expect("Unable to find datadir").to_str(),
);
let service: Service<ValidatorServiceClient, Keypair, T> =
Service::initialize_service(client_config, eth2_config, log.clone())
.map_err(|e| e.to_string())?;
Ok(Self {
context,
service: Arc::new(service),
exit_signals: Arc::new(RwLock::new(vec![])),
})
pub fn new_from_cli(
context: RuntimeContext<T>,
cli_args: &ArgMatches,
) -> impl Future<Item = Self, Error = String> {
Config::from_cli(&cli_args)
.into_future()
.map_err(|e| format!("Unable to initialize config: {}", e))
.and_then(|config| Self::new(context, config))
}
/// Starts the timers to trigger block and attestation production.
pub fn start_service(&self) -> Result<(), String> {
let service = self.clone().service;
let log = self.context.log.clone();
let duration_to_next_slot = service
.slot_clock
.duration_to_next_slot()
.ok_or_else(|| "Unable to determine duration to next slot. Exiting.".to_string())?;
// set up the validator work interval - start at next slot and proceed every slot
let interval = {
// Set the interval to start at the next slot, and every slot after
let slot_duration = Duration::from_millis(service.spec.milliseconds_per_slot);
//TODO: Handle checked add correctly
Interval::new(Instant::now() + duration_to_next_slot, slot_duration)
};
if service.slot_clock.now().is_none() {
warn!(
log,
"Starting node prior to genesis";
);
}
/// Instantiates the validator client, _without_ starting the timers to trigger block
/// and attestation production.
pub fn new(
mut context: RuntimeContext<T>,
config: Config,
) -> impl Future<Item = Self, Error = String> {
let log_1 = context.log.clone();
let log_2 = context.log.clone();
let log_3 = context.log.clone();
info!(
log,
"Waiting for next slot";
"seconds_to_wait" => duration_to_next_slot.as_secs()
log_1,
"Starting validator client";
"beacon_node" => &config.http_server,
"datadir" => format!("{:?}", config.data_dir),
);
let (exit_signal, exit_fut) = exit_future::signal();
RemoteBeaconNode::new_with_timeout(config.http_server.clone(), HTTP_TIMEOUT)
.map_err(|e| format!("Unable to init beacon node http client: {}", e))
.into_future()
.and_then(move |beacon_node| wait_for_node(beacon_node, log_2))
.and_then(|beacon_node| {
beacon_node
.http
.spec()
.get_eth2_config()
.map(|eth2_config| (beacon_node, eth2_config))
.map_err(|e| format!("Unable to read eth2 config from beacon node: {:?}", e))
})
.and_then(|(beacon_node, eth2_config)| {
beacon_node
.http
.beacon()
.get_genesis_time()
.map(|genesis_time| (beacon_node, eth2_config, genesis_time))
.map_err(|e| format!("Unable to read genesis time from beacon node: {:?}", e))
})
.and_then(move |(beacon_node, remote_eth2_config, genesis_time)| {
// Do not permit a connection to a beacon node using different spec constants.
if context.eth2_config.spec_constants != remote_eth2_config.spec_constants {
return Err(format!(
"Beacon node is using an incompatible spec. Got {}, expected {}",
remote_eth2_config.spec_constants, context.eth2_config.spec_constants
));
}
self.exit_signals.write().push(exit_signal);
// Note: here we just assume the spec variables of the remote node. This is very useful
// for testnets, but perhaps a security issue when it comes to mainnet.
//
// A damaging attack would be for a beacon node to convince the validator client of a
// different `SLOTS_PER_EPOCH` variable. This could result in slashable messages being
// produced. We are safe from this because `SLOTS_PER_EPOCH` is a type-level constant
// for Lighthouse.
context.eth2_config = remote_eth2_config;
/* kick off the core service */
self.context.executor.spawn(
interval
.map_err(move |e| {
error! {
log,
"Timer thread failed";
"error" => format!("{}", e)
}
let slot_clock = SystemTimeSlotClock::new(
context.eth2_config.spec.genesis_slot,
Duration::from_secs(genesis_time),
Duration::from_millis(context.eth2_config.spec.milliseconds_per_slot),
);
let fork_service = ForkServiceBuilder::new()
.slot_clock(slot_clock.clone())
.beacon_node(beacon_node.clone())
.runtime_context(context.service_context("fork"))
.build()?;
let validator_store: ValidatorStore<SystemTimeSlotClock, T> =
match &config.key_source {
// Load pre-existing validators from the data dir.
//
// Use the `account_manager` to generate these files.
KeySource::Disk => ValidatorStore::load_from_disk(
config.data_dir.clone(),
context.eth2_config.spec.clone(),
fork_service.clone(),
log_3.clone(),
)?,
// Generate ephemeral insecure keypairs for testing purposes.
//
// Do not use in production.
KeySource::InsecureKeypairs(indices) => {
ValidatorStore::insecure_ephemeral_validators(
&indices,
context.eth2_config.spec.clone(),
fork_service.clone(),
log_3.clone(),
)?
}
};
info!(
log_3,
"Loaded validator keypair store";
"voting_validators" => validator_store.num_voting_validators()
);
let duties_service = DutiesServiceBuilder::new()
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_node(beacon_node.clone())
.runtime_context(context.service_context("duties"))
.build()?;
let block_service = BlockServiceBuilder::new()
.duties_service(duties_service.clone())
.slot_clock(slot_clock.clone())
.validator_store(validator_store.clone())
.beacon_node(beacon_node.clone())
.runtime_context(context.service_context("block"))
.build()?;
let attestation_service = AttestationServiceBuilder::new()
.duties_service(duties_service.clone())
.slot_clock(slot_clock)
.validator_store(validator_store)
.beacon_node(beacon_node)
.runtime_context(context.service_context("attestation"))
.build()?;
Ok(Self {
context,
duties_service,
fork_service,
block_service,
attestation_service,
exit_signals: vec![],
})
.and_then(move |_| if exit_fut.is_live() { Ok(()) } else { Err(()) })
.for_each(move |_| {
// wait for node to process
std::thread::sleep(TIME_DELAY_FROM_SLOT);
// if a non-fatal error occurs, proceed to the next slot.
let _ignore_error = service.per_slot_execution();
// completed a slot process
Ok(())
}),
);
})
}
pub fn start_service(&mut self) -> Result<(), String> {
let duties_exit = self
.duties_service
.start_update_service(&self.context.eth2_config.spec)
.map_err(|e| format!("Unable to start duties service: {}", e))?;
let fork_exit = self
.fork_service
.start_update_service(&self.context.eth2_config.spec)
.map_err(|e| format!("Unable to start fork service: {}", e))?;
let block_exit = self
.block_service
.start_update_service(&self.context.eth2_config.spec)
.map_err(|e| format!("Unable to start block service: {}", e))?;
let attestation_exit = self
.attestation_service
.start_update_service(&self.context.eth2_config.spec)
.map_err(|e| format!("Unable to start attestation service: {}", e))?;
self.exit_signals = vec![duties_exit, fork_exit, block_exit, attestation_exit];
Ok(())
}
}
/// Parses the CLI arguments and attempts to load the client and eth2 configuration.
///
/// This is not a pure function, it reads from disk and may contact network servers.
fn get_configs(
cli_args: &ArgMatches,
mut log: &mut Logger,
) -> Result<(ClientConfig, Eth2Config), String> {
let mut client_config = ClientConfig::default();
/// Request the version from the node, looping back and trying again on failure. Exit once the node
/// has been contacted.
fn wait_for_node<E: EthSpec>(
beacon_node: RemoteBeaconNode<E>,
log: Logger,
) -> impl Future<Item = RemoteBeaconNode<E>, Error = String> {
// Try to get the version string from the node, looping until success is returned.
loop_fn(beacon_node.clone(), move |beacon_node| {
let log = log.clone();
beacon_node
.clone()
.http
.node()
.get_version()
.map_err(|e| format!("{:?}", e))
.then(move |result| {
let future: Box<dyn Future<Item = Loop<_, _>, Error = String> + Send> = match result
{
Ok(version) => {
info!(
log,
"Connected to beacon node";
"version" => version,
);
client_config.apply_cli_args(&cli_args, &mut log)?;
Box::new(future::ok(Loop::Break(beacon_node)))
}
Err(e) => {
error!(
log,
"Unable to connect to beacon node";
"error" => format!("{:?}", e),
);
if let Some(server) = cli_args.value_of("server") {
client_config.server = server.to_string();
}
Box::new(
Delay::new(Instant::now() + RETRY_DELAY)
.map_err(|e| format!("Failed to trigger delay: {:?}", e))
.and_then(|_| future::ok(Loop::Continue(beacon_node))),
)
}
};
if let Some(port) = cli_args.value_of("server-http-port") {
client_config.server_http_port = port
.parse::<u16>()
.map_err(|e| format!("Unable to parse HTTP port: {:?}", e))?;
}
if let Some(port) = cli_args.value_of("server-grpc-port") {
client_config.server_grpc_port = port
.parse::<u16>()
.map_err(|e| format!("Unable to parse gRPC port: {:?}", e))?;
}
info!(
*log,
"Beacon node connection info";
"grpc_port" => client_config.server_grpc_port,
"http_port" => client_config.server_http_port,
"server" => &client_config.server,
);
let (client_config, eth2_config) = match cli_args.subcommand() {
("testnet", Some(sub_cli_args)) => {
if cli_args.is_present("eth2-config") && sub_cli_args.is_present("bootstrap") {
return Err(
"Cannot specify --eth2-config and --bootstrap as it may result \
in ambiguity."
.into(),
);
}
process_testnet_subcommand(sub_cli_args, client_config, log)
}
_ => return Err("You must use the testnet command. See '--help'.".into()),
}?;
Ok((client_config, eth2_config))
}
/// Parses the `testnet` CLI subcommand.
///
/// This is not a pure function, it reads from disk and may contact network servers.
fn process_testnet_subcommand(
cli_args: &ArgMatches,
mut client_config: ClientConfig,
log: &Logger,
) -> Result<(ClientConfig, Eth2Config), String> {
let eth2_config = if cli_args.is_present("bootstrap") {
info!(log, "Connecting to bootstrap server");
let bootstrapper = Bootstrapper::connect(
format!(
"http://{}:{}",
client_config.server, client_config.server_http_port
),
&log,
)?;
let eth2_config = bootstrapper.eth2_config()?;
info!(
log,
"Bootstrapped eth2 config via HTTP";
"slot_time_millis" => eth2_config.spec.milliseconds_per_slot,
"spec" => &eth2_config.spec_constants,
);
eth2_config
} else {
match cli_args.value_of("spec") {
Some("mainnet") => Eth2Config::mainnet(),
Some("minimal") => Eth2Config::minimal(),
Some("interop") => Eth2Config::interop(),
_ => return Err("No --spec flag provided. See '--help'.".into()),
}
};
client_config.key_source = match cli_args.subcommand() {
("insecure", Some(sub_cli_args)) => {
let first = sub_cli_args
.value_of("first_validator")
.ok_or_else(|| "No first validator supplied")?
.parse::<usize>()
.map_err(|e| format!("Unable to parse first validator: {:?}", e))?;
let count = sub_cli_args
.value_of("validator_count")
.ok_or_else(|| "No validator count supplied")?
.parse::<usize>()
.map_err(|e| format!("Unable to parse validator count: {:?}", e))?;
info!(
log,
"Generating unsafe testing keys";
"first_validator" => first,
"count" => count
);
KeySource::TestingKeypairRange(first..first + count)
}
("interop-yaml", Some(sub_cli_args)) => {
let path = sub_cli_args
.value_of("path")
.ok_or_else(|| "No yaml path supplied")?
.parse::<PathBuf>()
.map_err(|e| format!("Unable to parse yaml path: {:?}", e))?;
info!(
log,
"Loading keypairs from interop YAML format";
"path" => format!("{:?}", path),
);
KeySource::YamlKeypairs(path)
}
_ => KeySource::Disk,
};
Ok((client_config, eth2_config))
future
})
})
.map(|_| beacon_node)
}

363
validator_client/src/service.rs
View File

@@ -1,363 +0,0 @@
/// The Validator Client service.
///
/// Connects to a beacon node and negotiates the correct chain id.
///
/// Once connected, the service loads known validators keypairs from disk. Every slot,
/// the service pings the beacon node, asking for new duties for each of the validators.
///
/// When a validator needs to either produce a block or sign an attestation, it requests the
/// data from the beacon node and performs the signing before publishing the block to the beacon
/// node.
use crate::attestation_producer::AttestationProducer;
use crate::block_producer::{BeaconBlockGrpcClient, BlockProducer};
use crate::config::Config as ValidatorConfig;
use crate::duties::{BeaconNodeDuties, DutiesManager, EpochDutiesMap};
use crate::error as error_chain;
use crate::signer::Signer;
use bls::Keypair;
use eth2_config::Eth2Config;
use grpcio::{ChannelBuilder, EnvBuilder};
use parking_lot::RwLock;
use protos::services::Empty;
use protos::services_grpc::{
AttestationServiceClient, BeaconBlockServiceClient, BeaconNodeServiceClient,
ValidatorServiceClient,
};
use slog::{crit, error, info, trace, warn};
use slot_clock::{SlotClock, SystemTimeSlotClock};
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use types::{ChainSpec, Epoch, EthSpec, Fork, Slot};
/// The validator service. This is the main thread that executes and maintains validator
/// duties.
//TODO: Generalize the BeaconNode types to use testing
pub struct Service<B: BeaconNodeDuties + 'static, S: Signer + 'static, E: EthSpec> {
/// The node's current fork version we are processing on.
fork: Fork,
/// The slot clock for this service.
pub slot_clock: SystemTimeSlotClock,
/// The slot that is currently, or was previously processed by the service.
current_slot: RwLock<Option<Slot>>,
slots_per_epoch: u64,
/// The chain specification for this clients instance.
pub spec: Arc<ChainSpec>,
/// The duties manager which maintains the state of when to perform actions.
duties_manager: Arc<DutiesManager<B, S>>,
// GRPC Clients
/// The beacon block GRPC client.
beacon_block_client: Arc<BeaconBlockGrpcClient>,
/// The attester GRPC client.
attestation_client: Arc<AttestationServiceClient>,
/// The validator client logger.
log: slog::Logger,
_phantom: PhantomData<E>,
}
impl<E: EthSpec> Service<ValidatorServiceClient, Keypair, E> {
/// Initial connection to the beacon node to determine its properties.
///
/// This tries to connect to a beacon node. Once connected, it initialised the gRPC clients
/// and returns an instance of the service.
pub fn initialize_service(
client_config: ValidatorConfig,
eth2_config: Eth2Config,
log: slog::Logger,
) -> error_chain::Result<Service<ValidatorServiceClient, Keypair, E>> {
let server_url = format!(
"{}:{}",
client_config.server, client_config.server_grpc_port
);
let env = Arc::new(EnvBuilder::new().build());
// Beacon node gRPC beacon node endpoints.
let beacon_node_client = {
let ch = ChannelBuilder::new(env.clone()).connect(&server_url);
BeaconNodeServiceClient::new(ch)
};
// retrieve node information and validate the beacon node
let node_info = loop {
match beacon_node_client.info(&Empty::new()) {
Err(e) => {
let retry_seconds = 5;
warn!(
log,
"Could not connect to beacon node";
"error" => format!("{:?}", e),
"retry_in" => format!("{} seconds", retry_seconds),
);
std::thread::sleep(Duration::from_secs(retry_seconds));
continue;
}
Ok(info) => {
// verify the node's network id
if eth2_config.spec.network_id != info.network_id as u8 {
error!(
log,
"Beacon Node's genesis time is in the future. No work to do.\n Exiting"
);
return Err(format!("Beacon node has the wrong chain id. Expected chain id: {}, node's chain id: {}", eth2_config.spec.network_id, info.network_id).into());
}
break info;
}
};
};
// build requisite objects to form Self
let genesis_time = node_info.get_genesis_time();
let genesis_slot = Slot::from(node_info.get_genesis_slot());
info!(
log,
"Beacon node connected";
"version" => node_info.version.clone(),
"network_id" => node_info.network_id,
"genesis_time" => genesis_time
);
let proto_fork = node_info.get_fork();
let mut previous_version: [u8; 4] = [0; 4];
let mut current_version: [u8; 4] = [0; 4];
previous_version.copy_from_slice(&proto_fork.get_previous_version()[..4]);
current_version.copy_from_slice(&proto_fork.get_current_version()[..4]);
let fork = Fork {
previous_version,
current_version,
epoch: Epoch::from(proto_fork.get_epoch()),
};
// initialize the RPC clients
// Beacon node gRPC beacon block endpoints.
let beacon_block_client = {
let ch = ChannelBuilder::new(env.clone()).connect(&server_url);
let beacon_block_service_client = Arc::new(BeaconBlockServiceClient::new(ch));
// a wrapper around the service client to implement the beacon block node trait
Arc::new(BeaconBlockGrpcClient::new(beacon_block_service_client))
};
// Beacon node gRPC validator endpoints.
let validator_client = {
let ch = ChannelBuilder::new(env.clone()).connect(&server_url);
Arc::new(ValidatorServiceClient::new(ch))
};
//Beacon node gRPC attester endpoints.
let attestation_client = {
let ch = ChannelBuilder::new(env.clone()).connect(&server_url);
Arc::new(AttestationServiceClient::new(ch))
};
// build the validator slot clock
let slot_clock = SystemTimeSlotClock::new(
genesis_slot,
Duration::from_secs(genesis_time),
Duration::from_millis(eth2_config.spec.milliseconds_per_slot),
);
/* Generate the duties manager */
// Load generated keypairs
let keypairs = Arc::new(client_config.fetch_keys(&log)?);
let slots_per_epoch = E::slots_per_epoch();
// TODO: keypairs are randomly generated; they should be loaded from a file or generated.
// https://github.com/sigp/lighthouse/issues/160
//let keypairs = Arc::new(generate_deterministic_keypairs(8));
// Builds a mapping of Epoch -> Map(PublicKey, EpochDuty)
// where EpochDuty contains slot numbers and attestation data that each validator needs to
// produce work on.
let duties_map = RwLock::new(EpochDutiesMap::new(slots_per_epoch));
// builds a manager which maintains the list of current duties for all known validators
// and can check when a validator needs to perform a task.
let duties_manager = Arc::new(DutiesManager {
duties_map,
// these are abstract objects capable of signing
signers: keypairs,
beacon_node: validator_client,
});
let spec = Arc::new(eth2_config.spec);
Ok(Service {
fork,
slot_clock,
current_slot: RwLock::new(None),
slots_per_epoch,
spec,
duties_manager,
beacon_block_client,
attestation_client,
log,
_phantom: PhantomData,
})
}
}
impl<B: BeaconNodeDuties + 'static, S: Signer + 'static, E: EthSpec> Service<B, S, E> {
/// The execution logic that runs every slot.
// Errors are logged to output, and core execution continues unless fatal errors occur.
pub fn per_slot_execution(&self) -> error_chain::Result<()> {
/* get the new current slot and epoch */
self.update_current_slot()?;
/* check for new duties */
self.check_for_duties();
/* process any required duties for validators */
self.process_duties();
trace!(
self.log,
"Per slot execution finished";
);
Ok(())
}
/// Updates the known current slot and epoch.
fn update_current_slot(&self) -> error_chain::Result<()> {
let wall_clock_slot = self
.slot_clock
.now()
.ok_or_else::<error_chain::Error, _>(|| {
"Genesis is not in the past. Exiting.".into()
})?;
let wall_clock_epoch = wall_clock_slot.epoch(self.slots_per_epoch);
let mut current_slot = self.current_slot.write();
// this is a non-fatal error. If the slot clock repeats, the node could
// have been slow to process the previous slot and is now duplicating tasks.
// We ignore duplicated but raise a critical error.
if let Some(current_slot) = *current_slot {
if wall_clock_slot <= current_slot {
crit!(
self.log,
"The validator tried to duplicate a slot. Likely missed the previous slot"
);
return Err("Duplicate slot".into());
}
}
*current_slot = Some(wall_clock_slot);
info!(self.log, "Processing"; "slot" => wall_clock_slot.as_u64(), "epoch" => wall_clock_epoch.as_u64());
Ok(())
}
/// For all known validator keypairs, update any known duties from the beacon node.
fn check_for_duties(&self) {
let cloned_manager = self.duties_manager.clone();
let cloned_log = self.log.clone();
let current_epoch = self
.current_slot
.read()
.expect("The current slot must be updated before checking for duties")
.epoch(self.slots_per_epoch);
trace!(
self.log,
"Checking for duties";
"epoch" => current_epoch
);
// spawn a new thread separate to the runtime
// TODO: Handle thread termination/timeout
// TODO: Add duties thread back in, with channel to process duties in duty change.
// leave sequential for now.
//std::thread::spawn(move || {
// the return value is a future which returns ready.
// built to be compatible with the tokio runtime.
let _empty = cloned_manager.run_update(current_epoch, cloned_log.clone());
//});
}
/// If there are any duties to process, spawn a separate thread and perform required actions.
fn process_duties(&self) {
if let Some(work) = self.duties_manager.get_current_work(
self.current_slot
.read()
.expect("The current slot must be updated before processing duties"),
) {
trace!(
self.log,
"Processing duties";
"work_items" => work.len()
);
for (signer_index, work_type) in work {
if work_type.produce_block {
// we need to produce a block
// spawns a thread to produce a beacon block
let signers = self.duties_manager.signers.clone(); // this is an arc
let fork = self.fork.clone();
let slot = self
.current_slot
.read()
.expect("The current slot must be updated before processing duties");
let spec = self.spec.clone();
let beacon_node = self.beacon_block_client.clone();
let log = self.log.clone();
let slots_per_epoch = self.slots_per_epoch;
std::thread::spawn(move || {
info!(
log,
"Producing a block";
"validator"=> format!("{}", signers[signer_index]),
"slot"=> slot
);
let signer = &signers[signer_index];
let mut block_producer = BlockProducer {
fork,
slot,
spec,
beacon_node,
signer,
slots_per_epoch,
_phantom: PhantomData::<E>,
log,
};
block_producer.handle_produce_block();
});
}
if work_type.attestation_duty.is_some() {
// we need to produce an attestation
// spawns a thread to produce and sign an attestation
let slot = self
.current_slot
.read()
.expect("The current slot must be updated before processing duties");
let signers = self.duties_manager.signers.clone(); // this is an arc
let fork = self.fork.clone();
let spec = self.spec.clone();
let beacon_node = self.attestation_client.clone();
let log = self.log.clone();
let slots_per_epoch = self.slots_per_epoch;
std::thread::spawn(move || {
info!(
log,
"Producing an attestation";
"validator"=> format!("{}", signers[signer_index]),
"slot"=> slot
);
let signer = &signers[signer_index];
let mut attestation_producer = AttestationProducer {
fork,
duty: work_type.attestation_duty.expect("Should never be none"),
spec,
beacon_node,
signer,
slots_per_epoch,
_phantom: PhantomData::<E>,
};
attestation_producer.handle_produce_attestation(log);
});
}
}
}
}
}

21
validator_client/src/signer.rs
View File

@@ -1,21 +0,0 @@
use std::fmt::Display;
use types::{Keypair, PublicKey, Signature};
/// Signs message using an internally-maintained private key.
pub trait Signer: Display + Send + Sync + Clone {
fn sign_message(&self, message: &[u8], domain: u64) -> Option<Signature>;
/// Returns a public key for the signer object.
fn to_public(&self) -> PublicKey;
}
/* Implements Display and Signer for Keypair */
impl Signer for Keypair {
fn to_public(&self) -> PublicKey {
self.pk.clone()
}
fn sign_message(&self, message: &[u8], domain: u64) -> Option<Signature> {
Some(Signature::new(message, domain, &self.sk))
}
}

404
validator_client/src/validator_directory.rs Normal file
View File

@@ -0,0 +1,404 @@
use bls::get_withdrawal_credentials;
use deposit_contract::eth1_tx_data;
use hex;
use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use std::fs;
use std::fs::File;
use std::io::prelude::*;
use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use types::{
test_utils::generate_deterministic_keypair, ChainSpec, DepositData, Hash256, Keypair,
PublicKey, SecretKey, Signature,
};
const VOTING_KEY_PREFIX: &str = "voting";
const WITHDRAWAL_KEY_PREFIX: &str = "withdrawal";
const ETH1_DEPOSIT_DATA_FILE: &str = "eth1_deposit_data.rlp";
/// Returns the filename of a keypair file.
fn keypair_file(prefix: &str) -> String {
format!("{}_keypair", prefix)
}
/// Returns the name of the folder to be generated for a validator with the given voting key.
fn dir_name(voting_pubkey: &PublicKey) -> String {
format!("0x{}", hex::encode(voting_pubkey.as_ssz_bytes()))
}
/// Represents the files/objects for each dedicated lighthouse validator directory.
///
/// Generally lives in `~/.lighthouse/validators/`.
#[derive(Debug, Clone, PartialEq)]
pub struct ValidatorDirectory {
pub directory: PathBuf,
pub voting_keypair: Option<Keypair>,
pub withdrawal_keypair: Option<Keypair>,
pub deposit_data: Option<Vec<u8>>,
}
impl ValidatorDirectory {
/// Attempts to load a validator from the given directory, requiring only components necessary
/// for signing messages.
pub fn load_for_signing(directory: PathBuf) -> Result<Self, String> {
if !directory.exists() {
return Err(format!(
"Validator directory does not exist: {:?}",
directory
));
}
Ok(Self {
voting_keypair: Some(
load_keypair(directory.clone(), VOTING_KEY_PREFIX)
.map_err(|e| format!("Unable to get voting keypair: {}", e))?,
),
withdrawal_keypair: load_keypair(directory.clone(), WITHDRAWAL_KEY_PREFIX).ok(),
deposit_data: load_eth1_deposit_data(directory.clone()).ok(),
directory,
})
}
}
/// Load a `Keypair` from a file.
fn load_keypair(base_path: PathBuf, file_prefix: &str) -> Result<Keypair, String> {
let path = base_path.join(keypair_file(file_prefix));
if !path.exists() {
return Err(format!("Keypair file does not exist: {:?}", path));
}
let mut bytes = vec![];
File::open(&path)
.map_err(|e| format!("Unable to open keypair file: {}", e))?
.read_to_end(&mut bytes)
.map_err(|e| format!("Unable to read keypair file: {}", e))?;
SszEncodableKeypair::from_ssz_bytes(&bytes)
.map(Into::into)
.map_err(|e| format!("Unable to decode keypair: {:?}", e))
}
/// Load eth1_deposit_data from file.
fn load_eth1_deposit_data(base_path: PathBuf) -> Result<Vec<u8>, String> {
let path = base_path.join(ETH1_DEPOSIT_DATA_FILE);
if !path.exists() {
return Err(format!("Eth1 deposit data file does not exist: {:?}", path));
}
let mut bytes = vec![];
File::open(&path)
.map_err(|e| format!("Unable to open eth1 deposit data file: {}", e))?
.read_to_end(&mut bytes)
.map_err(|e| format!("Unable to read eth1 deposit data file: {}", e))?;
let string = String::from_utf8_lossy(&bytes);
if string.starts_with("0x") {
hex::decode(&string[2..])
.map_err(|e| format!("Unable to decode eth1 data file as hex: {}", e))
} else {
Err(format!("String did not start with 0x: {}", string))
}
}
/// A helper struct to allow SSZ enc/dec for a `Keypair`.
#[derive(Encode, Decode)]
struct SszEncodableKeypair {
pk: PublicKey,
sk: SecretKey,
}
impl Into<Keypair> for SszEncodableKeypair {
fn into(self) -> Keypair {
Keypair {
sk: self.sk,
pk: self.pk,
}
}
}
impl From<Keypair> for SszEncodableKeypair {
fn from(kp: Keypair) -> Self {
Self {
sk: kp.sk,
pk: kp.pk,
}
}
}
/// Builds a `ValidatorDirectory`, both in-memory and on-disk.
#[derive(Default)]
pub struct ValidatorDirectoryBuilder {
directory: Option<PathBuf>,
voting_keypair: Option<Keypair>,
withdrawal_keypair: Option<Keypair>,
amount: Option<u64>,
deposit_data: Option<Vec<u8>>,
spec: Option<ChainSpec>,
}
impl ValidatorDirectoryBuilder {
pub fn spec(mut self, spec: ChainSpec) -> Self {
self.spec = Some(spec);
self
}
pub fn full_deposit_amount(mut self) -> Result<Self, String> {
let spec = self
.spec
.as_ref()
.ok_or_else(|| "full_deposit_amount requires a spec")?;
self.amount = Some(spec.max_effective_balance);
Ok(self)
}
pub fn custom_deposit_amount(mut self, gwei: u64) -> Self {
self.amount = Some(gwei);
self
}
pub fn thread_random_keypairs(mut self) -> Self {
self.voting_keypair = Some(Keypair::random());
self.withdrawal_keypair = Some(Keypair::random());
self
}
pub fn insecure_keypairs(mut self, index: usize) -> Self {
let keypair = generate_deterministic_keypair(index);
self.voting_keypair = Some(keypair.clone());
self.withdrawal_keypair = Some(keypair);
self
}
/// Creates a validator directory in the given `base_path` (e.g., `~/.lighthouse/validators/`).
pub fn create_directory(mut self, base_path: PathBuf) -> Result<Self, String> {
let voting_keypair = self
.voting_keypair
.as_ref()
.ok_or_else(|| "directory requires a voting_keypair")?;
let directory = base_path.join(dir_name(&voting_keypair.pk));
if directory.exists() {
return Err(format!(
"Validator directory already exists: {:?}",
directory
));
}
fs::create_dir_all(&directory)
.map_err(|e| format!("Unable to create validator directory: {}", e))?;
self.directory = Some(directory);
Ok(self)
}
pub fn write_keypair_files(self) -> Result<Self, String> {
let voting_keypair = self
.voting_keypair
.clone()
.ok_or_else(|| "write_keypair_files requires a voting_keypair")?;
let withdrawal_keypair = self
.withdrawal_keypair
.clone()
.ok_or_else(|| "write_keypair_files requires a withdrawal_keypair")?;
self.save_keypair(voting_keypair, VOTING_KEY_PREFIX)?;
self.save_keypair(withdrawal_keypair, WITHDRAWAL_KEY_PREFIX)?;
Ok(self)
}
fn save_keypair(&self, keypair: Keypair, file_prefix: &str) -> Result<(), String> {
let path = self
.directory
.as_ref()
.map(|directory| directory.join(keypair_file(file_prefix)))
.ok_or_else(|| "save_keypair requires a directory")?;
if path.exists() {
return Err(format!("Keypair file already exists at: {:?}", path));
}
let mut file = File::create(&path).map_err(|e| format!("Unable to create file: {}", e))?;
// Ensure file has correct permissions.
let mut perm = file
.metadata()
.map_err(|e| format!("Unable to get file metadata: {}", e))?
.permissions();
perm.set_mode((libc::S_IWUSR | libc::S_IRUSR) as u32);
file.set_permissions(perm)
.map_err(|e| format!("Unable to set file permissions: {}", e))?;
file.write_all(&SszEncodableKeypair::from(keypair).as_ssz_bytes())
.map_err(|e| format!("Unable to write keypair to file: {}", e))?;
Ok(())
}
pub fn write_eth1_data_file(mut self) -> Result<Self, String> {
let voting_keypair = self
.voting_keypair
.as_ref()
.ok_or_else(|| "write_eth1_data_file requires a voting_keypair")?;
let withdrawal_keypair = self
.withdrawal_keypair
.as_ref()
.ok_or_else(|| "write_eth1_data_file requires a withdrawal_keypair")?;
let amount = self
.amount
.ok_or_else(|| "write_eth1_data_file requires an amount")?;
let spec = self.spec.as_ref().ok_or_else(|| "build requires a spec")?;
let path = self
.directory
.as_ref()
.map(|directory| directory.join(ETH1_DEPOSIT_DATA_FILE))
.ok_or_else(|| "write_eth1_data_filer requires a directory")?;
let deposit_data = {
let withdrawal_credentials = Hash256::from_slice(&get_withdrawal_credentials(
&withdrawal_keypair.pk,
spec.bls_withdrawal_prefix_byte,
));
let mut deposit_data = DepositData {
pubkey: voting_keypair.pk.clone().into(),
withdrawal_credentials,
amount,
signature: Signature::empty_signature().into(),
};
deposit_data.signature = deposit_data.create_signature(&voting_keypair.sk, &spec);
eth1_tx_data(&deposit_data)
.map_err(|e| format!("Unable to encode eth1 deposit tx data: {:?}", e))?
};
if path.exists() {
return Err(format!("Eth1 data file already exists at: {:?}", path));
}
File::create(&path)
.map_err(|e| format!("Unable to create file: {}", e))?
.write_all(&format!("0x{}", hex::encode(&deposit_data)).as_bytes())
.map_err(|e| format!("Unable to write eth1 data file: {}", e))?;
self.deposit_data = Some(deposit_data);
Ok(self)
}
pub fn build(self) -> Result<ValidatorDirectory, String> {
Ok(ValidatorDirectory {
directory: self.directory.ok_or_else(|| "build requires a directory")?,
voting_keypair: self.voting_keypair,
withdrawal_keypair: self.withdrawal_keypair,
deposit_data: self.deposit_data,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempdir::TempDir;
use types::{EthSpec, MinimalEthSpec};
type E = MinimalEthSpec;
#[test]
fn random_keypairs_round_trip() {
let spec = E::default_spec();
let temp_dir = TempDir::new("acc_manager").expect("should create test dir");
let created_dir = ValidatorDirectoryBuilder::default()
.spec(spec)
.full_deposit_amount()
.expect("should set full deposit amount")
.thread_random_keypairs()
.create_directory(temp_dir.path().into())
.expect("should create directory")
.write_keypair_files()
.expect("should write keypair files")
.write_eth1_data_file()
.expect("should write eth1 data file")
.build()
.expect("should build dir");
let loaded_dir = ValidatorDirectory::load_for_signing(created_dir.directory.clone())
.expect("should load directory");
assert_eq!(
created_dir, loaded_dir,
"the directory created should match the one loaded"
);
}
#[test]
fn deterministic_keypairs_round_trip() {
let spec = E::default_spec();
let temp_dir = TempDir::new("acc_manager").expect("should create test dir");
let index = 42;
let created_dir = ValidatorDirectoryBuilder::default()
.spec(spec)
.full_deposit_amount()
.expect("should set full deposit amount")
.insecure_keypairs(index)
.create_directory(temp_dir.path().into())
.expect("should create directory")
.write_keypair_files()
.expect("should write keypair files")
.write_eth1_data_file()
.expect("should write eth1 data file")
.build()
.expect("should build dir");
assert!(
created_dir.directory.exists(),
"should have created directory"
);
let mut parent = created_dir.directory.clone();
parent.pop();
assert_eq!(
parent,
PathBuf::from(temp_dir.path()),
"should have created directory ontop of base dir"
);
let expected_keypair = generate_deterministic_keypair(index);
assert_eq!(
created_dir.voting_keypair,
Some(expected_keypair.clone()),
"voting keypair should be as expected"
);
assert_eq!(
created_dir.withdrawal_keypair,
Some(expected_keypair),
"withdrawal keypair should be as expected"
);
assert!(
created_dir
.deposit_data
.clone()
.expect("should have data")
.len()
> 0,
"should have some deposit data"
);
let loaded_dir = ValidatorDirectory::load_for_signing(created_dir.directory.clone())
.expect("should load directory");
assert_eq!(
created_dir, loaded_dir,
"the directory created should match the one loaded"
);
}
}

200
validator_client/src/validator_store.rs Normal file
View File

@@ -0,0 +1,200 @@
use crate::fork_service::ForkService;
use crate::validator_directory::{ValidatorDirectory, ValidatorDirectoryBuilder};
use parking_lot::RwLock;
use rayon::prelude::*;
use slog::{error, Logger};
use slot_clock::SlotClock;
use std::collections::HashMap;
use std::fs::read_dir;
use std::iter::FromIterator;
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::Arc;
use tempdir::TempDir;
use tree_hash::TreeHash;
use types::{
Attestation, BeaconBlock, ChainSpec, Domain, Epoch, EthSpec, Fork, PublicKey, Signature,
};
#[derive(Clone)]
pub struct ValidatorStore<T, E: EthSpec> {
validators: Arc<RwLock<HashMap<PublicKey, ValidatorDirectory>>>,
spec: Arc<ChainSpec>,
log: Logger,
temp_dir: Option<Arc<TempDir>>,
fork_service: ForkService<T, E>,
_phantom: PhantomData<E>,
}
impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
pub fn load_from_disk(
base_dir: PathBuf,
spec: ChainSpec,
fork_service: ForkService<T, E>,
log: Logger,
) -> Result<Self, String> {
let validator_iter = read_dir(&base_dir)
.map_err(|e| format!("Failed to read base directory: {:?}", e))?
.filter_map(|validator_dir| {
let path = validator_dir.ok()?.path();
if path.is_dir() {
match ValidatorDirectory::load_for_signing(path.clone()) {
Ok(validator_directory) => Some(validator_directory),
Err(e) => {
error!(
log,
"Failed to load a validator directory";
"error" => e,
"path" => path.to_str(),
);
None
}
}
} else {
None
}
})
.filter_map(|validator_directory| {
validator_directory
.voting_keypair
.clone()
.map(|voting_keypair| (voting_keypair.pk, validator_directory))
});
Ok(Self {
validators: Arc::new(RwLock::new(HashMap::from_iter(validator_iter))),
spec: Arc::new(spec),
log,
temp_dir: None,
fork_service,
_phantom: PhantomData,
})
}
pub fn insecure_ephemeral_validators(
validator_indices: &[usize],
spec: ChainSpec,
fork_service: ForkService<T, E>,
log: Logger,
) -> Result<Self, String> {
let temp_dir = TempDir::new("insecure_validator")
.map_err(|e| format!("Unable to create temp dir: {:?}", e))?;
let data_dir = PathBuf::from(temp_dir.path());
let validators = validator_indices
.par_iter()
.map(|index| {
ValidatorDirectoryBuilder::default()
.spec(spec.clone())
.full_deposit_amount()?
.insecure_keypairs(*index)
.create_directory(data_dir.clone())?
.write_keypair_files()?
.write_eth1_data_file()?
.build()
})
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.filter_map(|validator_directory| {
validator_directory
.voting_keypair
.clone()
.map(|voting_keypair| (voting_keypair.pk, validator_directory))
});
Ok(Self {
validators: Arc::new(RwLock::new(HashMap::from_iter(validators))),
spec: Arc::new(spec),
log,
temp_dir: Some(Arc::new(temp_dir)),
fork_service,
_phantom: PhantomData,
})
}
pub fn voting_pubkeys(&self) -> Vec<PublicKey> {
self.validators
.read()
.iter()
.map(|(pubkey, _dir)| pubkey.clone())
.collect()
}
pub fn num_voting_validators(&self) -> usize {
self.validators.read().len()
}
fn fork(&self) -> Option<Fork> {
if self.fork_service.fork().is_none() {
error!(
self.log,
"Unable to get Fork for signing";
);
}
self.fork_service.fork()
}
pub fn randao_reveal(&self, validator_pubkey: &PublicKey, epoch: Epoch) -> Option<Signature> {
// TODO: check this against the slot clock to make sure it's not an early reveal?
self.validators
.read()
.get(validator_pubkey)
.and_then(|validator_dir| {
let voting_keypair = validator_dir.voting_keypair.as_ref()?;
let message = epoch.tree_hash_root();
let domain = self.spec.get_domain(epoch, Domain::Randao, &self.fork()?);
Some(Signature::new(&message, domain, &voting_keypair.sk))
})
}
pub fn sign_block(
&self,
validator_pubkey: &PublicKey,
mut block: BeaconBlock<E>,
) -> Option<BeaconBlock<E>> {
// TODO: check for slashing.
self.validators
.read()
.get(validator_pubkey)
.and_then(|validator_dir| {
let voting_keypair = validator_dir.voting_keypair.as_ref()?;
block.sign(&voting_keypair.sk, &self.fork()?, &self.spec);
Some(block)
})
}
pub fn sign_attestation(
&self,
validator_pubkey: &PublicKey,
validator_committee_position: usize,
attestation: &mut Attestation<E>,
) -> Option<()> {
// TODO: check for slashing.
self.validators
.read()
.get(validator_pubkey)
.and_then(|validator_dir| {
let voting_keypair = validator_dir.voting_keypair.as_ref()?;
attestation
.sign(
&voting_keypair.sk,
validator_committee_position,
&self.fork()?,
&self.spec,
)
.map_err(|e| {
error!(
self.log,
"Error whilst signing attestation";
"error" => format!("{:?}", e)
)
})
.ok()?;
Some(())
})
}
}