cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Add persistent network identification

Browse Source
This commit is contained in:
Age Manning 2019-07-01 16:38:42 +10:00
parent 7dc5e2f959
commit af28d5e20c
No known key found for this signature in database
GPG Key ID: 05EED64B79E06A93
12 changed files with 220 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
account_manager/Cargo.toml
View File

@ -13,3 +13,4 @@ slog-async = "^2.3.0"
validator_client = { path = "../validator_client" }
types = { path = "../eth2/types" }
eth2_config = { path = "../eth2/utils/eth2_config" }
dirs = "2.0.1"

19
account_manager/src/main.rs
View File

@ -1,7 +1,7 @@
use bls::Keypair;
use clap::{App, Arg, SubCommand};
use eth2_config::get_data_dir;
use slog::{crit, debug, info, o, Drain};
use std::fs;
use std::path::PathBuf;
use types::test_utils::generate_deterministic_keypair;
use validator_client::Config as ValidatorClientConfig;
@ -61,13 +61,22 @@ fn main() {
)
.get_matches();
let data_dir = match get_data_dir(&matches, PathBuf::from(DEFAULT_DATA_DIR)) {
Ok(dir) => dir,
let mut default_dir = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
default_dir.push(DEFAULT_DATA_DIR);
let data_dir = &matches
.value_of("datadir")
.and_then(|v| Some(PathBuf::from(v)))
.unwrap_or_else(|| PathBuf::from(default_dir));
// create the directory if needed
match fs::create_dir_all(&data_dir) {
Ok(_) => {}
Err(e) => {
crit!(log, "Failed to initialize data dir"; "error" => format!("{:?}", e));
crit!(log, "Failed to initialize data dir"; "error" => format!("{}", e));
return;
}
};
}
let mut client_config = ValidatorClientConfig::default();

1
beacon_node/Cargo.toml
View File

@ -23,3 +23,4 @@ futures = "0.1.25"
exit-future = "0.1.3"
state_processing = { path = "../eth2/state_processing" }
env_logger = "0.6.1"
dirs = "2.0.1"

7
beacon_node/eth2-libp2p/Cargo.toml
View File

@ -7,9 +7,9 @@ edition = "2018"
[dependencies]
beacon_chain = { path = "../beacon_chain" }
clap = "2.32.0"
# SigP repository
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "8ff9f2001de0aea1175c1442f22bfbf382dc0e8b" }
enr = { git = "https://github.com/SigP/rust-libp2p/", rev = "8ff9f2001de0aea1175c1442f22bfbf382dc0e8b", features = ["serde"] }
#SigP repository
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "be5710bbde69d8c5be732c13ba64239e2f370a7b" }
enr = { git = "https://github.com/SigP/rust-libp2p/", rev = "be5710bbde69d8c5be732c13ba64239e2f370a7b", features = ["serde"] }
types = { path = "../../eth2/types" }
serde = "1.0"
serde_derive = "1.0"
@ -21,3 +21,4 @@ tokio = "0.1.16"
futures = "0.1.25"
error-chain = "0.12.0"
tokio-timer = "0.2.10"
dirs = "2.0.1"

14
beacon_node/eth2-libp2p/src/config.rs
View File

@ -2,6 +2,7 @@ use clap::ArgMatches;
use enr::Enr;
use libp2p::gossipsub::{GossipsubConfig, GossipsubConfigBuilder};
use serde_derive::{Deserialize, Serialize};
use std::path::PathBuf;
use std::time::Duration;
/// The beacon node topic string to subscribe to.
@ -14,6 +15,9 @@ pub const SHARD_TOPIC_PREFIX: &str = "shard";
#[serde(default)]
/// Network configuration for lighthouse.
pub struct Config {
/// Data directory where node's keyfile is stored
pub network_dir: PathBuf,
/// IP address to listen on.
pub listen_address: std::net::IpAddr,
@ -46,7 +50,11 @@ pub struct Config {
impl Default for Config {
/// Generate a default network configuration.
fn default() -> Self {
let mut network_dir = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
network_dir.push(".lighthouse");
network_dir.push("network");
Config {
network_dir,
listen_address: "127.0.0.1".parse().expect("vaild ip address"),
libp2p_port: 9000,
discovery_address: "127.0.0.1".parse().expect("valid ip address"),
@ -72,6 +80,12 @@ impl Config {
}
pub fn apply_cli_args(&mut self, args: &ArgMatches) -> Result<(), String> {
dbg!(self.network_dir.clone());
if let Some(dir) = args.value_of("datadir") {
self.network_dir = PathBuf::from(dir).join("network");
};
dbg!(self.network_dir.clone());
if let Some(listen_address_str) = args.value_of("listen-address") {
let listen_address = listen_address_str
.parse()

103
beacon_node/eth2-libp2p/src/discovery.rs
View File

@ -1,7 +1,7 @@
use crate::{error, NetworkConfig};
/// This manages the discovery and management of peers.
///
/// Currently using Kademlia for peer discovery.
/// Currently using discv5 for peer discovery.
///
use futures::prelude::*;
use libp2p::core::swarm::{
@ -13,6 +13,9 @@ use libp2p::enr::{Enr, EnrBuilder, NodeId};
use libp2p::multiaddr::Protocol;
use slog::{debug, info, o, warn};
use std::collections::HashSet;
use std::fs::File;
use std::io::prelude::*;
use std::str::FromStr;
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;
@ -21,6 +24,8 @@ use tokio_timer::Delay;
const MAX_TIME_BETWEEN_PEER_SEARCHES: u64 = 60;
/// Initial delay between peer searches.
const INITIAL_SEARCH_DELAY: u64 = 5;
/// Local ENR storage filename.
const ENR_FILENAME: &str = "enr.dat";
/// Lighthouse discovery behaviour. This provides peer management and discovery using the Discv5
/// libp2p protocol.
@ -53,28 +58,22 @@ pub struct Discovery<TSubstream> {
impl<TSubstream> Discovery<TSubstream> {
pub fn new(
local_key: &Keypair,
net_conf: &NetworkConfig,
config: &NetworkConfig,
log: &slog::Logger,
) -> error::Result<Self> {
let log = log.new(o!("Service" => "Libp2p-Discovery"));
// Build the local ENR.
// Note: Discovery should update the ENR record's IP to the external IP as seen by the
// majority of our peers.
// checks if current ENR matches that found on disk
let local_enr = load_enr(local_key, config, &log)?;
let local_enr = EnrBuilder::new()
.ip(net_conf.discovery_address.into())
.tcp(net_conf.libp2p_port)
.udp(net_conf.discovery_port)
.build(&local_key)
.map_err(|e| format!("Could not build Local ENR: {:?}", e))?;
info!(log, "Local ENR: {}", local_enr.to_base64());
debug!(log, "Local Node Id: {}", local_enr.node_id());
let mut discovery = Discv5::new(local_enr, local_key.clone(), net_conf.listen_address)
let mut discovery = Discv5::new(local_enr, local_key.clone(), config.listen_address)
.map_err(|e| format!("Discv5 service failed: {:?}", e))?;
// Add bootnodes to routing table
for bootnode_enr in net_conf.boot_nodes.clone() {
for bootnode_enr in config.boot_nodes.clone() {
debug!(
log,
"Adding node to routing table: {}",
@ -85,10 +84,10 @@ impl<TSubstream> Discovery<TSubstream> {
Ok(Self {
connected_peers: HashSet::new(),
max_peers: net_conf.max_peers,
max_peers: config.max_peers,
peer_discovery_delay: Delay::new(Instant::now()),
past_discovery_delay: INITIAL_SEARCH_DELAY,
tcp_port: net_conf.libp2p_port,
tcp_port: config.libp2p_port,
discovery,
log,
})
@ -238,3 +237,77 @@ where
Async::NotReady
}
}
/// Loads an ENR from file if it exists and matches the current NodeId and sequence number. If none
/// exists, generates a new one.
///
/// If an ENR exists, with the same NodeId and IP addresses, we use the disk-generated one as it's
/// ENR sequence will be equal or higher than a newly generated one.
fn load_enr(
local_key: &Keypair,
config: &NetworkConfig,
log: &slog::Logger,
) -> Result<Enr, String> {
// Build the local ENR.
// Note: Discovery should update the ENR record's IP to the external IP as seen by the
// majority of our peers.
let mut local_enr = EnrBuilder::new()
.ip(config.discovery_address.into())
.tcp(config.libp2p_port)
.udp(config.discovery_port)
.build(&local_key)
.map_err(|e| format!("Could not build Local ENR: {:?}", e))?;
let enr_f = config.network_dir.join(ENR_FILENAME);
if let Ok(mut enr_file) = File::open(enr_f.clone()) {
let mut enr_string = String::new();
match enr_file.read_to_string(&mut enr_string) {
Err(_) => debug!(log, "Could not read ENR from file"),
Ok(_) => {
match Enr::from_str(&enr_string) {
Ok(enr) => {
debug!(log, "ENR found in file: {:?}", enr_f);
if enr.node_id() == local_enr.node_id() {
if enr.ip() == config.discovery_address.into()
&& enr.tcp() == Some(config.libp2p_port)
&& enr.udp() == Some(config.discovery_port)
{
debug!(log, "ENR loaded from file");
// the stored ENR has the same configuration, use it
return Ok(enr);
}
// same node id, different configuration - update the sequence number
let new_seq_no = enr.seq().checked_add(1).ok_or_else(|| "ENR sequence number on file is too large. Remove it to generate a new NodeId")?;
local_enr.set_seq(new_seq_no, local_key).map_err(|e| {
format!("Could not update ENR sequence number: {:?}", e)
})?;
debug!(log, "ENR sequence number increased to: {}", new_seq_no);
}
}
Err(e) => {
warn!(log, "ENR from file could not be decoded: {:?}", e);
}
}
}
}
}
// write ENR to disk
let _ = std::fs::create_dir_all(&config.network_dir);
match File::create(enr_f.clone())
.and_then(|mut f| f.write_all(&local_enr.to_base64().as_bytes()))
{
Ok(_) => {
debug!(log, "ENR written to disk");
}
Err(e) => {
warn!(
log,
"Could not write ENR to file: {:?}. Error: {}", enr_f, e
);
}
}
Ok(local_enr)
}

65
beacon_node/eth2-libp2p/src/service.rs
View File

@ -8,22 +8,25 @@ use crate::{BEACON_ATTESTATION_TOPIC, BEACON_PUBSUB_TOPIC};
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
identity,
identity::Keypair,
multiaddr::Multiaddr,
muxing::StreamMuxerBox,
nodes::Substream,
transport::boxed::Boxed,
upgrade::{InboundUpgradeExt, OutboundUpgradeExt},
};
use libp2p::identify::protocol::IdentifyInfo;
use libp2p::{core, secio, PeerId, Swarm, Transport};
use slog::{debug, info, trace, warn};
use std::fs::File;
use std::io::prelude::*;
use std::io::{Error, ErrorKind};
use std::time::Duration;
type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;
const NETWORK_KEY_FILENAME: &str = "key";
/// The configuration and state of the libp2p components for the beacon node.
pub struct Service {
/// The libp2p Swarm handler.
@ -39,9 +42,9 @@ impl Service {
pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
debug!(log, "Network-libp2p Service starting");
// TODO: Save and recover node key from disk
// TODO: Currently using secp256k1 keypairs - currently required for discv5
let local_private_key = identity::Keypair::generate_secp256k1();
// load the private key from CLI flag, disk or generate a new one
let local_private_key = load_private_key(&config, &log);
let local_peer_id = PeerId::from(local_private_key.public());
info!(log, "Local peer id: {:?}", local_peer_id);
@ -142,7 +145,7 @@ impl Stream for Service {
/// The implementation supports TCP/IP, WebSockets over TCP/IP, secio as the encryption layer, and
/// mplex or yamux as the multiplexing layer.
fn build_transport(local_private_key: identity::Keypair) -> Boxed<(PeerId, StreamMuxerBox), Error> {
fn build_transport(local_private_key: Keypair) -> Boxed<(PeerId, StreamMuxerBox), Error> {
// TODO: The Wire protocol currently doesn't specify encryption and this will need to be customised
// in the future.
let transport = libp2p::tcp::TcpConfig::new();
@ -179,8 +182,6 @@ pub enum Libp2pEvent {
RPC(PeerId, RPCEvent),
/// Initiated the connection to a new peer.
PeerDialed(PeerId),
/// Received information about a peer on the network.
Identified(PeerId, Box<IdentifyInfo>),
/// Received pubsub message.
PubsubMessage {
source: PeerId,
@ -188,3 +189,51 @@ pub enum Libp2pEvent {
message: Box<PubsubMessage>,
},
}
/// Loads a private key from disk. If this fails, a new key is
/// generated and is then saved to disk.
///
/// Currently only secp256k1 keys are allowed, as these are the only keys supported by discv5.
fn load_private_key(config: &NetworkConfig, log: &slog::Logger) -> Keypair {
// TODO: Currently using secp256k1 keypairs - currently required for discv5
// check for key from disk
let network_key_f = config.network_dir.join(NETWORK_KEY_FILENAME);
if let Ok(mut network_key_file) = File::open(network_key_f.clone()) {
let mut key_bytes: Vec<u8> = Vec::with_capacity(36);
match network_key_file.read_to_end(&mut key_bytes) {
Err(_) => debug!(log, "Could not read network key file"),
Ok(_) => {
// only accept secp256k1 keys for now
if let Ok(secret_key) =
libp2p::core::identity::secp256k1::SecretKey::from_bytes(&mut key_bytes)
{
let kp: libp2p::core::identity::secp256k1::Keypair = secret_key.into();
debug!(log, "Loaded network key from disk.");
return Keypair::Secp256k1(kp);
} else {
debug!(log, "Network key file is not a valid secp256k1 key");
}
}
}
}
// if a key could not be loaded from disk, generate a new one and save it
let local_private_key = Keypair::generate_secp256k1();
if let Keypair::Secp256k1(key) = local_private_key.clone() {
let _ = std::fs::create_dir_all(&config.network_dir);
match File::create(network_key_f.clone())
.and_then(|mut f| f.write_all(&key.secret().to_bytes()))
{
Ok(_) => {
debug!(log, "New network key generated and written to disk");
}
Err(e) => {
warn!(
log,
"Could not write node key to file: {:?}. Error: {}", network_key_f, e
);
}
}
}
local_private_key
}

6
beacon_node/network/src/service.rs
View File

@ -126,12 +126,6 @@ fn network_service(
.send(HandlerMessage::PeerDialed(peer_id))
.map_err(|_| "failed to send rpc to handler")?;
}
Libp2pEvent::Identified(peer_id, info) => {
debug!(
log,
"We have identified peer: {:?} with {:?}", peer_id, info
);
}
Libp2pEvent::PubsubMessage {
source, message, ..
} => {

41
beacon_node/src/main.rs
View File

@ -3,8 +3,9 @@ mod run;
use clap::{App, Arg};
use client::{ClientConfig, Eth2Config};
use env_logger::{Builder, Env};
use eth2_config::{get_data_dir, read_from_file, write_to_file};
use eth2_config::{read_from_file, write_to_file};
use slog::{crit, o, Drain, Level};
use std::fs;
use std::path::PathBuf;
pub const DEFAULT_DATA_DIR: &str = ".lighthouse";
@ -27,7 +28,6 @@ fn main() {
.value_name("DIR")
.help("Data directory for keys and databases.")
.takes_value(true)
.default_value(DEFAULT_DATA_DIR),
)
// network related arguments
.arg(
@ -69,7 +69,7 @@ fn main() {
Arg::with_name("discovery-address")
.long("discovery-address")
.value_name("Address")
.help("The address to broadcast to other peers on how to reach this node.")
.help("The IP address to broadcast to other peers on how to reach this node.")
.takes_value(true),
)
// rpc related arguments
@ -159,15 +159,24 @@ fn main() {
_ => drain.filter_level(Level::Info),
};
let logger = slog::Logger::root(drain.fuse(), o!());
let log = slog::Logger::root(drain.fuse(), o!());
let data_dir = match get_data_dir(&matches, PathBuf::from(DEFAULT_DATA_DIR)) {
Ok(dir) => dir,
let mut default_dir = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
default_dir.push(DEFAULT_DATA_DIR);
let data_dir = &matches
.value_of("datadir")
.and_then(|v| Some(PathBuf::from(v)))
.unwrap_or_else(|| PathBuf::from(default_dir));
// create the directory if needed
match fs::create_dir_all(&data_dir) {
Ok(_) => {}
Err(e) => {
crit!(logger, "Failed to initialize data dir"; "error" => format!("{:?}", e));
crit!(log, "Failed to initialize data dir"; "error" => format!("{}", e));
return;
}
};
}
let client_config_path = data_dir.join(CLIENT_CONFIG_FILENAME);
@ -179,13 +188,13 @@ fn main() {
Ok(None) => {
let default = ClientConfig::default();
if let Err(e) = write_to_file(client_config_path, &default) {
crit!(logger, "Failed to write default ClientConfig to file"; "error" => format!("{:?}", e));
crit!(log, "Failed to write default ClientConfig to file"; "error" => format!("{:?}", e));
return;
}
default
}
Err(e) => {
crit!(logger, "Failed to load a ChainConfig file"; "error" => format!("{:?}", e));
crit!(log, "Failed to load a ChainConfig file"; "error" => format!("{:?}", e));
return;
}
};
@ -197,7 +206,7 @@ fn main() {
match client_config.apply_cli_args(&matches) {
Ok(()) => (),
Err(s) => {
crit!(logger, "Failed to parse ClientConfig CLI arguments"; "error" => s);
crit!(log, "Failed to parse ClientConfig CLI arguments"; "error" => s);
return;
}
};
@ -216,13 +225,13 @@ fn main() {
_ => unreachable!(), // Guarded by slog.
};
if let Err(e) = write_to_file(eth2_config_path, &default) {
crit!(logger, "Failed to write default Eth2Config to file"; "error" => format!("{:?}", e));
crit!(log, "Failed to write default Eth2Config to file"; "error" => format!("{:?}", e));
return;
}
default
}
Err(e) => {
crit!(logger, "Failed to load/generate an Eth2Config"; "error" => format!("{:?}", e));
crit!(log, "Failed to load/generate an Eth2Config"; "error" => format!("{:?}", e));
return;
}
};
@ -231,13 +240,13 @@ fn main() {
match eth2_config.apply_cli_args(&matches) {
Ok(()) => (),
Err(s) => {
crit!(logger, "Failed to parse Eth2Config CLI arguments"; "error" => s);
crit!(log, "Failed to parse Eth2Config CLI arguments"; "error" => s);
return;
}
};
match run::run_beacon_node(client_config, eth2_config, &logger) {
match run::run_beacon_node(client_config, eth2_config, &log) {
Ok(_) => {}
Err(e) => crit!(logger, "Beacon node failed to start"; "reason" => format!("{:}", e)),
Err(e) => crit!(log, "Beacon node failed to start"; "reason" => format!("{:}", e)),
}
}

13
eth2/utils/eth2_config/src/lib.rs
View File

@ -1,6 +1,5 @@
use clap::ArgMatches;
use serde_derive::{Deserialize, Serialize};
use std::fs;
use std::fs::File;
use std::io::prelude::*;
use std::path::PathBuf;
@ -105,15 +104,3 @@ where
Ok(None)
}
}
pub fn get_data_dir(args: &ArgMatches, default_data_dir: PathBuf) -> Result<PathBuf, &'static str> {
if let Some(data_dir) = args.value_of("data_dir") {
Ok(PathBuf::from(data_dir))
} else {
let path = dirs::home_dir()
.ok_or_else(|| "Unable to locate home directory")?
.join(&default_data_dir);
fs::create_dir_all(&path).map_err(|_| "Unable to create data_dir")?;
Ok(path)
}
}

1
validator_client/Cargo.toml
View File

@ -34,3 +34,4 @@ toml = "^0.5"
error-chain = "0.12.0"
bincode = "^1.1.2"
futures = "0.1.25"
dirs = "2.0.1"

20
validator_client/src/main.rs
View File

@ -9,9 +9,10 @@ mod signer;
use crate::config::Config as ValidatorClientConfig;
use crate::service::Service as ValidatorService;
use clap::{App, Arg};
use eth2_config::{get_data_dir, read_from_file, write_to_file, Eth2Config};
use eth2_config::{read_from_file, write_to_file, Eth2Config};
use protos::services_grpc::ValidatorServiceClient;
use slog::{crit, error, info, o, Drain};
use std::fs;
use std::path::PathBuf;
use types::{Keypair, MainnetEthSpec, MinimalEthSpec};
@ -66,13 +67,22 @@ fn main() {
)
.get_matches();
let data_dir = match get_data_dir(&matches, PathBuf::from(DEFAULT_DATA_DIR)) {
Ok(dir) => dir,
let mut default_dir = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
default_dir.push(DEFAULT_DATA_DIR);
let data_dir = &matches
.value_of("datadir")
.and_then(|v| Some(PathBuf::from(v)))
.unwrap_or_else(|| PathBuf::from(default_dir));
// create the directory if needed
match fs::create_dir_all(&data_dir) {
Ok(_) => {}
Err(e) => {
crit!(log, "Failed to initialize data dir"; "error" => format!("{:?}", e));
crit!(log, "Failed to initialize data dir"; "error" => format!("{}", e));
return;
}
};
}
let client_config_path = data_dir.join(CLIENT_CONFIG_FILENAME);