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

Merge branch 'unstable' into merge-unstable-to-deneb-20231005

Browse Source 
# Conflicts:
#	.github/workflows/test-suite.yml
#	Cargo.lock
#	beacon_node/execution_layer/Cargo.toml
#	beacon_node/execution_layer/src/test_utils/mock_builder.rs
#	beacon_node/execution_layer/src/test_utils/mod.rs
#	beacon_node/network/src/service/tests.rs
#	consensus/types/src/builder_bid.rs
This commit is contained in:
Jimmy Chen
2023-10-05 15:54:44 +11:00
parent c7ddf1f0b1 7d537214b7
commit c5c84f1213
48 changed files with 1791 additions and 1491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
beacon_node/beacon_chain/src/test_utils.rs
+8 -6
View File
@@ -20,8 +20,8 @@ use execution_layer::test_utils::generate_genesis_header;
use execution_layer::{
auth::JwtKey,
test_utils::{
ExecutionBlockGenerator, MockBuilder, MockBuilderServer, MockExecutionLayer,
DEFAULT_JWT_SECRET, DEFAULT_TERMINAL_BLOCK,
ExecutionBlockGenerator, MockBuilder, MockExecutionLayer, DEFAULT_JWT_SECRET,
DEFAULT_TERMINAL_BLOCK,
},
ExecutionLayer,
};
@@ -663,7 +663,10 @@ where
.execution_block_generator()
}
pub fn set_mock_builder(&mut self, beacon_url: SensitiveUrl) -> MockBuilderServer {
pub fn set_mock_builder(
&mut self,
beacon_url: SensitiveUrl,
) -> impl futures::Future<Output = ()> {
let mock_el = self
.mock_execution_layer
.as_ref()
@@ -672,7 +675,7 @@ where
let mock_el_url = SensitiveUrl::parse(mock_el.server.url().as_str()).unwrap();
// Create the builder, listening on a free port.
let (mock_builder, mock_builder_server) = MockBuilder::new_for_testing(
let (mock_builder, (addr, mock_builder_server)) = MockBuilder::new_for_testing(
mock_el_url,
beacon_url,
self.spec.clone(),
@@ -680,8 +683,7 @@ where
);
// Set the builder URL in the execution layer now that its port is known.
let builder_listen_addr = mock_builder_server.local_addr();
let port = builder_listen_addr.port();
let port = addr.port();
mock_el
.el
.set_builder_url(
beacon_node/execution_layer/Cargo.toml
-6
View File
@@ -42,12 +42,6 @@ lazy_static = { workspace = true }
ethers-core = { workspace = true }
builder_client = { path = "../builder_client" }
fork_choice = { workspace = true }
#PR: https://github.com/ralexstokes/mev-rs/pull/124
mev-rs = { git = "https://github.com/jimmygchen/mev-rs", rev = "dedc77a" }
axum = "0.6"
hyper = "0.14"
ethereum-consensus = { git = "https://github.com/ralexstokes/ethereum-consensus", rev = "12508c1f9b0c8f4bf4c5e9b6d441e840c1b37fd9" }
ssz_rs = "0.9.0"
tokio-stream = { workspace = true }
strum = { workspace = true }
keccak-hash = "0.10.0"
beacon_node/execution_layer/src/test_utils/mock_builder.rs
+445 -398
View File
@@ -1,51 +1,30 @@
use crate::test_utils::{DEFAULT_BUILDER_PAYLOAD_VALUE_WEI, DEFAULT_JWT_SECRET};
use crate::{Config, ExecutionLayer, PayloadAttributes};
use async_trait::async_trait;
use eth2::types::{BlobsBundle, BlockId, StateId, ValidatorId};
use eth2::{BeaconNodeHttpClient, Timeouts};
pub use ethereum_consensus::state_transition::Context;
use ethereum_consensus::{
crypto::{SecretKey, Signature},
primitives::{BlsPublicKey, BlsSignature, ExecutionAddress, Hash32, Root, U256},
state_transition::Error,
};
use fork_choice::ForkchoiceUpdateParameters;
use mev_rs::{
blinded_block_provider::Server as BlindedBlockProviderServer,
signing::{sign_builder_message, verify_signed_builder_message},
types::{
bellatrix::{
BuilderBid as BuilderBidBellatrix, SignedBuilderBid as SignedBuilderBidBellatrix,
},
capella::{BuilderBid as BuilderBidCapella, SignedBuilderBid as SignedBuilderBidCapella},
deneb::{BuilderBid as BuilderBidDeneb, SignedBuilderBid as SignedBuilderBidDeneb},
BidRequest, BuilderBid, ExecutionPayload as ServerPayload, SignedBlindedBeaconBlock,
SignedBuilderBid, SignedValidatorRegistration,
},
Error as MevError,
};
use parking_lot::RwLock;
use sensitive_url::SensitiveUrl;
use ssz::{Decode, Encode};
use ssz_rs::{Merkleized, SimpleSerialize};
use std::collections::HashMap;
use std::fmt::Debug;
use std::net::Ipv4Addr;
use std::future::Future;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use std::time::Duration;
use task_executor::TaskExecutor;
use tempfile::NamedTempFile;
use tree_hash::TreeHash;
use types::builder_bid::BlindedBlobsBundle;
use types::builder_bid::{
BuilderBid, BuilderBidCapella, BuilderBidDeneb, BuilderBidMerge, SignedBuilderBid,
};
use types::{
Address, BeaconState, ChainSpec, EthSpec, ExecPayload, ExecutionPayload,
ExecutionPayloadHeader, ForkName, ForkVersionedResponse, Hash256, Slot, Uint256,
ExecutionPayloadHeaderRefMut, ForkName, ForkVersionedResponse, Hash256, PublicKeyBytes,
Signature, SignedBlindedBeaconBlock, SignedRoot, SignedValidatorRegistrationData, Slot,
Uint256,
};
pub type MockBuilderServer = axum::Server<
hyper::server::conn::AddrIncoming,
axum::routing::IntoMakeService<axum::routing::Router>,
>;
use types::{ExecutionBlockHash, SecretKey};
use warp::{Filter, Rejection};
#[derive(Clone)]
pub enum Operation {
@@ -60,115 +39,156 @@ pub enum Operation {
}
impl Operation {
fn apply<B: BidStuff>(self, bid: &mut B) -> Result<(), MevError> {
fn apply<E: EthSpec, B: BidStuff<E>>(self, bid: &mut B) {
match self {
Operation::FeeRecipient(fee_recipient) => {
*bid.fee_recipient_mut() = to_ssz_rs(&fee_recipient)?
}
Operation::GasLimit(gas_limit) => *bid.gas_limit_mut() = gas_limit as u64,
Operation::Value(value) => *bid.value_mut() = to_ssz_rs(&value)?,
Operation::ParentHash(parent_hash) => *bid.parent_hash_mut() = to_ssz_rs(&parent_hash)?,
Operation::PrevRandao(prev_randao) => *bid.prev_randao_mut() = to_ssz_rs(&prev_randao)?,
Operation::BlockNumber(block_number) => *bid.block_number_mut() = block_number as u64,
Operation::Timestamp(timestamp) => *bid.timestamp_mut() = timestamp as u64,
Operation::WithdrawalsRoot(root) => *bid.withdrawals_root_mut()? = to_ssz_rs(&root)?,
Operation::FeeRecipient(fee_recipient) => bid.set_fee_recipient(fee_recipient),
Operation::GasLimit(gas_limit) => bid.set_gas_limit(gas_limit as u64),
Operation::Value(value) => bid.set_value(value),
Operation::ParentHash(parent_hash) => bid.set_parent_hash(parent_hash),
Operation::PrevRandao(prev_randao) => bid.set_prev_randao(prev_randao),
Operation::BlockNumber(block_number) => bid.set_block_number(block_number as u64),
Operation::Timestamp(timestamp) => bid.set_timestamp(timestamp as u64),
Operation::WithdrawalsRoot(root) => bid.set_withdrawals_root(root),
}
Ok(())
}
}
#[derive(Debug)]
struct Custom(String);
impl warp::reject::Reject for Custom {}
// contains functions we need for BuilderBids.. not sure what to call this
pub trait BidStuff {
fn fee_recipient_mut(&mut self) -> &mut ExecutionAddress;
fn gas_limit_mut(&mut self) -> &mut u64;
fn value_mut(&mut self) -> &mut U256;
fn parent_hash_mut(&mut self) -> &mut Hash32;
fn prev_randao_mut(&mut self) -> &mut Hash32;
fn block_number_mut(&mut self) -> &mut u64;
fn timestamp_mut(&mut self) -> &mut u64;
fn withdrawals_root_mut(&mut self) -> Result<&mut Root, MevError>;
pub trait BidStuff<E: EthSpec> {
fn set_fee_recipient(&mut self, fee_recipient_address: Address);
fn set_gas_limit(&mut self, gas_limit: u64);
fn set_value(&mut self, value: Uint256);
fn set_parent_hash(&mut self, parent_hash: Hash256);
fn set_prev_randao(&mut self, randao: Hash256);
fn set_block_number(&mut self, block_number: u64);
fn set_timestamp(&mut self, timestamp: u64);
fn set_withdrawals_root(&mut self, withdrawals_root: Hash256);
fn sign_builder_message(
&mut self,
signing_key: &SecretKey,
context: &Context,
) -> Result<BlsSignature, Error>;
fn sign_builder_message(&mut self, sk: &SecretKey, spec: &ChainSpec) -> Signature;
fn to_signed_bid(self, signature: BlsSignature) -> SignedBuilderBid;
fn to_signed_bid(self, signature: Signature) -> SignedBuilderBid<E>;
}
macro_rules! map_builder_bid {
($self_ident:ident, |$var:ident| $expr:expr) => {
match $self_ident {
BuilderBid::Bellatrix($var) => $expr,
BuilderBid::Capella($var) => $expr,
BuilderBid::Deneb($var) => $expr,
}
};
}
impl BidStuff for BuilderBid {
fn fee_recipient_mut(&mut self) -> &mut ExecutionAddress {
map_builder_bid!(self, |bid| &mut bid.header.fee_recipient)
}
fn gas_limit_mut(&mut self) -> &mut u64 {
map_builder_bid!(self, |bid| &mut bid.header.gas_limit)
}
fn value_mut(&mut self) -> &mut U256 {
map_builder_bid!(self, |bid| &mut bid.value)
}
fn parent_hash_mut(&mut self) -> &mut Hash32 {
map_builder_bid!(self, |bid| &mut bid.header.parent_hash)
}
fn prev_randao_mut(&mut self) -> &mut Hash32 {
map_builder_bid!(self, |bid| &mut bid.header.prev_randao)
}
fn block_number_mut(&mut self) -> &mut u64 {
map_builder_bid!(self, |bid| &mut bid.header.block_number)
}
fn timestamp_mut(&mut self) -> &mut u64 {
map_builder_bid!(self, |bid| &mut bid.header.timestamp)
}
fn withdrawals_root_mut(&mut self) -> Result<&mut Root, MevError> {
match self {
Self::Bellatrix(_) => Err(MevError::InvalidFork),
Self::Capella(bid) => Ok(&mut bid.header.withdrawals_root),
Self::Deneb(bid) => Ok(&mut bid.header.withdrawals_root),
impl<E: EthSpec> BidStuff<E> for BuilderBid<E> {
fn set_fee_recipient(&mut self, fee_recipient: Address) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.fee_recipient = fee_recipient;
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.fee_recipient = fee_recipient;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.fee_recipient = fee_recipient;
}
}
}
fn sign_builder_message(
&mut self,
signing_key: &SecretKey,
context: &Context,
) -> Result<Signature, Error> {
map_builder_bid!(self, |message| sign_builder_message(
message,
signing_key,
context
))
fn set_gas_limit(&mut self, gas_limit: u64) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.gas_limit = gas_limit;
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.gas_limit = gas_limit;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.gas_limit = gas_limit;
}
}
}
fn to_signed_bid(self, signature: Signature) -> SignedBuilderBid {
match self {
Self::Bellatrix(message) => {
SignedBuilderBid::Bellatrix(SignedBuilderBidBellatrix { message, signature })
fn set_value(&mut self, value: Uint256) {
*self.value_mut() = value;
}
fn set_parent_hash(&mut self, parent_hash: Hash256) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.parent_hash = ExecutionBlockHash::from_root(parent_hash);
}
Self::Capella(message) => {
SignedBuilderBid::Capella(SignedBuilderBidCapella { message, signature })
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.parent_hash = ExecutionBlockHash::from_root(parent_hash);
}
Self::Deneb(message) => {
SignedBuilderBid::Deneb(SignedBuilderBidDeneb { message, signature })
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.parent_hash = ExecutionBlockHash::from_root(parent_hash);
}
}
}
fn set_prev_randao(&mut self, prev_randao: Hash256) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.prev_randao = prev_randao;
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.prev_randao = prev_randao;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.prev_randao = prev_randao;
}
}
}
fn set_block_number(&mut self, block_number: u64) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.block_number = block_number;
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.block_number = block_number;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.block_number = block_number;
}
}
}
fn set_timestamp(&mut self, timestamp: u64) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(header) => {
header.timestamp = timestamp;
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.timestamp = timestamp;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.timestamp = timestamp;
}
}
}
fn set_withdrawals_root(&mut self, withdrawals_root: Hash256) {
match self.to_mut().header_mut() {
ExecutionPayloadHeaderRefMut::Merge(_) => {
panic!("no withdrawals before capella")
}
ExecutionPayloadHeaderRefMut::Capella(header) => {
header.withdrawals_root = withdrawals_root;
}
ExecutionPayloadHeaderRefMut::Deneb(header) => {
header.withdrawals_root = withdrawals_root;
}
}
}
fn sign_builder_message(&mut self, sk: &SecretKey, spec: &ChainSpec) -> Signature {
let domain = spec.get_builder_domain();
let message = self.signing_root(domain);
sk.sign(message)
}
fn to_signed_bid(self, signature: Signature) -> SignedBuilderBid<E> {
SignedBuilderBid {
message: self,
signature,
}
}
}
#[derive(Clone)]
@@ -176,8 +196,7 @@ pub struct MockBuilder<E: EthSpec> {
el: ExecutionLayer<E>,
beacon_client: BeaconNodeHttpClient,
spec: ChainSpec,
context: Arc<Context>,
val_registration_cache: Arc<RwLock<HashMap<BlsPublicKey, SignedValidatorRegistration>>>,
val_registration_cache: Arc<RwLock<HashMap<PublicKeyBytes, SignedValidatorRegistrationData>>>,
builder_sk: SecretKey,
operations: Arc<RwLock<Vec<Operation>>>,
invalidate_signatures: Arc<RwLock<bool>>,
@@ -189,7 +208,7 @@ impl<E: EthSpec> MockBuilder<E> {
beacon_url: SensitiveUrl,
spec: ChainSpec,
executor: TaskExecutor,
) -> (Self, MockBuilderServer) {
) -> (Self, (SocketAddr, impl Future<Output = ()>)) {
let file = NamedTempFile::new().unwrap();
let path = file.path().into();
std::fs::write(&path, hex::encode(DEFAULT_JWT_SECRET)).unwrap();
@@ -205,23 +224,14 @@ impl<E: EthSpec> MockBuilder<E> {
let el =
ExecutionLayer::from_config(config, executor.clone(), executor.log().clone()).unwrap();
// This should probably be done for all fields, we only update ones we are testing with so far.
let mut context = Context::for_mainnet();
context.terminal_total_difficulty = to_ssz_rs(&spec.terminal_total_difficulty).unwrap();
context.terminal_block_hash = to_ssz_rs(&spec.terminal_block_hash).unwrap();
context.terminal_block_hash_activation_epoch =
to_ssz_rs(&spec.terminal_block_hash_activation_epoch).unwrap();
let builder = MockBuilder::new(
el,
BeaconNodeHttpClient::new(beacon_url, Timeouts::set_all(Duration::from_secs(1))),
spec,
context,
);
let host: Ipv4Addr = Ipv4Addr::LOCALHOST;
let port = 0;
let provider = BlindedBlockProviderServer::new(host, port, builder.clone());
let server = provider.serve();
let server = serve(host, port, builder.clone()).expect("mock builder server should start");
(builder, server)
}
@@ -229,15 +239,13 @@ impl<E: EthSpec> MockBuilder<E> {
el: ExecutionLayer<E>,
beacon_client: BeaconNodeHttpClient,
spec: ChainSpec,
context: Context,
) -> Self {
let sk = SecretKey::random(&mut rand::thread_rng()).unwrap();
let sk = SecretKey::random();
Self {
el,
beacon_client,
// Should keep spec and context consistent somehow
spec,
context: Arc::new(context),
val_registration_cache: Arc::new(RwLock::new(HashMap::new())),
builder_sk: sk,
operations: Arc::new(RwLock::new(vec![])),
@@ -259,296 +267,335 @@ impl<E: EthSpec> MockBuilder<E> {
*self.invalidate_signatures.write() = false;
}
fn apply_operations<B: BidStuff>(&self, bid: &mut B) -> Result<(), MevError> {
fn apply_operations<B: BidStuff<E>>(&self, bid: &mut B) {
let mut guard = self.operations.write();
while let Some(op) = guard.pop() {
op.apply(bid)?;
op.apply(bid);
}
Ok(())
}
pub fn pubkey(&self) -> ethereum_consensus::crypto::PublicKey {
self.builder_sk.public_key()
}
}
#[async_trait]
impl<E: EthSpec> mev_rs::BlindedBlockProvider for MockBuilder<E> {
async fn register_validators(
&self,
registrations: &mut [SignedValidatorRegistration],
) -> Result<(), MevError> {
for registration in registrations {
let pubkey = registration.message.public_key.clone();
let message = &mut registration.message;
verify_signed_builder_message(
message,
&registration.signature,
&pubkey,
&self.context,
)?;
self.val_registration_cache.write().insert(
registration.message.public_key.clone(),
registration.clone(),
);
}
pub fn serve<E: EthSpec>(
listen_addr: Ipv4Addr,
listen_port: u16,
builder: MockBuilder<E>,
) -> Result<(SocketAddr, impl Future<Output = ()>), crate::test_utils::Error> {
let inner_ctx = builder.clone();
let ctx_filter = warp::any().map(move || inner_ctx.clone());
Ok(())
}
let prefix = warp::path("eth")
.and(warp::path("v1"))
.and(warp::path("builder"));
async fn fetch_best_bid(&self, bid_request: &BidRequest) -> Result<SignedBuilderBid, MevError> {
let slot = Slot::new(bid_request.slot);
let fork = self.spec.fork_name_at_slot::<E>(slot);
let signed_cached_data = self
.val_registration_cache
.read()
.get(&bid_request.public_key)
.ok_or_else(|| convert_err("missing registration"))?
.clone();
let cached_data = signed_cached_data.message;
let validators = prefix
.and(warp::path("validators"))
.and(warp::body::json())
.and(warp::path::end())
.and(ctx_filter.clone())
.and_then(
|registrations: Vec<SignedValidatorRegistrationData>, builder: MockBuilder<E>| async move {
for registration in registrations {
if !registration.verify_signature(&builder.spec) {
return Err(reject("invalid signature"));
}
builder
.val_registration_cache
.write()
.insert(registration.message.pubkey, registration);
}
Ok(warp::reply())
},
);
let head = self
.beacon_client
.get_beacon_blocks::<E>(BlockId::Head)
.await
.map_err(convert_err)?
.ok_or_else(|| convert_err("missing head block"))?;
let blinded_block = prefix
.and(warp::path("blinded_blocks"))
.and(warp::body::json())
.and(warp::path::end())
.and(ctx_filter.clone())
.and_then(
|block: SignedBlindedBeaconBlock<E>, builder: MockBuilder<E>| async move {
let slot = block.slot();
let root = match block {
SignedBlindedBeaconBlock::Base(_) | types::SignedBeaconBlock::Altair(_) => {
return Err(reject("invalid fork"));
}
SignedBlindedBeaconBlock::Merge(block) => {
block.message.body.execution_payload.tree_hash_root()
}
SignedBlindedBeaconBlock::Capella(block) => {
block.message.body.execution_payload.tree_hash_root()
}
SignedBlindedBeaconBlock::Deneb(block) => {
block.message.body.execution_payload.tree_hash_root()
}
};
let block = head.data.message();
let head_block_root = block.tree_hash_root();
let head_execution_hash = block
.body()
.execution_payload()
.map_err(convert_err)?
.block_hash();
if head_execution_hash != from_ssz_rs(&bid_request.parent_hash)? {
return Err(custom_err(format!(
"head mismatch: {} {}",
head_execution_hash, bid_request.parent_hash
)));
}
let fork_name = builder.spec.fork_name_at_slot::<E>(slot);
let payload = builder
.el
.get_payload_by_root(&root)
.ok_or_else(|| reject("missing payload for tx root"))?;
let resp = ForkVersionedResponse {
version: Some(fork_name),
data: payload,
};
let finalized_execution_hash = self
.beacon_client
.get_beacon_blocks::<E>(BlockId::Finalized)
.await
.map_err(convert_err)?
.ok_or_else(|| convert_err("missing finalized block"))?
.data
.message()
.body()
.execution_payload()
.map_err(convert_err)?
.block_hash();
let json_payload = serde_json::to_string(&resp)
.map_err(|_| reject("coudn't serialize response"))?;
Ok::<_, warp::reject::Rejection>(
warp::http::Response::builder()
.status(200)
.body(
serde_json::to_string(&json_payload)
.map_err(|_| reject("nvalid JSON"))?,
)
.unwrap(),
)
},
);
let justified_execution_hash = self
.beacon_client
.get_beacon_blocks::<E>(BlockId::Justified)
.await
.map_err(convert_err)?
.ok_or_else(|| convert_err("missing finalized block"))?
.data
.message()
.body()
.execution_payload()
.map_err(convert_err)?
.block_hash();
let status = prefix
.and(warp::path("status"))
.then(|| async { warp::reply() });
let val_index = self
.beacon_client
.get_beacon_states_validator_id(
StateId::Head,
&ValidatorId::PublicKey(from_ssz_rs(&cached_data.public_key)?),
)
.await
.map_err(convert_err)?
.ok_or_else(|| convert_err("missing validator from state"))?
.data
.index;
let fee_recipient = from_ssz_rs(&cached_data.fee_recipient)?;
let slots_since_genesis = slot.as_u64() - self.spec.genesis_slot.as_u64();
let header = prefix
.and(warp::path("header"))
.and(warp::path::param::<Slot>().or_else(|_| async { Err(reject("Invalid slot")) }))
.and(
warp::path::param::<ExecutionBlockHash>()
.or_else(|_| async { Err(reject("Invalid parent hash")) }),
)
.and(
warp::path::param::<PublicKeyBytes>()
.or_else(|_| async { Err(reject("Invalid pubkey")) }),
)
.and(warp::path::end())
.and(ctx_filter.clone())
.and_then(
|slot: Slot,
parent_hash: ExecutionBlockHash,
pubkey: PublicKeyBytes,
builder: MockBuilder<E>| async move {
let fork = builder.spec.fork_name_at_slot::<E>(slot);
let signed_cached_data = builder
.val_registration_cache
.read()
.get(&pubkey)
.ok_or_else(|| reject("missing registration"))?
.clone();
let cached_data = signed_cached_data.message;
let genesis_time = self
.beacon_client
.get_beacon_genesis()
.await
.map_err(convert_err)?
.data
.genesis_time;
let timestamp = (slots_since_genesis * self.spec.seconds_per_slot) + genesis_time;
let head_state: BeaconState<E> = self
.beacon_client
.get_debug_beacon_states(StateId::Head)
.await
.map_err(convert_err)?
.ok_or_else(|| custom_err("missing head state".to_string()))?
.data;
let prev_randao = head_state
.get_randao_mix(head_state.current_epoch())
.map_err(convert_err)?;
let expected_withdrawals = match fork {
ForkName::Base | ForkName::Altair | ForkName::Merge => None,
ForkName::Capella | ForkName::Deneb => Some(
self.beacon_client
.get_expected_withdrawals(&StateId::Head)
let head = builder
.beacon_client
.get_beacon_blocks::<E>(BlockId::Head)
.await
.unwrap()
.data,
),
};
.map_err(|_| reject("couldn't get head"))?
.ok_or_else(|| reject("missing head block"))?;
let payload_attributes = match fork {
// the withdrawals root is filled in by operations, but we supply the valid withdrawals
// first to avoid polluting the execution block generator with invalid payload attributes
// NOTE: this was part of an effort to add payload attribute uniqueness checks,
// which was abandoned because it broke too many tests in subtle ways.
ForkName::Merge | ForkName::Capella => PayloadAttributes::new(
timestamp,
*prev_randao,
fee_recipient,
expected_withdrawals,
None,
),
ForkName::Deneb => PayloadAttributes::new(
timestamp,
*prev_randao,
fee_recipient,
expected_withdrawals,
Some(head_block_root),
),
ForkName::Base | ForkName::Altair => {
return Err(MevError::InvalidFork);
}
};
let block = head.data.message();
let head_block_root = block.tree_hash_root();
let head_execution_hash = block
.body()
.execution_payload()
.map_err(|_| reject("pre-merge block"))?
.block_hash();
if head_execution_hash != parent_hash {
return Err(reject("head mismatch"));
}
self.el
.insert_proposer(slot, head_block_root, val_index, payload_attributes.clone())
.await;
let finalized_execution_hash = builder
.beacon_client
.get_beacon_blocks::<E>(BlockId::Finalized)
.await
.map_err(|_| reject("couldn't get finalized block"))?
.ok_or_else(|| reject("missing finalized block"))?
.data
.message()
.body()
.execution_payload()
.map_err(|_| reject("pre-merge block"))?
.block_hash();
let forkchoice_update_params = ForkchoiceUpdateParameters {
head_root: Hash256::zero(),
head_hash: None,
justified_hash: Some(justified_execution_hash),
finalized_hash: Some(finalized_execution_hash),
};
let justified_execution_hash = builder
.beacon_client
.get_beacon_blocks::<E>(BlockId::Justified)
.await
.map_err(|_| reject("couldn't get justified block"))?
.ok_or_else(|| reject("missing justified block"))?
.data
.message()
.body()
.execution_payload()
.map_err(|_| reject("pre-merge block"))?
.block_hash();
let (payload, _block_value, maybe_blobs_bundle): (
ExecutionPayload<E>,
Uint256,
Option<BlobsBundle<E>>,
) = self
.el
.get_full_payload_caching(
head_execution_hash,
&payload_attributes,
forkchoice_update_params,
fork,
)
.await
.map_err(convert_err)?
.into();
let val_index = builder
.beacon_client
.get_beacon_states_validator_id(StateId::Head, &ValidatorId::PublicKey(pubkey))
.await
.map_err(|_| reject("couldn't get validator"))?
.ok_or_else(|| reject("missing validator"))?
.data
.index;
let fee_recipient = cached_data.fee_recipient;
let slots_since_genesis = slot.as_u64() - builder.spec.genesis_slot.as_u64();
let header = match payload {
ExecutionPayload::Merge(payload) => ExecutionPayloadHeader::Merge((&payload).into()),
ExecutionPayload::Capella(payload) => {
ExecutionPayloadHeader::Capella((&payload).into())
}
ExecutionPayload::Deneb(payload) => ExecutionPayloadHeader::Deneb((&payload).into()),
};
let genesis_data = builder
.beacon_client
.get_beacon_genesis()
.await
.map_err(|_| reject("couldn't get beacon genesis"))?
.data;
let genesis_time = genesis_data.genesis_time;
let timestamp =
(slots_since_genesis * builder.spec.seconds_per_slot) + genesis_time;
let mut message = match fork {
ForkName::Deneb => {
let blinded_blobs: BlindedBlobsBundle<E> =
maybe_blobs_bundle.map(Into::into).unwrap_or_default();
BuilderBid::Deneb(BuilderBidDeneb {
header: to_ssz_rs(&header)?,
blinded_blobs_bundle: to_ssz_rs(&blinded_blobs)?,
value: to_ssz_rs(&Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI))?,
public_key: self.builder_sk.public_key(),
})
}
ForkName::Capella => BuilderBid::Capella(BuilderBidCapella {
header: to_ssz_rs(&header)?,
value: to_ssz_rs(&Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI))?,
public_key: self.builder_sk.public_key(),
}),
ForkName::Merge => BuilderBid::Bellatrix(BuilderBidBellatrix {
header: to_ssz_rs(&header)?,
value: to_ssz_rs(&Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI))?,
public_key: self.builder_sk.public_key(),
}),
ForkName::Base | ForkName::Altair => return Err(MevError::InvalidFork),
};
*message.gas_limit_mut() = cached_data.gas_limit;
let head_state: BeaconState<E> = builder
.beacon_client
.get_debug_beacon_states(StateId::Head)
.await
.map_err(|_| reject("couldn't get state"))?
.ok_or_else(|| reject("missing state"))?
.data;
let prev_randao = head_state
.get_randao_mix(head_state.current_epoch())
.map_err(|_| reject("couldn't get prev randao"))?;
let expected_withdrawals = match fork {
ForkName::Base | ForkName::Altair | ForkName::Merge => None,
ForkName::Capella | ForkName::Deneb => Some(
builder
.beacon_client
.get_expected_withdrawals(&StateId::Head)
.await
.unwrap()
.data,
),
};
self.apply_operations(&mut message)?;
let mut signature =
message.sign_builder_message(&self.builder_sk, self.context.as_ref())?;
let payload_attributes = match fork {
// the withdrawals root is filled in by operations, but we supply the valid withdrawals
// first to avoid polluting the execution block generator with invalid payload attributes
// NOTE: this was part of an effort to add payload attribute uniqueness checks,
// which was abandoned because it broke too many tests in subtle ways.
ForkName::Merge | ForkName::Capella => PayloadAttributes::new(
timestamp,
*prev_randao,
fee_recipient,
expected_withdrawals,
None,
),
ForkName::Deneb => PayloadAttributes::new(
timestamp,
*prev_randao,
fee_recipient,
expected_withdrawals,
Some(head_block_root),
),
ForkName::Base | ForkName::Altair => {
return Err(reject("invalid fork"));
}
};
if *self.invalidate_signatures.read() {
signature = Signature::default();
}
builder
.el
.insert_proposer(slot, head_block_root, val_index, payload_attributes.clone())
.await;
Ok(message.to_signed_bid(signature))
}
let forkchoice_update_params = ForkchoiceUpdateParameters {
head_root: Hash256::zero(),
head_hash: None,
justified_hash: Some(justified_execution_hash),
finalized_hash: Some(finalized_execution_hash),
};
async fn open_bid(
&self,
signed_block: &mut SignedBlindedBeaconBlock,
) -> Result<ServerPayload, MevError> {
let node = match signed_block {
SignedBlindedBeaconBlock::Bellatrix(block) => {
block.message.body.execution_payload_header.hash_tree_root()
}
SignedBlindedBeaconBlock::Capella(block) => {
block.message.body.execution_payload_header.hash_tree_root()
}
SignedBlindedBeaconBlock::Deneb(block_and_blobs) => block_and_blobs
.signed_blinded_block
.message
.body
.execution_payload_header
.hash_tree_root(),
}
.map_err(convert_err)?;
let (payload, _block_value, maybe_blobs_bundle): (
ExecutionPayload<E>,
Uint256,
Option<BlobsBundle<E>>,
) = builder
.el
.get_full_payload_caching(
head_execution_hash,
&payload_attributes,
forkchoice_update_params,
fork,
)
.await
.map_err(|_| reject("couldn't get payload"))?
.into();
let payload = self
.el
.get_payload_by_root(&from_ssz_rs(&node)?)
.ok_or_else(|| convert_err("missing payload for tx root"))?;
let mut message = match fork {
ForkName::Deneb => BuilderBid::Deneb(BuilderBidDeneb {
header: payload
.as_deneb()
.map_err(|_| reject("incorrect payload variant"))?
.into(),
blinded_blobs_bundle: maybe_blobs_bundle
.map(Into::into)
.unwrap_or_default(),
value: Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI),
pubkey: builder.builder_sk.public_key().compress(),
}),
ForkName::Capella => BuilderBid::Capella(BuilderBidCapella {
header: payload
.as_capella()
.map_err(|_| reject("incorrect payload variant"))?
.into(),
value: Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI),
pubkey: builder.builder_sk.public_key().compress(),
}),
ForkName::Merge => BuilderBid::Merge(BuilderBidMerge {
header: payload
.as_merge()
.map_err(|_| reject("incorrect payload variant"))?
.into(),
value: Uint256::from(DEFAULT_BUILDER_PAYLOAD_VALUE_WEI),
pubkey: builder.builder_sk.public_key().compress(),
}),
ForkName::Base | ForkName::Altair => return Err(reject("invalid fork")),
};
let fork = payload.payload_ref().fork_name();
let resp = ForkVersionedResponse {
version: Some(fork),
data: payload,
};
message.set_gas_limit(cached_data.gas_limit);
let json_payload = serde_json::to_string(&resp).map_err(convert_err)?;
serde_json::from_str(json_payload.as_str()).map_err(convert_err)
}
builder.apply_operations(&mut message);
let mut signature =
message.sign_builder_message(&builder.builder_sk, &builder.spec);
if *builder.invalidate_signatures.read() {
signature = Signature::empty();
}
let fork_name = builder
.spec
.fork_name_at_epoch(slot.epoch(E::slots_per_epoch()));
let signed_bid = SignedBuilderBid { message, signature };
let resp = ForkVersionedResponse {
version: Some(fork_name),
data: signed_bid,
};
let json_bid = serde_json::to_string(&resp)
.map_err(|_| reject("coudn't serialize signed bid"))?;
Ok::<_, Rejection>(
warp::http::Response::builder()
.status(200)
.body(json_bid)
.unwrap(),
)
},
);
let routes = warp::post()
.and(validators.or(blinded_block))
.or(warp::get().and(status).or(header))
.map(|reply| warp::reply::with_header(reply, "Server", "lighthouse-mock-builder-server"));
let (listening_socket, server) = warp::serve(routes)
.try_bind_ephemeral(SocketAddrV4::new(listen_addr, listen_port))
.expect("mock builder server should start");
Ok((listening_socket, server))
}
pub fn from_ssz_rs<T: SimpleSerialize, U: Decode>(ssz_rs_data: &T) -> Result<U, MevError> {
U::from_ssz_bytes(
ssz_rs::serialize(ssz_rs_data)
.map_err(convert_err)?
.as_ref(),
)
.map_err(convert_err)
}
pub fn to_ssz_rs<T: Encode, U: SimpleSerialize>(ssz_data: &T) -> Result<U, MevError> {
ssz_rs::deserialize::<U>(&ssz_data.as_ssz_bytes()).map_err(convert_err)
}
pub fn convert_err<E: Debug>(e: E) -> MevError {
custom_err(format!("{e:?}"))
}
// This is a bit of a hack since the `Custom` variant was removed from `mev_rs::Error`.
pub fn custom_err(s: String) -> MevError {
MevError::Consensus(ethereum_consensus::state_transition::Error::Io(
std::io::Error::new(std::io::ErrorKind::Other, s),
))
fn reject(msg: &'static str) -> Rejection {
warp::reject::custom(Custom(msg.to_string()))
}
beacon_node/execution_layer/src/test_utils/mod.rs
+1 -4
View File
@@ -29,10 +29,7 @@ pub use execution_block_generator::{
Block, ExecutionBlockGenerator,
};
pub use hook::Hook;
pub use mock_builder::{
convert_err, custom_err, from_ssz_rs, to_ssz_rs, Context as MockBuilderContext, MockBuilder,
MockBuilderServer, Operation,
};
pub use mock_builder::{MockBuilder, Operation};
pub use mock_execution_layer::MockExecutionLayer;
pub const DEFAULT_TERMINAL_DIFFICULTY: u64 = 6400;
beacon_node/http_api/src/lib.rs
+13 -7
View File
@@ -1632,13 +1632,19 @@ pub fn serve<T: BeaconChainTypes>(
|block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || {
let (block, execution_optimistic, finalized) =
block_id.blinded_block(&chain)?;
Ok(api_types::GenericResponse::from(api_types::RootData::from(
block.canonical_root(),
))
.add_execution_optimistic_finalized(execution_optimistic, finalized))
// Prioritise requests for the head block root, as it is used by some VCs (including
// the Lighthouse VC) to create sync committee messages.
let priority = if let BlockId(eth2::types::BlockId::Head) = block_id {
Priority::P0
} else {
Priority::P1
};
task_spawner.blocking_json_task(priority, move || {
let (block_root, execution_optimistic, finalized) = block_id.root(&chain)?;
Ok(
api_types::GenericResponse::from(api_types::RootData::from(block_root))
.add_execution_optimistic_finalized(execution_optimistic, finalized),
)
})
},
);
beacon_node/http_api/tests/tests.rs
+1 -5
View File
@@ -267,11 +267,7 @@ impl ApiTester {
// Start the mock builder service prior to building the chain out.
harness.runtime.task_executor.spawn(
async move {
if let Err(e) = mock_builder_server.await {
panic!("error in mock builder server: {e:?}");
}
},
async move { mock_builder_server.await },
"mock_builder_server",
);
beacon_node/lighthouse_network/src/config.rs
+16 -10
View File
@@ -11,11 +11,17 @@ use libp2p::Multiaddr;
use serde_derive::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::net::{Ipv4Addr, Ipv6Addr};
use std::num::NonZeroU16;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use types::{ForkContext, ForkName};
pub const DEFAULT_IPV4_ADDRESS: Ipv4Addr = Ipv4Addr::UNSPECIFIED;
pub const DEFAULT_TCP_PORT: u16 = 9000u16;
pub const DEFAULT_DISC_PORT: u16 = 9000u16;
pub const DEFAULT_QUIC_PORT: u16 = 9001u16;
/// The cache time is set to accommodate the circulation time of an attestation.
///
/// The p2p spec declares that we accept attestations within the following range:
@@ -59,22 +65,22 @@ pub struct Config {
pub enr_address: (Option<Ipv4Addr>, Option<Ipv6Addr>),
/// The udp ipv4 port to broadcast to peers in order to reach back for discovery.
pub enr_udp4_port: Option<u16>,
pub enr_udp4_port: Option<NonZeroU16>,
/// The quic ipv4 port to broadcast to peers in order to reach back for libp2p services.
pub enr_quic4_port: Option<u16>,
pub enr_quic4_port: Option<NonZeroU16>,
/// The tcp ipv4 port to broadcast to peers in order to reach back for libp2p services.
pub enr_tcp4_port: Option<u16>,
pub enr_tcp4_port: Option<NonZeroU16>,
/// The udp ipv6 port to broadcast to peers in order to reach back for discovery.
pub enr_udp6_port: Option<u16>,
pub enr_udp6_port: Option<NonZeroU16>,
/// The tcp ipv6 port to broadcast to peers in order to reach back for libp2p services.
pub enr_tcp6_port: Option<u16>,
pub enr_tcp6_port: Option<NonZeroU16>,
/// The quic ipv6 port to broadcast to peers in order to reach back for libp2p services.
pub enr_quic6_port: Option<u16>,
pub enr_quic6_port: Option<NonZeroU16>,
/// Target number of connected peers.
pub target_peers: usize,
@@ -304,10 +310,10 @@ impl Default for Config {
.expect("The total rate limit has been specified"),
);
let listen_addresses = ListenAddress::V4(ListenAddr {
addr: Ipv4Addr::UNSPECIFIED,
disc_port: 9000,
quic_port: 9001,
tcp_port: 9000,
addr: DEFAULT_IPV4_ADDRESS,
disc_port: DEFAULT_DISC_PORT,
quic_port: DEFAULT_QUIC_PORT,
tcp_port: DEFAULT_TCP_PORT,
});
let discv5_listen_config =
beacon_node/lighthouse_network/src/discovery/enr.rs
+30 -20
View File
@@ -158,11 +158,11 @@ pub fn create_enr_builder_from_config<T: EnrKey>(
}
if let Some(udp4_port) = config.enr_udp4_port {
builder.udp4(udp4_port);
builder.udp4(udp4_port.get());
}
if let Some(udp6_port) = config.enr_udp6_port {
builder.udp6(udp6_port);
builder.udp6(udp6_port.get());
}
if enable_libp2p {
@@ -171,35 +171,45 @@ pub fn create_enr_builder_from_config<T: EnrKey>(
// the related fields should only be added when both QUIC and libp2p are enabled
if !config.disable_quic_support {
// If we are listening on ipv4, add the quic ipv4 port.
if let Some(quic4_port) = config
.enr_quic4_port
.or_else(|| config.listen_addrs().v4().map(|v4_addr| v4_addr.quic_port))
{
builder.add_value(QUIC_ENR_KEY, &quic4_port);
if let Some(quic4_port) = config.enr_quic4_port.or_else(|| {
config
.listen_addrs()
.v4()
.and_then(|v4_addr| v4_addr.quic_port.try_into().ok())
}) {
builder.add_value(QUIC_ENR_KEY, &quic4_port.get());
}
// If we are listening on ipv6, add the quic ipv6 port.
if let Some(quic6_port) = config
.enr_quic6_port
.or_else(|| config.listen_addrs().v6().map(|v6_addr| v6_addr.quic_port))
{
builder.add_value(QUIC6_ENR_KEY, &quic6_port);
if let Some(quic6_port) = config.enr_quic6_port.or_else(|| {
config
.listen_addrs()
.v6()
.and_then(|v6_addr| v6_addr.quic_port.try_into().ok())
}) {
builder.add_value(QUIC6_ENR_KEY, &quic6_port.get());
}
}
// If the ENR port is not set, and we are listening over that ip version, use the listening port instead.
let tcp4_port = config
.enr_tcp4_port
.or_else(|| config.listen_addrs().v4().map(|v4_addr| v4_addr.tcp_port));
let tcp4_port = config.enr_tcp4_port.or_else(|| {
config
.listen_addrs()
.v4()
.and_then(|v4_addr| v4_addr.tcp_port.try_into().ok())
});
if let Some(tcp4_port) = tcp4_port {
builder.tcp4(tcp4_port);
builder.tcp4(tcp4_port.get());
}
let tcp6_port = config
.enr_tcp6_port
.or_else(|| config.listen_addrs().v6().map(|v6_addr| v6_addr.tcp_port));
let tcp6_port = config.enr_tcp6_port.or_else(|| {
config
.listen_addrs()
.v6()
.and_then(|v6_addr| v6_addr.tcp_port.try_into().ok())
});
if let Some(tcp6_port) = tcp6_port {
builder.tcp6(tcp6_port);
builder.tcp6(tcp6_port.get());
}
}
builder
beacon_node/lighthouse_network/src/discovery/mod.rs
+94 -17
View File
@@ -21,10 +21,11 @@ pub use libp2p::identity::{Keypair, PublicKey};
use enr::{ATTESTATION_BITFIELD_ENR_KEY, ETH2_ENR_KEY, SYNC_COMMITTEE_BITFIELD_ENR_KEY};
use futures::prelude::*;
use futures::stream::FuturesUnordered;
use libp2p::multiaddr::Protocol;
use libp2p::swarm::behaviour::{DialFailure, FromSwarm};
use libp2p::swarm::THandlerInEvent;
pub use libp2p::{
core::{ConnectedPoint, Multiaddr},
core::{transport::ListenerId, ConnectedPoint, Multiaddr},
identity::PeerId,
swarm::{
dummy::ConnectionHandler, ConnectionId, DialError, NetworkBehaviour, NotifyHandler,
@@ -77,6 +78,19 @@ pub struct DiscoveredPeers {
pub peers: HashMap<Enr, Option<Instant>>,
}
/// Specifies which port numbers should be modified after start of the discovery service
#[derive(Debug)]
pub struct UpdatePorts {
/// TCP port associated wih IPv4 address (if present)
pub tcp4: bool,
/// TCP port associated wih IPv6 address (if present)
pub tcp6: bool,
/// QUIC port associated wih IPv4 address (if present)
pub quic4: bool,
/// QUIC port associated wih IPv6 address (if present)
pub quic6: bool,
}
#[derive(Clone, PartialEq)]
struct SubnetQuery {
subnet: Subnet,
@@ -177,12 +191,8 @@ pub struct Discovery<TSpec: EthSpec> {
/// always false.
pub started: bool,
/// This keeps track of whether an external UDP port change should also indicate an internal
/// TCP port change. As we cannot detect our external TCP port, we assume that the external UDP
/// port is also our external TCP port. This assumption only holds if the user has not
/// explicitly set their ENR TCP port via the CLI config. The first indicates tcp4 and the
/// second indicates tcp6.
update_tcp_port: (bool, bool),
/// Specifies whether various port numbers should be updated after the discovery service has been started
update_ports: UpdatePorts,
/// Logger for the discovery behaviour.
log: slog::Logger,
@@ -300,10 +310,12 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
}
}
let update_tcp_port = (
config.enr_tcp4_port.is_none(),
config.enr_tcp6_port.is_none(),
);
let update_ports = UpdatePorts {
tcp4: config.enr_tcp4_port.is_none(),
tcp6: config.enr_tcp6_port.is_none(),
quic4: config.enr_quic4_port.is_none(),
quic6: config.enr_quic6_port.is_none(),
};
Ok(Self {
cached_enrs: LruCache::new(50),
@@ -314,7 +326,7 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
discv5,
event_stream,
started: !config.disable_discovery,
update_tcp_port,
update_ports,
log,
enr_dir,
})
@@ -555,8 +567,6 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
if let Ok(node_id) = peer_id_to_node_id(peer_id) {
// If we could convert this peer id, remove it from the DHT and ban it from discovery.
self.discv5.ban_node(&node_id, None);
// Remove the node from the routing table.
self.discv5.remove_node(&node_id);
}
for ip_address in ip_addresses {
@@ -1006,8 +1016,8 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
// Discv5 will have updated our local ENR. We save the updated version
// to disk.
if (self.update_tcp_port.0 && socket_addr.is_ipv4())
|| (self.update_tcp_port.1 && socket_addr.is_ipv6())
if (self.update_ports.tcp4 && socket_addr.is_ipv4())
|| (self.update_ports.tcp6 && socket_addr.is_ipv6())
{
// Update the TCP port in the ENR
self.discv5.update_local_enr_socket(socket_addr, true);
@@ -1036,12 +1046,79 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
FromSwarm::DialFailure(DialFailure { peer_id, error, .. }) => {
self.on_dial_failure(peer_id, error)
}
FromSwarm::NewListenAddr(ev) => {
let addr = ev.addr;
let listener_id = ev.listener_id;
trace!(self.log, "Received NewListenAddr event from swarm"; "listener_id" => ?listener_id, "addr" => ?addr);
let mut addr_iter = addr.iter();
let attempt_enr_update = match addr_iter.next() {
Some(Protocol::Ip4(_)) => match (addr_iter.next(), addr_iter.next()) {
(Some(Protocol::Tcp(port)), None) => {
if !self.update_ports.tcp4 {
debug!(self.log, "Skipping ENR update"; "multiaddr" => ?addr);
return;
}
self.update_enr_tcp_port(port)
}
(Some(Protocol::Udp(port)), Some(Protocol::QuicV1)) => {
if !self.update_ports.quic4 {
debug!(self.log, "Skipping ENR update"; "multiaddr" => ?addr);
return;
}
self.update_enr_quic_port(port)
}
_ => {
debug!(self.log, "Encountered unacceptable multiaddr for listening (unsupported transport)"; "addr" => ?addr);
return;
}
},
Some(Protocol::Ip6(_)) => match (addr_iter.next(), addr_iter.next()) {
(Some(Protocol::Tcp(port)), None) => {
if !self.update_ports.tcp6 {
debug!(self.log, "Skipping ENR update"; "multiaddr" => ?addr);
return;
}
self.update_enr_tcp_port(port)
}
(Some(Protocol::Udp(port)), Some(Protocol::QuicV1)) => {
if !self.update_ports.quic6 {
debug!(self.log, "Skipping ENR update"; "multiaddr" => ?addr);
return;
}
self.update_enr_quic_port(port)
}
_ => {
debug!(self.log, "Encountered unacceptable multiaddr for listening (unsupported transport)"; "addr" => ?addr);
return;
}
},
_ => {
debug!(self.log, "Encountered unacceptable multiaddr for listening (no IP)"; "addr" => ?addr);
return;
}
};
let local_enr: Enr = self.discv5.local_enr();
match attempt_enr_update {
Ok(_) => {
info!(self.log, "Updated local ENR"; "enr" => local_enr.to_base64(), "seq" => local_enr.seq(), "id"=> %local_enr.node_id(), "ip4" => ?local_enr.ip4(), "udp4"=> ?local_enr.udp4(), "tcp4" => ?local_enr.tcp4(), "tcp6" => ?local_enr.tcp6(), "udp6" => ?local_enr.udp6())
}
Err(e) => warn!(self.log, "Failed to update ENR"; "error" => ?e),
}
}
FromSwarm::ConnectionEstablished(_)
| FromSwarm::ConnectionClosed(_)
| FromSwarm::AddressChange(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
beacon_node/lighthouse_network/src/peer_manager/mod.rs
+2 -2
View File
@@ -415,7 +415,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
/// Reports if a peer is banned or not.
///
/// This is used to determine if we should accept incoming connections.
pub fn ban_status(&self, peer_id: &PeerId) -> BanResult {
pub fn ban_status(&self, peer_id: &PeerId) -> Option<BanResult> {
self.network_globals.peers.read().ban_status(peer_id)
}
@@ -815,7 +815,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
) -> bool {
{
let mut peerdb = self.network_globals.peers.write();
if !matches!(peerdb.ban_status(peer_id), BanResult::NotBanned) {
if peerdb.ban_status(peer_id).is_some() {
// don't connect if the peer is banned
error!(self.log, "Connection has been allowed to a banned peer"; "peer_id" => %peer_id);
}
beacon_node/lighthouse_network/src/peer_manager/network_behaviour.rs
+53 -43
View File
@@ -1,5 +1,6 @@
//! Implementation of [`NetworkBehaviour`] for the [`PeerManager`].
use std::net::IpAddr;
use std::task::{Context, Poll};
use futures::StreamExt;
@@ -8,17 +9,17 @@ use libp2p::identity::PeerId;
use libp2p::swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
use libp2p::swarm::dummy::ConnectionHandler;
use libp2p::swarm::{ConnectionId, NetworkBehaviour, PollParameters, ToSwarm};
use slog::{debug, error};
use libp2p::swarm::{ConnectionDenied, ConnectionId, NetworkBehaviour, PollParameters, ToSwarm};
use slog::{debug, error, trace};
use types::EthSpec;
use crate::discovery::enr_ext::EnrExt;
use crate::peer_manager::peerdb::BanResult;
use crate::rpc::GoodbyeReason;
use crate::types::SyncState;
use crate::{metrics, ClearDialError};
use super::peerdb::BanResult;
use super::{ConnectingType, PeerManager, PeerManagerEvent, ReportSource};
use super::{ConnectingType, PeerManager, PeerManagerEvent};
impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
type ConnectionHandler = ConnectionHandler;
@@ -169,26 +170,64 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
}
}
fn handle_pending_inbound_connection(
&mut self,
_connection_id: ConnectionId,
_local_addr: &libp2p::Multiaddr,
remote_addr: &libp2p::Multiaddr,
) -> Result<(), ConnectionDenied> {
// get the IP address to verify it's not banned.
let ip = match remote_addr.iter().next() {
Some(libp2p::multiaddr::Protocol::Ip6(ip)) => IpAddr::V6(ip),
Some(libp2p::multiaddr::Protocol::Ip4(ip)) => IpAddr::V4(ip),
_ => {
return Err(ConnectionDenied::new(format!(
"Connection to peer rejected: invalid multiaddr: {remote_addr}"
)))
}
};
if self.network_globals.peers.read().is_ip_banned(&ip) {
return Err(ConnectionDenied::new(format!(
"Connection to peer rejected: peer {ip} is banned"
)));
}
Ok(())
}
fn handle_established_inbound_connection(
&mut self,
_connection_id: ConnectionId,
_peer: PeerId,
peer_id: PeerId,
_local_addr: &libp2p::Multiaddr,
_remote_addr: &libp2p::Multiaddr,
) -> Result<libp2p::swarm::THandler<Self>, libp2p::swarm::ConnectionDenied> {
// TODO: we might want to check if we accept this peer or not in the future.
remote_addr: &libp2p::Multiaddr,
) -> Result<libp2p::swarm::THandler<Self>, ConnectionDenied> {
trace!(self.log, "Inbound connection"; "peer_id" => %peer_id, "multiaddr" => %remote_addr);
// We already checked if the peer was banned on `handle_pending_inbound_connection`.
if let Some(BanResult::BadScore) = self.ban_status(&peer_id) {
return Err(ConnectionDenied::new(
"Connection to peer rejected: peer has a bad score",
));
}
Ok(ConnectionHandler)
}
fn handle_established_outbound_connection(
&mut self,
_connection_id: ConnectionId,
_peer: PeerId,
_addr: &libp2p::Multiaddr,
peer_id: PeerId,
addr: &libp2p::Multiaddr,
_role_override: libp2p::core::Endpoint,
) -> Result<libp2p::swarm::THandler<Self>, libp2p::swarm::ConnectionDenied> {
// TODO: we might want to check if we accept this peer or not in the future.
Ok(ConnectionHandler)
trace!(self.log, "Outbound connection"; "peer_id" => %peer_id, "multiaddr" => %addr);
match self.ban_status(&peer_id) {
Some(cause) => {
error!(self.log, "Connected a banned peer. Rejecting connection"; "peer_id" => %peer_id);
Err(ConnectionDenied::new(cause))
}
None => Ok(ConnectionHandler),
}
}
}
@@ -215,10 +254,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
// increment prometheus metrics
if self.metrics_enabled {
let remote_addr = match endpoint {
ConnectedPoint::Dialer { address, .. } => address,
ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
};
let remote_addr = endpoint.get_remote_address();
match remote_addr.iter().find(|proto| {
matches!(
proto,
@@ -241,28 +277,6 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
}
// Check to make sure the peer is not supposed to be banned
match self.ban_status(&peer_id) {
// TODO: directly emit the ban event?
BanResult::BadScore => {
// This is a faulty state
error!(self.log, "Connected to a banned peer. Re-banning"; "peer_id" => %peer_id);
// Disconnect the peer.
self.goodbye_peer(&peer_id, GoodbyeReason::Banned, ReportSource::PeerManager);
// Re-ban the peer to prevent repeated errors.
self.events.push(PeerManagerEvent::Banned(peer_id, vec![]));
return;
}
BanResult::BannedIp(ip_addr) => {
// A good peer has connected to us via a banned IP address. We ban the peer and
// prevent future connections.
debug!(self.log, "Peer connected via banned IP. Banning"; "peer_id" => %peer_id, "banned_ip" => %ip_addr);
self.goodbye_peer(&peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager);
return;
}
BanResult::NotBanned => {}
}
// Count dialing peers in the limit if the peer dialed us.
let count_dialing = endpoint.is_listener();
// Check the connection limits
@@ -326,11 +340,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
// reference so that peer manager can track this peer.
self.inject_disconnect(&peer_id);
let remote_addr = match endpoint {
ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
ConnectedPoint::Dialer { address, .. } => address,
};
let remote_addr = endpoint.get_remote_address();
// Update the prometheus metrics
if self.metrics_enabled {
match remote_addr.iter().find(|proto| {
beacon_node/lighthouse_network/src/peer_manager/peerdb.rs
+52 -55
View File
@@ -3,10 +3,13 @@ use peer_info::{ConnectionDirection, PeerConnectionStatus, PeerInfo};
use rand::seq::SliceRandom;
use score::{PeerAction, ReportSource, Score, ScoreState};
use slog::{crit, debug, error, trace, warn};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::net::IpAddr;
use std::time::Instant;
use std::{cmp::Ordering, fmt::Display};
use std::{
collections::{HashMap, HashSet},
fmt::Formatter,
};
use sync_status::SyncStatus;
use types::EthSpec;
@@ -136,26 +139,18 @@ impl<TSpec: EthSpec> PeerDB<TSpec> {
}
}
/// Returns the current [`BanResult`] of the peer. This doesn't check the connection state, rather the
/// Returns the current [`BanResult`] of the peer if banned. This doesn't check the connection state, rather the
/// underlying score of the peer. A peer may be banned but still in the connected state
/// temporarily.
///
/// This is used to determine if we should accept incoming connections or not.
pub fn ban_status(&self, peer_id: &PeerId) -> BanResult {
if let Some(peer) = self.peers.get(peer_id) {
match peer.score_state() {
ScoreState::Banned => BanResult::BadScore,
_ => {
if let Some(ip) = self.ip_is_banned(peer) {
BanResult::BannedIp(ip)
} else {
BanResult::NotBanned
}
}
}
} else {
BanResult::NotBanned
}
pub fn ban_status(&self, peer_id: &PeerId) -> Option<BanResult> {
self.peers
.get(peer_id)
.and_then(|peer| match peer.score_state() {
ScoreState::Banned => Some(BanResult::BadScore),
_ => self.ip_is_banned(peer).map(BanResult::BannedIp),
})
}
/// Checks if the peer's known addresses are currently banned.
@@ -1183,23 +1178,25 @@ pub enum BanOperation {
}
/// When checking if a peer is banned, it can be banned for multiple reasons.
#[derive(Copy, Clone, Debug)]
pub enum BanResult {
/// The peer's score is too low causing it to be banned.
BadScore,
/// The peer should be banned because it is connecting from a banned IP address.
BannedIp(IpAddr),
/// The peer is not banned.
NotBanned,
}
// Helper function for unit tests
#[cfg(test)]
impl BanResult {
pub fn is_banned(&self) -> bool {
!matches!(self, BanResult::NotBanned)
impl Display for BanResult {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
BanResult::BadScore => write!(f, "Peer has a bad score"),
BanResult::BannedIp(addr) => write!(f, "Peer address: {} is banned", addr),
}
}
}
impl std::error::Error for BanResult {}
#[derive(Default)]
pub struct BannedPeersCount {
/// The number of banned peers in the database.
@@ -1852,11 +1849,11 @@ mod tests {
}
//check that ip1 and ip2 are banned but ip3-5 not
assert!(pdb.ban_status(&p1).is_banned());
assert!(pdb.ban_status(&p2).is_banned());
assert!(!pdb.ban_status(&p3).is_banned());
assert!(!pdb.ban_status(&p4).is_banned());
assert!(!pdb.ban_status(&p5).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_some());
assert!(pdb.ban_status(&p3).is_none());
assert!(pdb.ban_status(&p4).is_none());
assert!(pdb.ban_status(&p5).is_none());
//ban also the last peer in peers
let _ = pdb.report_peer(
@@ -1868,11 +1865,11 @@ mod tests {
pdb.inject_disconnect(&peers[BANNED_PEERS_PER_IP_THRESHOLD + 1]);
//check that ip1-ip4 are banned but ip5 not
assert!(pdb.ban_status(&p1).is_banned());
assert!(pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p3).is_banned());
assert!(pdb.ban_status(&p4).is_banned());
assert!(!pdb.ban_status(&p5).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_some());
assert!(pdb.ban_status(&p3).is_some());
assert!(pdb.ban_status(&p4).is_some());
assert!(pdb.ban_status(&p5).is_none());
//peers[0] gets unbanned
reset_score(&mut pdb, &peers[0]);
@@ -1880,11 +1877,11 @@ mod tests {
let _ = pdb.shrink_to_fit();
//nothing changed
assert!(pdb.ban_status(&p1).is_banned());
assert!(pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p3).is_banned());
assert!(pdb.ban_status(&p4).is_banned());
assert!(!pdb.ban_status(&p5).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_some());
assert!(pdb.ban_status(&p3).is_some());
assert!(pdb.ban_status(&p4).is_some());
assert!(pdb.ban_status(&p5).is_none());
//peers[1] gets unbanned
reset_score(&mut pdb, &peers[1]);
@@ -1892,11 +1889,11 @@ mod tests {
let _ = pdb.shrink_to_fit();
//all ips are unbanned
assert!(!pdb.ban_status(&p1).is_banned());
assert!(!pdb.ban_status(&p2).is_banned());
assert!(!pdb.ban_status(&p3).is_banned());
assert!(!pdb.ban_status(&p4).is_banned());
assert!(!pdb.ban_status(&p5).is_banned());
assert!(pdb.ban_status(&p1).is_none());
assert!(pdb.ban_status(&p2).is_none());
assert!(pdb.ban_status(&p3).is_none());
assert!(pdb.ban_status(&p4).is_none());
assert!(pdb.ban_status(&p5).is_none());
}
#[test]
@@ -1921,8 +1918,8 @@ mod tests {
}
// check ip is banned
assert!(pdb.ban_status(&p1).is_banned());
assert!(!pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_none());
// unban a peer
reset_score(&mut pdb, &peers[0]);
@@ -1930,8 +1927,8 @@ mod tests {
let _ = pdb.shrink_to_fit();
// check not banned anymore
assert!(!pdb.ban_status(&p1).is_banned());
assert!(!pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p1).is_none());
assert!(pdb.ban_status(&p2).is_none());
// unban all peers
for p in &peers {
@@ -1950,8 +1947,8 @@ mod tests {
}
// both IP's are now banned
assert!(pdb.ban_status(&p1).is_banned());
assert!(pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_some());
// unban all peers
for p in &peers {
@@ -1967,16 +1964,16 @@ mod tests {
}
// nothing is banned
assert!(!pdb.ban_status(&p1).is_banned());
assert!(!pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p1).is_none());
assert!(pdb.ban_status(&p2).is_none());
// reban last peer
let _ = pdb.report_peer(&peers[0], PeerAction::Fatal, ReportSource::PeerManager, "");
pdb.inject_disconnect(&peers[0]);
//Ip's are banned again
assert!(pdb.ban_status(&p1).is_banned());
assert!(pdb.ban_status(&p2).is_banned());
assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_some());
}
#[test]
beacon_node/lighthouse_network/src/service/behaviour.rs
-2
View File
@@ -20,8 +20,6 @@ where
AppReqId: ReqId,
TSpec: EthSpec,
{
/// Peers banned.
pub banned_peers: libp2p::allow_block_list::Behaviour<libp2p::allow_block_list::BlockedPeers>,
/// Keep track of active and pending connections to enforce hard limits.
pub connection_limits: libp2p::connection_limits::Behaviour,
/// The routing pub-sub mechanism for eth2.
beacon_node/lighthouse_network/src/service/mod.rs
+33 -9
View File
@@ -27,6 +27,7 @@ use gossipsub_scoring_parameters::{lighthouse_gossip_thresholds, PeerScoreSettin
use libp2p::bandwidth::BandwidthSinks;
use libp2p::gossipsub::{
self, IdentTopic as Topic, MessageAcceptance, MessageAuthenticity, MessageId, PublishError,
TopicScoreParams,
};
use libp2p::identify;
use libp2p::multiaddr::{Multiaddr, Protocol as MProtocol};
@@ -353,11 +354,8 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
libp2p::connection_limits::Behaviour::new(limits)
};
let banned_peers = libp2p::allow_block_list::Behaviour::default();
let behaviour = {
Behaviour {
banned_peers,
gossipsub,
eth2_rpc,
discovery,
@@ -637,6 +635,38 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
}
}
/// Remove topic weight from all topics that don't have the given fork digest.
pub fn remove_topic_weight_except(&mut self, except: [u8; 4]) {
let new_param = TopicScoreParams {
topic_weight: 0.0,
..Default::default()
};
let subscriptions = self.network_globals.gossipsub_subscriptions.read().clone();
for topic in subscriptions
.iter()
.filter(|topic| topic.fork_digest != except)
{
let libp2p_topic: Topic = topic.clone().into();
match self
.gossipsub_mut()
.set_topic_params(libp2p_topic, new_param.clone())
{
Ok(_) => debug!(self.log, "Removed topic weight"; "topic" => %topic),
Err(e) => {
warn!(self.log, "Failed to remove topic weight"; "topic" => %topic, "error" => e)
}
}
}
}
/// Returns the scoring parameters for a topic if set.
pub fn get_topic_params(&self, topic: GossipTopic) -> Option<&TopicScoreParams> {
self.swarm
.behaviour()
.gossipsub
.get_topic_params(&topic.into())
}
/// Subscribes to a gossipsub topic.
///
/// Returns `true` if the subscription was successful and `false` otherwise.
@@ -1445,15 +1475,10 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
Some(NetworkEvent::PeerDisconnected(peer_id))
}
PeerManagerEvent::Banned(peer_id, associated_ips) => {
self.swarm.behaviour_mut().banned_peers.block_peer(peer_id);
self.discovery_mut().ban_peer(&peer_id, associated_ips);
None
}
PeerManagerEvent::UnBanned(peer_id, associated_ips) => {
self.swarm
.behaviour_mut()
.banned_peers
.unblock_peer(peer_id);
self.discovery_mut().unban_peer(&peer_id, associated_ips);
None
}
@@ -1502,7 +1527,6 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
let maybe_event = match swarm_event {
SwarmEvent::Behaviour(behaviour_event) => match behaviour_event {
// Handle sub-behaviour events.
BehaviourEvent::BannedPeers(void) => void::unreachable(void),
BehaviourEvent::Gossipsub(ge) => self.inject_gs_event(ge),
BehaviourEvent::Eth2Rpc(re) => self.inject_rpc_event(re),
// Inform the peer manager about discovered peers.
beacon_node/network/src/network_beacon_processor/gossip_methods.rs
+9 -1
View File
@@ -978,9 +978,17 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore);
return None;
}
Err(BlockError::BlockIsAlreadyKnown) => {
debug!(
self.log,
"Gossip block is already known";
"block_root" => %block_root,
);
self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore);
return None;
}
Err(e @ BlockError::FutureSlot { .. })
| Err(e @ BlockError::WouldRevertFinalizedSlot { .. })
| Err(e @ BlockError::BlockIsAlreadyKnown)
| Err(e @ BlockError::NotFinalizedDescendant { .. }) => {
debug!(self.log, "Could not verify block for gossip. Ignoring the block";
"error" => %e);
beacon_node/network/src/service.rs
+34 -4
View File
@@ -215,15 +215,18 @@ pub struct NetworkService<T: BeaconChainTypes> {
}
impl<T: BeaconChainTypes> NetworkService<T> {
#[allow(clippy::type_complexity)]
pub async fn start(
async fn build(
beacon_chain: Arc<BeaconChain<T>>,
config: &NetworkConfig,
executor: task_executor::TaskExecutor,
gossipsub_registry: Option<&'_ mut Registry>,
beacon_processor_send: BeaconProcessorSend<T::EthSpec>,
beacon_processor_reprocess_tx: mpsc::Sender<ReprocessQueueMessage>,
) -> error::Result<(Arc<NetworkGlobals<T::EthSpec>>, NetworkSenders<T::EthSpec>)> {
) -> error::Result<(
NetworkService<T>,
Arc<NetworkGlobals<T::EthSpec>>,
NetworkSenders<T::EthSpec>,
)> {
let network_log = executor.log().clone();
// build the channels for external comms
let (network_senders, network_recievers) = NetworkSenders::new();
@@ -369,6 +372,28 @@ impl<T: BeaconChainTypes> NetworkService<T> {
enable_light_client_server: config.enable_light_client_server,
};
Ok((network_service, network_globals, network_senders))
}
#[allow(clippy::type_complexity)]
pub async fn start(
beacon_chain: Arc<BeaconChain<T>>,
config: &NetworkConfig,
executor: task_executor::TaskExecutor,
gossipsub_registry: Option<&'_ mut Registry>,
beacon_processor_send: BeaconProcessorSend<T::EthSpec>,
beacon_processor_reprocess_tx: mpsc::Sender<ReprocessQueueMessage>,
) -> error::Result<(Arc<NetworkGlobals<T::EthSpec>>, NetworkSenders<T::EthSpec>)> {
let (network_service, network_globals, network_senders) = Self::build(
beacon_chain,
config,
executor.clone(),
gossipsub_registry,
beacon_processor_send,
beacon_processor_reprocess_tx,
)
.await?;
network_service.spawn_service(executor);
Ok((network_globals, network_senders))
@@ -885,9 +910,10 @@ impl<T: BeaconChainTypes> NetworkService<T> {
fn update_next_fork(&mut self) {
let new_enr_fork_id = self.beacon_chain.enr_fork_id();
let new_fork_digest = new_enr_fork_id.fork_digest;
let fork_context = &self.fork_context;
if let Some(new_fork_name) = fork_context.from_context_bytes(new_enr_fork_id.fork_digest) {
if let Some(new_fork_name) = fork_context.from_context_bytes(new_fork_digest) {
info!(
self.log,
"Transitioned to new fork";
@@ -910,6 +936,10 @@ impl<T: BeaconChainTypes> NetworkService<T> {
Box::pin(next_fork_subscriptions_delay(&self.beacon_chain).into());
self.next_unsubscribe = Box::pin(Some(tokio::time::sleep(unsubscribe_delay)).into());
info!(self.log, "Network will unsubscribe from old fork gossip topics in a few epochs"; "remaining_epochs" => UNSUBSCRIBE_DELAY_EPOCHS);
// Remove topic weight from old fork topics to prevent peers that left on the mesh on
// old topics from being penalized for not sending us messages.
self.libp2p.remove_topic_weight_except(new_fork_digest);
} else {
crit!(self.log, "Unknown new enr fork id"; "new_fork_id" => ?new_enr_fork_id);
}
beacon_node/network/src/service/tests.rs
+138 -4
View File
@@ -4,14 +4,26 @@ mod tests {
use crate::persisted_dht::load_dht;
use crate::{NetworkConfig, NetworkService};
use beacon_chain::test_utils::BeaconChainHarness;
use beacon_processor::BeaconProcessorChannels;
use lighthouse_network::Enr;
use beacon_chain::BeaconChainTypes;
use beacon_processor::{BeaconProcessorChannels, BeaconProcessorConfig};
use futures::StreamExt;
use lighthouse_network::types::{GossipEncoding, GossipKind};
use lighthouse_network::{Enr, GossipTopic};
use slog::{o, Drain, Level, Logger};
use sloggers::{null::NullLoggerBuilder, Build};
use std::str::FromStr;
use std::sync::Arc;
use tokio::runtime::Runtime;
use types::MinimalEthSpec as E;
use types::{Epoch, EthSpec, ForkName, MinimalEthSpec, SubnetId};
impl<T: BeaconChainTypes> NetworkService<T> {
fn get_topic_params(
&self,
topic: GossipTopic,
) -> Option<&lighthouse_network::libp2p::gossipsub::TopicScoreParams> {
self.libp2p.get_topic_params(topic)
}
}
fn get_logger(actual_log: bool) -> Logger {
if actual_log {
@@ -35,7 +47,7 @@ mod tests {
fn test_dht_persistence() {
let log = get_logger(false);
let beacon_chain = BeaconChainHarness::builder(E)
let beacon_chain = BeaconChainHarness::builder(MinimalEthSpec)
.default_spec()
.deterministic_keypairs(8)
.fresh_ephemeral_store()
@@ -102,4 +114,126 @@ mod tests {
"should have persisted the second ENR to store"
);
}
// Test removing topic weight on old topics when a fork happens.
#[test]
fn test_removing_topic_weight_on_old_topics() {
let runtime = Arc::new(Runtime::new().unwrap());
// Capella spec
let mut spec = MinimalEthSpec::default_spec();
spec.altair_fork_epoch = Some(Epoch::new(0));
spec.bellatrix_fork_epoch = Some(Epoch::new(0));
spec.capella_fork_epoch = Some(Epoch::new(1));
// Build beacon chain.
let beacon_chain = BeaconChainHarness::builder(MinimalEthSpec)
.spec(spec.clone())
.deterministic_keypairs(8)
.fresh_ephemeral_store()
.mock_execution_layer()
.build()
.chain;
let (next_fork_name, _) = beacon_chain.duration_to_next_fork().expect("next fork");
assert_eq!(next_fork_name, ForkName::Capella);
// Build network service.
let (mut network_service, network_globals, _network_senders) = runtime.block_on(async {
let (_, exit) = exit_future::signal();
let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
let executor = task_executor::TaskExecutor::new(
Arc::downgrade(&runtime),
exit,
get_logger(false),
shutdown_tx,
);
let mut config = NetworkConfig::default();
config.set_ipv4_listening_address(std::net::Ipv4Addr::UNSPECIFIED, 21214, 21214, 21215);
config.discv5_config.table_filter = |_| true; // Do not ignore local IPs
config.upnp_enabled = false;
let beacon_processor_channels =
BeaconProcessorChannels::new(&BeaconProcessorConfig::default());
NetworkService::build(
beacon_chain.clone(),
&config,
executor.clone(),
None,
beacon_processor_channels.beacon_processor_tx,
beacon_processor_channels.work_reprocessing_tx,
)
.await
.unwrap()
});
// Subscribe to the topics.
runtime.block_on(async {
while network_globals.gossipsub_subscriptions.read().len() < 2 {
if let Some(msg) = network_service.attestation_service.next().await {
network_service.on_attestation_service_msg(msg);
}
}
});
// Make sure the service is subscribed to the topics.
let (old_topic1, old_topic2) = {
let mut subnets = SubnetId::compute_subnets_for_epoch::<MinimalEthSpec>(
network_globals.local_enr().node_id().raw().into(),
beacon_chain.epoch().unwrap(),
&spec,
)
.unwrap()
.0
.collect::<Vec<_>>();
assert_eq!(2, subnets.len());
let old_fork_digest = beacon_chain.enr_fork_id().fork_digest;
let old_topic1 = GossipTopic::new(
GossipKind::Attestation(subnets.pop().unwrap()),
GossipEncoding::SSZSnappy,
old_fork_digest,
);
let old_topic2 = GossipTopic::new(
GossipKind::Attestation(subnets.pop().unwrap()),
GossipEncoding::SSZSnappy,
old_fork_digest,
);
(old_topic1, old_topic2)
};
let subscriptions = network_globals.gossipsub_subscriptions.read().clone();
assert_eq!(2, subscriptions.len());
assert!(subscriptions.contains(&old_topic1));
assert!(subscriptions.contains(&old_topic2));
let old_topic_params1 = network_service
.get_topic_params(old_topic1.clone())
.expect("topic score params");
assert!(old_topic_params1.topic_weight > 0.0);
let old_topic_params2 = network_service
.get_topic_params(old_topic2.clone())
.expect("topic score params");
assert!(old_topic_params2.topic_weight > 0.0);
// Advance slot to the next fork
for _ in 0..MinimalEthSpec::slots_per_epoch() {
beacon_chain.slot_clock.advance_slot();
}
// Run `NetworkService::update_next_fork()`.
runtime.block_on(async {
network_service.update_next_fork();
});
// Check that topic_weight on the old topics has been zeroed.
let old_topic_params1 = network_service
.get_topic_params(old_topic1)
.expect("topic score params");
assert_eq!(0.0, old_topic_params1.topic_weight);
let old_topic_params2 = network_service
.get_topic_params(old_topic2)
.expect("topic score params");
assert_eq!(0.0, old_topic_params2.topic_weight);
}
}
beacon_node/src/config.rs
+27 -13
View File
@@ -22,6 +22,7 @@ use std::fmt::Debug;
use std::fs;
use std::net::Ipv6Addr;
use std::net::{IpAddr, Ipv4Addr, ToSocketAddrs};
use std::num::NonZeroU16;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::time::Duration;
@@ -1216,23 +1217,23 @@ pub fn set_network_config(
if let Some(enr_udp_port_str) = cli_args.value_of("enr-udp-port") {
config.enr_udp4_port = Some(
enr_udp_port_str
.parse::<u16>()
.map_err(|_| format!("Invalid discovery port: {}", enr_udp_port_str))?,
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR discovery port: {}", enr_udp_port_str))?,
);
}
if let Some(enr_quic_port_str) = cli_args.value_of("enr-quic-port") {
config.enr_quic4_port = Some(
enr_quic_port_str
.parse::<u16>()
.map_err(|_| format!("Invalid quic port: {}", enr_quic_port_str))?,
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR quic port: {}", enr_quic_port_str))?,
);
}
if let Some(enr_tcp_port_str) = cli_args.value_of("enr-tcp-port") {
config.enr_tcp4_port = Some(
enr_tcp_port_str
.parse::<u16>()
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR TCP port: {}", enr_tcp_port_str))?,
);
}
@@ -1240,23 +1241,23 @@ pub fn set_network_config(
if let Some(enr_udp_port_str) = cli_args.value_of("enr-udp6-port") {
config.enr_udp6_port = Some(
enr_udp_port_str
.parse::<u16>()
.map_err(|_| format!("Invalid discovery port: {}", enr_udp_port_str))?,
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR discovery port: {}", enr_udp_port_str))?,
);
}
if let Some(enr_quic_port_str) = cli_args.value_of("enr-quic6-port") {
config.enr_quic6_port = Some(
enr_quic_port_str
.parse::<u16>()
.map_err(|_| format!("Invalid quic port: {}", enr_quic_port_str))?,
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR quic port: {}", enr_quic_port_str))?,
);
}
if let Some(enr_tcp_port_str) = cli_args.value_of("enr-tcp6-port") {
config.enr_tcp6_port = Some(
enr_tcp_port_str
.parse::<u16>()
.parse::<NonZeroU16>()
.map_err(|_| format!("Invalid ENR TCP port: {}", enr_tcp_port_str))?,
);
}
@@ -1264,25 +1265,38 @@ pub fn set_network_config(
if cli_args.is_present("enr-match") {
// Match the IP and UDP port in the ENR.
// Set the ENR address to localhost if the address is unspecified.
if let Some(ipv4_addr) = config.listen_addrs().v4().cloned() {
// ensure the port is valid to be advertised
let disc_port = ipv4_addr
.disc_port
.try_into()
.map_err(|_| "enr-match can only be used with non-zero listening ports")?;
// Set the ENR address to localhost if the address is unspecified.
let ipv4_enr_addr = if ipv4_addr.addr == Ipv4Addr::UNSPECIFIED {
Ipv4Addr::LOCALHOST
} else {
ipv4_addr.addr
};
config.enr_address.0 = Some(ipv4_enr_addr);
config.enr_udp4_port = Some(ipv4_addr.disc_port);
config.enr_udp4_port = Some(disc_port);
}
if let Some(ipv6_addr) = config.listen_addrs().v6().cloned() {
// ensure the port is valid to be advertised
let disc_port = ipv6_addr
.disc_port
.try_into()
.map_err(|_| "enr-match can only be used with non-zero listening ports")?;
// Set the ENR address to localhost if the address is unspecified.
let ipv6_enr_addr = if ipv6_addr.addr == Ipv6Addr::UNSPECIFIED {
Ipv6Addr::LOCALHOST
} else {
ipv6_addr.addr
};
config.enr_address.1 = Some(ipv6_enr_addr);
config.enr_udp6_port = Some(ipv6_addr.disc_port);
config.enr_udp6_port = Some(disc_port);
}
}