feat(discv5): add crate for interfacing reth network and sigp/discv5 (#7336)

Co-authored-by: Dan Cline <6798349+Rjected@users.noreply.github.com>
Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>

crates/net/discv5/Cargo.toml

@@ -0,0 +1,45 @@
+[package]
+name = "reth-discv5"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+description = "Ethereum peer discovery V5"
+
+[lints]
+workspace = true
+
+[dependencies]
+# reth
+reth-primitives.workspace = true
+reth-metrics.workspace = true
+
+# ethereum
+alloy-rlp.workspace = true
+rlp = "0.5.2"
+discv5 = { workspace = true, features = ["libp2p"] }
+enr = { workspace = true, default-features = false, features = ["rust-secp256k1"] }
+multiaddr = { version = "0.18", default-features = false }
+libp2p-identity = "0.2"
+secp256k1.workspace = true
+
+# async/futures
+tokio.workspace = true
+futures.workspace = true
+
+# io
+rand.workspace = true
+
+# misc
+derive_more.workspace = true
+tracing.workspace = true
+thiserror.workspace = true
+itertools.workspace = true
+metrics.workspace = true
+
+[dev-dependencies]
+reth-tracing.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"] }
+secp256k1 = { workspace = true, features = ["rand-std"] }

crates/net/discv5/README.md

@@ -0,0 +1,3 @@
+# Discv5
+
+Thin wrapper around sigp/discv5.

crates/net/discv5/src/config.rs

@@ -0,0 +1,326 @@
+//! Wrapper around [`discv5::Config`].
+
+use std::{
+    collections::HashSet,
+    net::{IpAddr, SocketAddr},
+};
+
+use derive_more::Display;
+use discv5::ListenConfig;
+use multiaddr::{Multiaddr, Protocol};
+use reth_primitives::{Bytes, ForkId, NodeRecord, MAINNET};
+
+use crate::{enr::discv4_id_to_multiaddr_id, filter::MustNotIncludeKeys};
+
+/// L1 EL
+pub const ETH: &[u8] = b"eth";
+/// L1 CL
+pub const ETH2: &[u8] = b"eth2";
+/// Optimism
+pub const OPSTACK: &[u8] = b"opstack";
+
+/// Default interval in seconds at which to run a self-lookup up query.
+///
+/// Default is 60 seconds.
+const DEFAULT_SECONDS_LOOKUP_INTERVAL: u64 = 60;
+
+/// Optimism mainnet and base mainnet boot nodes.
+const BOOT_NODES_OP_MAINNET_AND_BASE_MAINNET: &[&str] = &["enode://ca2774c3c401325850b2477fd7d0f27911efbf79b1e8b335066516e2bd8c4c9e0ba9696a94b1cb030a88eac582305ff55e905e64fb77fe0edcd70a4e5296d3ec@34.65.175.185:30305", "enode://dd751a9ef8912be1bfa7a5e34e2c3785cc5253110bd929f385e07ba7ac19929fb0e0c5d93f77827291f4da02b2232240fbc47ea7ce04c46e333e452f8656b667@34.65.107.0:30305", "enode://c5d289b56a77b6a2342ca29956dfd07aadf45364dde8ab20d1dc4efd4d1bc6b4655d902501daea308f4d8950737a4e93a4dfedd17b49cd5760ffd127837ca965@34.65.202.239:30305", "enode://87a32fd13bd596b2ffca97020e31aef4ddcc1bbd4b95bb633d16c1329f654f34049ed240a36b449fda5e5225d70fe40bc667f53c304b71f8e68fc9d448690b51@3.231.138.188:30301", "enode://ca21ea8f176adb2e229ce2d700830c844af0ea941a1d8152a9513b966fe525e809c3a6c73a2c18a12b74ed6ec4380edf91662778fe0b79f6a591236e49e176f9@184.72.129.189:30301", "enode://acf4507a211ba7c1e52cdf4eef62cdc3c32e7c9c47998954f7ba024026f9a6b2150cd3f0b734d9c78e507ab70d59ba61dfe5c45e1078c7ad0775fb251d7735a2@3.220.145.177:30301", "enode://8a5a5006159bf079d06a04e5eceab2a1ce6e0f721875b2a9c96905336219dbe14203d38f70f3754686a6324f786c2f9852d8c0dd3adac2d080f4db35efc678c5@3.231.11.52:30301", "enode://cdadbe835308ad3557f9a1de8db411da1a260a98f8421d62da90e71da66e55e98aaa8e90aa7ce01b408a54e4bd2253d701218081ded3dbe5efbbc7b41d7cef79@54.198.153.150:30301"];
+
+/// Optimism sepolia and base sepolia boot nodes.
+const BOOT_NODES_OP_SEPOLIA_AND_BASE_SEPOLIA: &[&str] = &["enode://09d1a6110757b95628cc54ab6cc50a29773075ed00e3a25bd9388807c9a6c007664e88646a6fefd82baad5d8374ba555e426e8aed93f0f0c517e2eb5d929b2a2@34.65.21.188:30304?discport=30303"];
+
+/// Builds a [`Config`].
+#[derive(Debug, Default)]
+pub struct ConfigBuilder {
+    /// Config used by [`discv5::Discv5`]. Contains the discovery listen socket.
+    discv5_config: Option<discv5::Config>,
+    /// Nodes to boot from.
+    bootstrap_nodes: HashSet<BootNode>,
+    /// [`ForkId`] to set in local node record.
+    fork: Option<(&'static [u8], ForkId)>,
+    /// RLPx TCP port to advertise. Note: so long as `reth_network` handles [`NodeRecord`]s as
+    /// opposed to [`Enr`](enr::Enr)s, TCP is limited to same IP address as UDP, since
+    /// [`NodeRecord`] doesn't supply an extra field for and alternative TCP address.
+    tcp_port: u16,
+    /// Additional kv-pairs that should be advertised to peers by including in local node record.
+    other_enr_data: Vec<(&'static str, Bytes)>,
+    /// Interval in seconds at which to run a lookup up query to populate kbuckets.
+    lookup_interval: Option<u64>,
+    /// Custom filter rules to apply to a discovered peer in order to determine if it should be
+    /// passed up to rlpx or dropped.
+    discovered_peer_filter: Option<MustNotIncludeKeys>,
+}
+
+impl ConfigBuilder {
+    /// Returns a new builder, with all fields set like given instance.
+    pub fn new_from(discv5_config: Config) -> Self {
+        let Config {
+            discv5_config,
+            bootstrap_nodes,
+            fork: fork_id,
+            tcp_port,
+            other_enr_data,
+            lookup_interval,
+            discovered_peer_filter,
+        } = discv5_config;
+
+        Self {
+            discv5_config: Some(discv5_config),
+            bootstrap_nodes,
+            fork: Some(fork_id),
+            tcp_port,
+            other_enr_data,
+            lookup_interval: Some(lookup_interval),
+            discovered_peer_filter: Some(discovered_peer_filter),
+        }
+    }
+
+    /// Set [`discv5::Config`], which contains the [`discv5::Discv5`] listen socket.
+    pub fn discv5_config(mut self, discv5_config: discv5::Config) -> Self {
+        self.discv5_config = Some(discv5_config);
+        self
+    }
+
+    /// Adds multiple boot nodes from a list of [`Enr`](discv5::Enr)s.
+    pub fn add_signed_boot_nodes(mut self, nodes: impl IntoIterator<Item = discv5::Enr>) -> Self {
+        self.bootstrap_nodes.extend(nodes.into_iter().map(BootNode::Enr));
+        self
+    }
+
+    /// Parses a comma-separated list of serialized [`Enr`](discv5::Enr)s, signed node records, and
+    /// adds any successfully deserialized records to boot nodes. Note: this type is serialized in
+    /// CL format since [`discv5`] is originally a CL library.
+    pub fn add_cl_serialized_signed_boot_nodes(mut self, enrs: &str) -> Self {
+        let bootstrap_nodes = &mut self.bootstrap_nodes;
+        for node in enrs.split(&[',']).flat_map(|record| record.trim().parse::<discv5::Enr>()) {
+            bootstrap_nodes.insert(BootNode::Enr(node));
+        }
+        self
+    }
+
+    /// Adds boot nodes in the form a list of [`NodeRecord`]s, parsed enodes.
+    pub fn add_unsigned_boot_nodes(mut self, enodes: Vec<NodeRecord>) -> Self {
+        for node in enodes {
+            if let Ok(node) = BootNode::from_unsigned(node) {
+                self.bootstrap_nodes.insert(node);
+            }
+        }
+
+        self
+    }
+
+    /// Adds a comma-separated list of enodes, serialized unsigned node records, to boot nodes.
+    pub fn add_serialized_unsigned_boot_nodes(mut self, enodes: &[&str]) -> Self {
+        for node in enodes {
+            if let Ok(node) = node.parse() {
+                if let Ok(node) = BootNode::from_unsigned(node) {
+                    self.bootstrap_nodes.insert(node);
+                }
+            }
+        }
+
+        self
+    }
+
+    /// Add optimism mainnet boot nodes.
+    pub fn add_optimism_mainnet_boot_nodes(self) -> Self {
+        self.add_serialized_unsigned_boot_nodes(BOOT_NODES_OP_MAINNET_AND_BASE_MAINNET)
+    }
+
+    /// Add optimism sepolia boot nodes.
+    pub fn add_optimism_sepolia_boot_nodes(self) -> Self {
+        self.add_serialized_unsigned_boot_nodes(BOOT_NODES_OP_SEPOLIA_AND_BASE_SEPOLIA)
+    }
+
+    /// Set [`ForkId`], and key used to identify it, to set in local [`Enr`](discv5::enr::Enr).
+    pub fn fork(mut self, key: &'static [u8], value: ForkId) -> Self {
+        self.fork = Some((key, value));
+        self
+    }
+
+    /// Sets the tcp port to advertise in the local [`Enr`](discv5::enr::Enr).
+    fn tcp_port(mut self, port: u16) -> Self {
+        self.tcp_port = port;
+        self
+    }
+
+    /// Adds an additional kv-pair to include in the local [`Enr`](discv5::enr::Enr).
+    pub fn add_enr_kv_pair(mut self, kv_pair: (&'static str, Bytes)) -> Self {
+        self.other_enr_data.push(kv_pair);
+        self
+    }
+
+    /// Adds keys to disallow when filtering a discovered peer, to determine whether or not it
+    /// should be passed to rlpx. The discovered node record is scanned for any kv-pairs where the
+    /// key matches the disallowed keys. If not explicitly set, b"eth2" key will be disallowed.
+    pub fn must_not_include_keys(mut self, not_keys: &[&'static [u8]]) -> Self {
+        let mut filter = self.discovered_peer_filter.unwrap_or_default();
+        filter.add_disallowed_keys(not_keys);
+        self.discovered_peer_filter = Some(filter);
+        self
+    }
+
+    /// Returns a new [`Config`].
+    pub fn build(self) -> Config {
+        let Self {
+            discv5_config,
+            bootstrap_nodes,
+            fork,
+            tcp_port,
+            other_enr_data,
+            lookup_interval,
+            discovered_peer_filter,
+        } = self;
+
+        let discv5_config = discv5_config
+            .unwrap_or_else(|| discv5::ConfigBuilder::new(ListenConfig::default()).build());
+
+        let fork = fork.unwrap_or((ETH, MAINNET.latest_fork_id()));
+
+        let lookup_interval = lookup_interval.unwrap_or(DEFAULT_SECONDS_LOOKUP_INTERVAL);
+
+        let discovered_peer_filter =
+            discovered_peer_filter.unwrap_or_else(|| MustNotIncludeKeys::new(&[ETH2]));
+
+        Config {
+            discv5_config,
+            bootstrap_nodes,
+            fork,
+            tcp_port,
+            other_enr_data,
+            lookup_interval,
+            discovered_peer_filter,
+        }
+    }
+}
+
+/// Config used to bootstrap [`discv5::Discv5`].
+#[derive(Debug)]
+pub struct Config {
+    /// Config used by [`discv5::Discv5`]. Contains the [`ListenConfig`], with the discovery listen
+    /// socket.
+    pub(super) discv5_config: discv5::Config,
+    /// Nodes to boot from.
+    pub(super) bootstrap_nodes: HashSet<BootNode>,
+    /// [`ForkId`] to set in local node record.
+    pub(super) fork: (&'static [u8], ForkId),
+    /// RLPx TCP port to advertise.
+    pub(super) tcp_port: u16,
+    /// Additional kv-pairs to include in local node record.
+    pub(super) other_enr_data: Vec<(&'static str, Bytes)>,
+    /// Interval in seconds at which to run a lookup up query with to populate kbuckets.
+    pub(super) lookup_interval: u64,
+    /// Custom filter rules to apply to a discovered peer in order to determine if it should be
+    /// passed up to rlpx or dropped.
+    pub(super) discovered_peer_filter: MustNotIncludeKeys,
+}
+
+impl Config {
+    /// Returns a new [`ConfigBuilder`], with the RLPx TCP port set to the given port.
+    pub fn builder(rlpx_tcp_port: u16) -> ConfigBuilder {
+        ConfigBuilder::default().tcp_port(rlpx_tcp_port)
+    }
+}
+
+impl Config {
+    /// Returns the discovery (UDP) socket contained in the [`discv5::Config`]. Returns the IPv6
+    /// socket, if both IPv4 and v6 are configured. This socket will be advertised to peers in the
+    /// local [`Enr`](discv5::enr::Enr).
+    pub fn discovery_socket(&self) -> SocketAddr {
+        match self.discv5_config.listen_config {
+            ListenConfig::Ipv4 { ip, port } => (ip, port).into(),
+            ListenConfig::Ipv6 { ip, port } => (ip, port).into(),
+            ListenConfig::DualStack { ipv6, ipv6_port, .. } => (ipv6, ipv6_port).into(),
+        }
+    }
+
+    /// Returns the RLPx (TCP) socket contained in the [`discv5::Config`]. This socket will be
+    /// advertised to peers in the local [`Enr`](discv5::enr::Enr).
+    pub fn rlpx_socket(&self) -> SocketAddr {
+        let port = self.tcp_port;
+        match self.discv5_config.listen_config {
+            ListenConfig::Ipv4 { ip, .. } => (ip, port).into(),
+            ListenConfig::Ipv6 { ip, .. } => (ip, port).into(),
+            ListenConfig::DualStack { ipv4, .. } => (ipv4, port).into(),
+        }
+    }
+}
+
+/// A boot node can be added either as a string in either 'enode' URL scheme or serialized from
+/// [`Enr`](discv5::Enr) type.
+#[derive(Debug, PartialEq, Eq, Hash, Display)]
+pub enum BootNode {
+    /// An unsigned node record.
+    #[display(fmt = "{_0}")]
+    Enode(Multiaddr),
+    /// A signed node record.
+    #[display(fmt = "{_0:?}")]
+    Enr(discv5::Enr),
+}
+
+impl BootNode {
+    /// Parses a [`NodeRecord`] and serializes according to CL format. Note: [`discv5`] is
+    /// originally a CL library hence needs this format to add the node.
+    pub fn from_unsigned(node_record: NodeRecord) -> Result<Self, secp256k1::Error> {
+        let NodeRecord { address, udp_port, id, .. } = node_record;
+        let mut multi_address = Multiaddr::empty();
+        match address {
+            IpAddr::V4(ip) => multi_address.push(Protocol::Ip4(ip)),
+            IpAddr::V6(ip) => multi_address.push(Protocol::Ip6(ip)),
+        }
+
+        multi_address.push(Protocol::Udp(udp_port));
+        let id = discv4_id_to_multiaddr_id(id)?;
+        multi_address.push(Protocol::P2p(id));
+
+        Ok(Self::Enode(multi_address))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::net::SocketAddrV4;
+
+    use reth_primitives::hex;
+
+    use super::*;
+
+    const MULTI_ADDRESSES: &str = "/ip4/184.72.129.189/udp/30301/p2p/16Uiu2HAmSG2hdLwyQHQmG4bcJBgD64xnW63WMTLcrNq6KoZREfGb,/ip4/3.231.11.52/udp/30301/p2p/16Uiu2HAmMy4V8bi3XP7KDfSLQcLACSvTLroRRwEsTyFUKo8NCkkp,/ip4/54.198.153.150/udp/30301/p2p/16Uiu2HAmSVsb7MbRf1jg3Dvd6a3n5YNqKQwn1fqHCFgnbqCsFZKe,/ip4/3.220.145.177/udp/30301/p2p/16Uiu2HAm74pBDGdQ84XCZK27GRQbGFFwQ7RsSqsPwcGmCR3Cwn3B,/ip4/3.231.138.188/udp/30301/p2p/16Uiu2HAmMnTiJwgFtSVGV14ZNpwAvS1LUoF4pWWeNtURuV6C3zYB";
+
+    #[test]
+    fn parse_boot_nodes() {
+        const OP_SEPOLIA_CL_BOOTNODES: &str ="enr:-J64QBwRIWAco7lv6jImSOjPU_W266lHXzpAS5YOh7WmgTyBZkgLgOwo_mxKJq3wz2XRbsoBItbv1dCyjIoNq67mFguGAYrTxM42gmlkgnY0gmlwhBLSsHKHb3BzdGFja4S0lAUAiXNlY3AyNTZrMaEDmoWSi8hcsRpQf2eJsNUx-sqv6fH4btmo2HsAzZFAKnKDdGNwgiQGg3VkcIIkBg,enr:-J64QFa3qMsONLGphfjEkeYyF6Jkil_jCuJmm7_a42ckZeUQGLVzrzstZNb1dgBp1GGx9bzImq5VxJLP-BaptZThGiWGAYrTytOvgmlkgnY0gmlwhGsV-zeHb3BzdGFja4S0lAUAiXNlY3AyNTZrMaEDahfSECTIS_cXyZ8IyNf4leANlZnrsMEWTkEYxf4GMCmDdGNwgiQGg3VkcIIkBg";
+
+        let config = Config::builder(30303)
+            .add_cl_serialized_signed_boot_nodes(OP_SEPOLIA_CL_BOOTNODES)
+            .build();
+
+        let socket_1 = "18.210.176.114:9222".parse::<SocketAddrV4>().unwrap();
+        let socket_2 = "107.21.251.55:9222".parse::<SocketAddrV4>().unwrap();
+
+        for node in config.bootstrap_nodes {
+            let BootNode::Enr(node) = node else { panic!() };
+            assert!(
+                socket_1 == node.udp4_socket().unwrap() && socket_1 == node.tcp4_socket().unwrap() ||
+                    socket_2 == node.udp4_socket().unwrap() &&
+                        socket_2 == node.tcp4_socket().unwrap()
+            );
+            assert_eq!("84b4940500", hex::encode(node.get_raw_rlp("opstack").unwrap()));
+        }
+    }
+
+    #[test]
+    fn parse_enodes() {
+        let config = Config::builder(30303)
+            .add_serialized_unsigned_boot_nodes(BOOT_NODES_OP_MAINNET_AND_BASE_MAINNET)
+            .build();
+
+        let bootstrap_nodes =
+            config.bootstrap_nodes.into_iter().map(|node| format!("{node}")).collect::<Vec<_>>();
+
+        for node in MULTI_ADDRESSES.split(&[',']) {
+            assert!(bootstrap_nodes.contains(&node.to_string()));
+        }
+    }
+}

crates/net/discv5/src/enr.rs

@@ -0,0 +1,113 @@
+//! Interface between node identification on protocol version 5 and 4. Specifically, between types
+//! [`discv5::enr::NodeId`] and [`PeerId`].
+
+use discv5::enr::{CombinedPublicKey, Enr, EnrPublicKey, NodeId};
+use reth_primitives::{id2pk, pk2id, PeerId};
+use secp256k1::{PublicKey, SecretKey};
+
+/// Extracts a [`CombinedPublicKey::Secp256k1`] from a [`discv5::Enr`] and converts it to a
+/// [`PeerId`]. Note: conversion from discv5 ID to discv4 ID is not possible.
+pub fn enr_to_discv4_id(enr: &discv5::Enr) -> Option<PeerId> {
+    let pk = enr.public_key();
+    if !matches!(pk, CombinedPublicKey::Secp256k1(_)) {
+        return None
+    }
+
+    let pk = PublicKey::from_slice(&pk.encode()).unwrap();
+
+    Some(pk2id(&pk))
+}
+
+/// Converts a [`PeerId`] to a [`discv5::enr::NodeId`].
+pub fn discv4_id_to_discv5_id(peer_id: PeerId) -> Result<NodeId, secp256k1::Error> {
+    Ok(id2pk(peer_id)?.into())
+}
+
+/// Converts a [`PeerId`] to a [`libp2p_identity::PeerId `].
+pub fn discv4_id_to_multiaddr_id(
+    peer_id: PeerId,
+) -> Result<libp2p_identity::PeerId, secp256k1::Error> {
+    let pk = id2pk(peer_id)?.encode();
+    let pk: libp2p_identity::PublicKey =
+        libp2p_identity::secp256k1::PublicKey::try_from_bytes(&pk).unwrap().into();
+
+    Ok(pk.to_peer_id())
+}
+
+/// Wrapper around [`discv5::Enr`] ([`Enr<CombinedKey>`]).
+#[derive(Debug, Clone)]
+pub struct EnrCombinedKeyWrapper(pub discv5::Enr);
+
+impl From<Enr<SecretKey>> for EnrCombinedKeyWrapper {
+    fn from(value: Enr<SecretKey>) -> Self {
+        let encoded_enr = rlp::encode(&value);
+        let enr = rlp::decode::<discv5::Enr>(&encoded_enr).unwrap();
+
+        Self(enr)
+    }
+}
+
+impl From<EnrCombinedKeyWrapper> for Enr<SecretKey> {
+    fn from(val: EnrCombinedKeyWrapper) -> Self {
+        let EnrCombinedKeyWrapper(enr) = val;
+        let encoded_enr = rlp::encode(&enr);
+
+        rlp::decode::<Enr<SecretKey>>(&encoded_enr).unwrap()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use alloy_rlp::Encodable;
+    use discv5::enr::{CombinedKey, EnrKey};
+    use reth_primitives::{pk_to_id, Hardfork, NodeRecord, MAINNET};
+
+    use super::*;
+
+    #[test]
+    fn discv5_discv4_id_conversion() {
+        let discv5_pk = CombinedKey::generate_secp256k1().public();
+        let discv5_peer_id = NodeId::from(discv5_pk.clone());
+
+        // convert to discv4 id
+        let pk = secp256k1::PublicKey::from_slice(&discv5_pk.encode()).unwrap();
+        let discv4_peer_id = pk2id(&pk);
+        // convert back to discv5 id
+        let discv5_peer_id_from_discv4_peer_id = discv4_id_to_discv5_id(discv4_peer_id).unwrap();
+
+        assert_eq!(discv5_peer_id, discv5_peer_id_from_discv4_peer_id)
+    }
+
+    #[test]
+    fn conversion_to_node_record_from_enr() {
+        const IP: &str = "::";
+        const TCP_PORT: u16 = 30303;
+        const UDP_PORT: u16 = 9000;
+
+        let key = CombinedKey::generate_secp256k1();
+
+        let mut buf = Vec::new();
+        let fork_id = MAINNET.hardfork_fork_id(Hardfork::Frontier);
+        fork_id.unwrap().encode(&mut buf);
+
+        let enr = Enr::builder()
+            .ip6(IP.parse().unwrap())
+            .udp6(UDP_PORT)
+            .tcp6(TCP_PORT)
+            .build(&key)
+            .unwrap();
+
+        let enr = EnrCombinedKeyWrapper(enr).into();
+        let node_record = NodeRecord::try_from(&enr).unwrap();
+
+        assert_eq!(
+            NodeRecord {
+                address: IP.parse().unwrap(),
+                tcp_port: TCP_PORT,
+                udp_port: UDP_PORT,
+                id: pk_to_id(&enr.public_key())
+            },
+            node_record
+        )
+    }
+}

crates/net/discv5/src/error.rs

@@ -0,0 +1,38 @@
+//! Errors interfacing with [`discv5::Discv5`].
+
+use discv5::IpMode;
+
+/// Errors interfacing with [`discv5::Discv5`].
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    /// Failure adding node to [`discv5::Discv5`].
+    #[error("failed adding node to discv5, {0}")]
+    AddNodeToDiscv5Failed(&'static str),
+    /// Node record has incompatible key type.
+    #[error("incompatible key type (not secp256k1)")]
+    IncompatibleKeyType,
+    /// Missing key used to identify rlpx network.
+    #[error("fork missing on enr, 'eth' key missing")]
+    ForkMissing,
+    /// Failed to decode [`ForkId`](reth_primitives::ForkId) rlp value.
+    #[error("failed to decode fork id, 'eth': {0:?}")]
+    ForkIdDecodeError(#[from] alloy_rlp::Error),
+    /// Peer is unreachable over discovery.
+    #[error("discovery socket missing")]
+    UnreachableDiscovery,
+    /// Peer is unreachable over rlpx.
+    #[error("RLPx TCP socket missing")]
+    UnreachableRlpx,
+    /// Peer is not using same IP version as local node in rlpx.
+    #[error("RLPx TCP socket is unsupported IP version, local ip mode: {0:?}")]
+    IpVersionMismatchRlpx(IpMode),
+    /// Failed to initialize [`discv5::Discv5`].
+    #[error("init failed, {0}")]
+    InitFailure(&'static str),
+    /// An error from underlying [`discv5::Discv5`] node.
+    #[error("{0}")]
+    Discv5Error(discv5::Error),
+    /// An error from underlying [`discv5::Discv5`] node.
+    #[error("{0}")]
+    Discv5ErrorStr(&'static str),
+}

crates/net/discv5/src/filter.rs

@@ -0,0 +1,123 @@
+//! Predicates to constraint peer lookups.
+
+use std::collections::HashSet;
+
+use derive_more::Constructor;
+use itertools::Itertools;
+
+/// Outcome of applying filtering rules on node record.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum FilterOutcome {
+    /// ENR passes filter rules.
+    Ok,
+    /// ENR doesn't pass filter rules, for the given reason.
+    Ignore {
+        /// Reason for filtering out node record.
+        reason: String,
+    },
+}
+
+impl FilterOutcome {
+    /// Returns `true` for [`FilterOutcome::Ok`].
+    pub fn is_ok(&self) -> bool {
+        matches!(self, FilterOutcome::Ok)
+    }
+}
+
+/// Filter requiring that peers advertise that they belong to some fork of a certain key.
+#[derive(Debug, Constructor, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MustIncludeKey {
+    /// Kv-pair key which node record must advertise.
+    key: &'static [u8],
+}
+
+impl MustIncludeKey {
+    /// Returns [`FilterOutcome::Ok`] if [`Enr`](discv5::Enr) contains the configured kv-pair key.
+    pub fn filter(&self, enr: &discv5::Enr) -> FilterOutcome {
+        if enr.get_raw_rlp(self.key).is_none() {
+            return FilterOutcome::Ignore { reason: self.ignore_reason() }
+        }
+        FilterOutcome::Ok
+    }
+
+    fn ignore_reason(&self) -> String {
+        format!("{} fork required", String::from_utf8_lossy(self.key))
+    }
+}
+
+/// Filter requiring that peers not advertise kv-pairs using certain keys, e.g. b"eth2".
+#[derive(Debug, Clone, Default)]
+pub struct MustNotIncludeKeys {
+    keys: HashSet<MustIncludeKey>,
+}
+
+impl MustNotIncludeKeys {
+    /// Returns a new instance that disallows node records with a kv-pair that has any of the given
+    /// keys.
+    pub fn new(disallow_keys: &[&'static [u8]]) -> Self {
+        let mut keys = HashSet::with_capacity(disallow_keys.len());
+        for key in disallow_keys {
+            _ = keys.insert(MustIncludeKey::new(key));
+        }
+
+        MustNotIncludeKeys { keys }
+    }
+}
+
+impl MustNotIncludeKeys {
+    /// Returns `true` if [`Enr`](discv5::Enr) passes filtering rules.
+    pub fn filter(&self, enr: &discv5::Enr) -> FilterOutcome {
+        for key in self.keys.iter() {
+            if matches!(key.filter(enr), FilterOutcome::Ok) {
+                return FilterOutcome::Ignore { reason: self.ignore_reason() }
+            }
+        }
+
+        FilterOutcome::Ok
+    }
+
+    fn ignore_reason(&self) -> String {
+        format!(
+            "{} forks not allowed",
+            self.keys.iter().map(|key| String::from_utf8_lossy(key.key)).format(",")
+        )
+    }
+
+    /// Adds a key that must not be present for any kv-pair in a node record.
+    pub fn add_disallowed_keys(&mut self, keys: &[&'static [u8]]) {
+        for key in keys {
+            self.keys.insert(MustIncludeKey::new(key));
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use alloy_rlp::Bytes;
+    use discv5::enr::{CombinedKey, Enr};
+
+    use crate::config::{ETH, ETH2};
+
+    use super::*;
+
+    #[test]
+    fn must_not_include_key_filter() {
+        // rig test
+
+        let filter = MustNotIncludeKeys::new(&[ETH, ETH2]);
+
+        // enr_1 advertises a fork from one of the keys configured in filter
+        let sk = CombinedKey::generate_secp256k1();
+        let enr_1 =
+            Enr::builder().add_value_rlp(ETH as &[u8], Bytes::from("cancun")).build(&sk).unwrap();
+
+        // enr_2 advertises a fork from one the other key configured in filter
+        let sk = CombinedKey::generate_secp256k1();
+        let enr_2 = Enr::builder().add_value_rlp(ETH2, Bytes::from("deneb")).build(&sk).unwrap();
+
+        // test
+
+        assert!(matches!(filter.filter(&enr_1), FilterOutcome::Ignore { .. }));
+        assert!(matches!(filter.filter(&enr_2), FilterOutcome::Ignore { .. }));
+    }
+}

crates/net/discv5/src/lib.rs

@@ -0,0 +1,788 @@
+//! Wrapper around [`discv5::Discv5`].
+
+#![doc(
+    html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
+    html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
+    issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
+)]
+#![cfg_attr(not(test), warn(unused_crate_dependencies))]
+#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
+
+use std::{
+    collections::HashSet,
+    fmt,
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+    sync::Arc,
+    time::Duration,
+};
+
+use ::enr::Enr;
+use alloy_rlp::Decodable;
+use derive_more::Deref;
+use discv5::ListenConfig;
+use enr::{discv4_id_to_discv5_id, EnrCombinedKeyWrapper};
+use futures::future::join_all;
+use itertools::Itertools;
+use reth_primitives::{bytes::Bytes, ForkId, NodeRecord, PeerId};
+use secp256k1::SecretKey;
+use tokio::{sync::mpsc, task};
+use tracing::{debug, error, trace};
+
+pub mod config;
+pub mod enr;
+pub mod error;
+pub mod filter;
+pub mod metrics;
+
+pub use discv5::{self, IpMode};
+
+pub use config::{BootNode, Config, ConfigBuilder};
+pub use enr::enr_to_discv4_id;
+pub use error::Error;
+pub use filter::{FilterOutcome, MustNotIncludeKeys};
+use metrics::Discv5Metrics;
+
+/// The max log2 distance, is equivalent to the index of the last bit in a discv5 node id.
+const MAX_LOG2_DISTANCE: usize = 255;
+
+/// Transparent wrapper around [`discv5::Discv5`].
+#[derive(Deref, Clone)]
+pub struct Discv5 {
+    #[deref]
+    /// sigp/discv5 node.
+    discv5: Arc<discv5::Discv5>,
+    /// [`IpMode`] of the the node.
+    ip_mode: IpMode,
+    /// Key used in kv-pair to ID chain.
+    fork_id_key: &'static [u8],
+    /// Filter applied to a discovered peers before passing it up to app.
+    discovered_peer_filter: MustNotIncludeKeys,
+    /// Metrics for underlying [`discv5::Discv5`] node and filtered discovered peers.
+    metrics: Discv5Metrics,
+}
+
+impl Discv5 {
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    // Minimal interface with `reth_network::discovery`
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+
+    /// Adds the node to the table, if it is not already present.
+    pub fn add_node_to_routing_table(&self, node_record: Enr<SecretKey>) -> Result<(), Error> {
+        let EnrCombinedKeyWrapper(enr) = node_record.into();
+        self.add_enr(enr).map_err(Error::AddNodeToDiscv5Failed)
+    }
+
+    /// Sets the pair in the EIP-868 [`Enr`] of the node.
+    ///
+    /// If the key already exists, this will update it.
+    ///
+    /// CAUTION: The value **must** be rlp encoded
+    pub fn set_eip868_in_local_enr(&self, key: Vec<u8>, rlp: Bytes) {
+        let Ok(key_str) = std::str::from_utf8(&key) else {
+            error!(target: "discv5",
+                err="key not utf-8",
+                "failed to update local enr"
+            );
+            return
+        };
+        if let Err(err) = self.enr_insert(key_str, &rlp) {
+            error!(target: "discv5",
+                %err,
+                "failed to update local enr"
+            );
+        }
+    }
+
+    /// Sets the pair in the EIP-868 [`Enr`] of the node.
+    ///
+    /// If the key already exists, this will update it.
+    pub fn encode_and_set_eip868_in_local_enr(
+        &self,
+        key: Vec<u8>,
+        value: impl alloy_rlp::Encodable,
+    ) {
+        let mut buf = Vec::new();
+        value.encode(&mut buf);
+        self.set_eip868_in_local_enr(key, buf.into())
+    }
+
+    /// Adds the peer and id to the ban list.
+    ///
+    /// This will prevent any future inclusion in the table
+    pub fn ban_peer_by_ip_and_node_id(&self, peer_id: PeerId, ip: IpAddr) {
+        match discv4_id_to_discv5_id(peer_id) {
+            Ok(node_id) => {
+                self.ban_node(&node_id, None);
+                self.ban_peer_by_ip(ip);
+            }
+            Err(err) => error!(target: "discv5",
+                %err,
+                "failed to ban peer"
+            ),
+        }
+    }
+
+    /// Adds the ip to the ban list.
+    ///
+    /// This will prevent any future inclusion in the table
+    pub fn ban_peer_by_ip(&self, ip: IpAddr) {
+        self.ban_ip(ip, None);
+    }
+
+    /// Returns the [`NodeRecord`] of the local node.
+    ///
+    /// This includes the currently tracked external IP address of the node.
+    pub fn node_record(&self) -> NodeRecord {
+        let enr: Enr<_> = EnrCombinedKeyWrapper(self.local_enr()).into();
+        (&enr).try_into().unwrap()
+    }
+
+    /// Spawns [`discv5::Discv5`]. Returns [`discv5::Discv5`] handle in reth compatible wrapper type
+    /// [`Discv5`], a receiver of [`discv5::Event`]s from the underlying node, and the local
+    /// [`Enr`](discv5::Enr) converted into the reth compatible [`NodeRecord`] type.
+    pub async fn start(
+        sk: &SecretKey,
+        discv5_config: Config,
+    ) -> Result<(Self, mpsc::Receiver<discv5::Event>, NodeRecord), Error> {
+        //
+        // 1. make local enr from listen config
+        //
+        let Config {
+            discv5_config,
+            bootstrap_nodes,
+            fork,
+            tcp_port,
+            other_enr_data,
+            lookup_interval,
+            discovered_peer_filter,
+        } = discv5_config;
+
+        let (enr, bc_enr, ip_mode, fork_id_key) = {
+            let mut builder = discv5::enr::Enr::builder();
+
+            let (ip_mode, socket) = match discv5_config.listen_config {
+                ListenConfig::Ipv4 { ip, port } => {
+                    if ip != Ipv4Addr::UNSPECIFIED {
+                        builder.ip4(ip);
+                    }
+                    builder.udp4(port);
+                    builder.tcp4(tcp_port);
+
+                    (IpMode::Ip4, (ip, port).into())
+                }
+                ListenConfig::Ipv6 { ip, port } => {
+                    if ip != Ipv6Addr::UNSPECIFIED {
+                        builder.ip6(ip);
+                    }
+                    builder.udp6(port);
+                    builder.tcp6(tcp_port);
+
+                    (IpMode::Ip6, (ip, port).into())
+                }
+                ListenConfig::DualStack { ipv4, ipv4_port, ipv6, ipv6_port } => {
+                    if ipv4 != Ipv4Addr::UNSPECIFIED {
+                        builder.ip4(ipv4);
+                    }
+                    builder.udp4(ipv4_port);
+                    builder.tcp4(tcp_port);
+
+                    if ipv6 != Ipv6Addr::UNSPECIFIED {
+                        builder.ip6(ipv6);
+                    }
+                    builder.udp6(ipv6_port);
+
+                    (IpMode::DualStack, (ipv6, ipv6_port).into())
+                }
+            };
+
+            // add fork id
+            let (chain, fork_id) = fork;
+            builder.add_value_rlp(chain, alloy_rlp::encode(fork_id).into());
+
+            // add other data
+            for (key, value) in other_enr_data {
+                builder.add_value_rlp(key, alloy_rlp::encode(value).into());
+            }
+
+            // enr v4 not to get confused with discv4, independent versioning enr and
+            // discovery
+            let enr = builder.build(sk).expect("should build enr v4");
+            let EnrCombinedKeyWrapper(enr) = enr.into();
+
+            trace!(target: "net::discv5",
+                ?enr,
+                "local ENR"
+            );
+
+            // backwards compatible enr
+            let bc_enr = NodeRecord::from_secret_key(socket, sk);
+
+            (enr, bc_enr, ip_mode, chain)
+        };
+
+        //
+        // 3. start discv5
+        //
+        let sk = discv5::enr::CombinedKey::secp256k1_from_bytes(&mut sk.secret_bytes()).unwrap();
+        let mut discv5 = match discv5::Discv5::new(enr, sk, discv5_config) {
+            Ok(discv5) => discv5,
+            Err(err) => return Err(Error::InitFailure(err)),
+        };
+        discv5.start().await.map_err(Error::Discv5Error)?;
+
+        // start discv5 updates stream
+        let discv5_updates = discv5.event_stream().await.map_err(Error::Discv5Error)?;
+
+        let discv5 = Arc::new(discv5);
+
+        //
+        // 4. add boot nodes
+        //
+        Self::bootstrap(bootstrap_nodes, &discv5)?;
+
+        let metrics = Discv5Metrics::default();
+
+        //
+        // 5. bg kbuckets maintenance
+        //
+        Self::spawn_populate_kbuckets_bg(lookup_interval, metrics.clone(), discv5.clone());
+
+        Ok((
+            Self { discv5, ip_mode, fork_id_key, discovered_peer_filter, metrics },
+            discv5_updates,
+            bc_enr,
+        ))
+    }
+
+    /// Bootstraps underlying [`discv5::Discv5`] node with configured peers.
+    fn bootstrap(
+        bootstrap_nodes: HashSet<BootNode>,
+        discv5: &Arc<discv5::Discv5>,
+    ) -> Result<(), Error> {
+        trace!(target: "net::discv5",
+            ?bootstrap_nodes,
+            "adding bootstrap nodes .."
+        );
+
+        let mut enr_requests = vec![];
+        for node in bootstrap_nodes {
+            match node {
+                BootNode::Enr(node) => {
+                    if let Err(err) = discv5.add_enr(node) {
+                        return Err(Error::Discv5ErrorStr(err))
+                    }
+                }
+                BootNode::Enode(enode) => {
+                    let discv5 = discv5.clone();
+                    enr_requests.push(async move {
+                        if let Err(err) = discv5.request_enr(enode.to_string()).await {
+                            debug!(target: "net::discv5",
+                                ?enode,
+                                %err,
+                                "failed adding boot node"
+                            );
+                        }
+                    })
+                }
+            }
+        }
+        _ = join_all(enr_requests);
+
+        debug!(target: "net::discv5",
+            nodes=format!("[{:#}]", discv5.with_kbuckets(|kbuckets| kbuckets
+                .write()
+                .iter()
+                .map(|peer| format!("enr: {:?}, status: {:?}", peer.node.value, peer.status)).collect::<Vec<_>>()
+            ).into_iter().format(", ")),
+            "added boot nodes"
+        );
+
+        Ok(())
+    }
+
+    /// Backgrounds regular look up queries, in order to keep kbuckets populated.
+    fn spawn_populate_kbuckets_bg(
+        lookup_interval: u64,
+        metrics: Discv5Metrics,
+        discv5: Arc<discv5::Discv5>,
+    ) {
+        // initiate regular lookups to populate kbuckets
+        task::spawn({
+            let local_node_id = discv5.local_enr().node_id();
+            let lookup_interval = Duration::from_secs(lookup_interval);
+            let mut metrics = metrics.discovered_peers;
+            let mut log2_distance = 0usize;
+            // todo: graceful shutdown
+
+            async move {
+                loop {
+                    metrics.set_total_sessions(discv5.metrics().active_sessions);
+                    metrics.set_total_kbucket_peers(
+                        discv5.with_kbuckets(|kbuckets| kbuckets.read().iter_ref().count()),
+                    );
+
+                    trace!(target: "net::discv5",
+                        lookup_interval=format!("{:#?}", lookup_interval),
+                        "starting periodic lookup query"
+                    );
+                    // make sure node is connected to each subtree in the network by target
+                    // selection (ref kademlia)
+                    let target = get_lookup_target(log2_distance, local_node_id);
+                    if log2_distance < MAX_LOG2_DISTANCE {
+                        // try to populate bucket one step further away
+                        log2_distance += 1
+                    } else {
+                        // start over with self lookup
+                        log2_distance = 0
+                    }
+                    match discv5.find_node(target).await {
+                        Err(err) => trace!(target: "net::discv5",
+                            lookup_interval=format!("{:#?}", lookup_interval),
+                            %err,
+                            "periodic lookup query failed"
+                        ),
+                        Ok(peers) => trace!(target: "net::discv5",
+                            lookup_interval=format!("{:#?}", lookup_interval),
+                            peers_count=peers.len(),
+                            peers=format!("[{:#}]", peers.iter()
+                                .map(|enr| enr.node_id()
+                            ).format(", ")),
+                            "peers returned by periodic lookup query"
+                        ),
+                    }
+
+                    // `Discv5::connected_peers` can be subset of sessions, not all peers make it
+                    // into kbuckets, e.g. incoming sessions from peers with
+                    // unreachable enrs
+                    debug!(target: "net::discv5",
+                        connected_peers=discv5.connected_peers(),
+                        "connected peers in routing table"
+                    );
+                    tokio::time::sleep(lookup_interval).await;
+                }
+            }
+        });
+    }
+
+    /// Process an event from the underlying [`discv5::Discv5`] node.
+    pub fn on_discv5_update(&mut self, update: discv5::Event) -> Option<DiscoveredPeer> {
+        match update {
+            discv5::Event::SocketUpdated(_) | discv5::Event::TalkRequest(_) |
+            // `EnrAdded` not used in discv5 codebase
+            discv5::Event::EnrAdded { .. } |
+            // `Discovered` not unique discovered peers
+            discv5::Event::Discovered(_) => None,
+            discv5::Event::NodeInserted { replaced: _, .. } => {
+
+                // node has been inserted into kbuckets
+
+                // `replaced` covers `reth_discv4::DiscoveryUpdate::Removed(_)` .. but we can't get 
+                // a `PeerId` from a `NodeId`
+
+                self.metrics.discovered_peers.increment_kbucket_insertions(1);
+
+                None
+            }
+            discv5::Event::SessionEstablished(enr, remote_socket) => {
+                // covers `reth_discv4::DiscoveryUpdate` equivalents `DiscoveryUpdate::Added(_)` 
+                // and `DiscoveryUpdate::DiscoveredAtCapacity(_)
+
+                // peer has been discovered as part of query, or, by incoming session (peer has 
+                // discovered us)
+
+                self.metrics.discovered_peers_advertised_networks.increment_once_by_network_type(&enr);
+
+                self.metrics.discovered_peers.increment_established_sessions_raw(1);
+
+                self.on_discovered_peer(&enr, remote_socket)
+            }
+        }
+    }
+
+    /// Processes a discovered peer. Returns `true` if peer is added to
+    fn on_discovered_peer(
+        &mut self,
+        enr: &discv5::Enr,
+        socket: SocketAddr,
+    ) -> Option<DiscoveredPeer> {
+        let node_record = match self.try_into_reachable(enr, socket) {
+            Ok(enr_bc) => enr_bc,
+            Err(err) => {
+                trace!(target: "net::discovery::discv5",
+                    %err,
+                    "discovered peer is unreachable"
+                );
+
+                self.metrics.discovered_peers.increment_established_sessions_unreachable_enr(1);
+
+                return None
+            }
+        };
+        let fork_id = match self.filter_discovered_peer(enr) {
+            FilterOutcome::Ok => self.get_fork_id(enr).ok(),
+            FilterOutcome::Ignore { reason } => {
+                trace!(target: "net::discovery::discv5",
+                    ?enr,
+                    reason,
+                    "filtered out discovered peer"
+                );
+
+                self.metrics.discovered_peers.increment_established_sessions_filtered(1);
+
+                return None
+            }
+        };
+
+        trace!(target: "net::discovery::discv5",
+            ?fork_id,
+            ?enr,
+            "discovered peer"
+        );
+
+        Some(DiscoveredPeer { node_record, fork_id })
+    }
+
+    /// Tries to convert an [`Enr`](discv5::Enr) into the backwards compatible type [`NodeRecord`],
+    /// w.r.t. local [`IpMode`]. Tries the socket from which the ENR was sent, if socket is missing
+    /// from ENR.
+    ///
+    ///  Note: [`discv5::Discv5`] won't initiate a session with any peer with a malformed node
+    /// record, that advertises a reserved IP address on a WAN network.
+    fn try_into_reachable(
+        &self,
+        enr: &discv5::Enr,
+        socket: SocketAddr,
+    ) -> Result<NodeRecord, Error> {
+        let id = enr_to_discv4_id(enr).ok_or(Error::IncompatibleKeyType)?;
+
+        let udp_socket = self.ip_mode().get_contactable_addr(enr).unwrap_or(socket);
+
+        // since we, on bootstrap, set tcp4 in local ENR for `IpMode::Dual`, we prefer tcp4 here
+        // too
+        let Some(tcp_port) = (match self.ip_mode() {
+            IpMode::Ip4 | IpMode::DualStack => enr.tcp4(),
+            IpMode::Ip6 => enr.tcp6(),
+        }) else {
+            return Err(Error::IpVersionMismatchRlpx(self.ip_mode()))
+        };
+
+        Ok(NodeRecord { address: udp_socket.ip(), tcp_port, udp_port: udp_socket.port(), id })
+    }
+
+    /// Applies filtering rules on an ENR. Returns [`Ok`](FilterOutcome::Ok) if peer should be
+    /// passed up to app, and [`Ignore`](FilterOutcome::Ignore) if peer should instead be dropped.
+    fn filter_discovered_peer(&self, enr: &discv5::Enr) -> FilterOutcome {
+        self.discovered_peer_filter.filter(enr)
+    }
+
+    /// Returns the [`ForkId`] of the given [`Enr`](discv5::Enr), if field is set.
+    fn get_fork_id<K: discv5::enr::EnrKey>(
+        &self,
+        enr: &discv5::enr::Enr<K>,
+    ) -> Result<ForkId, Error> {
+        let mut fork_id_bytes = enr.get_raw_rlp(self.fork_id_key()).ok_or(Error::ForkMissing)?;
+
+        Ok(ForkId::decode(&mut fork_id_bytes)?)
+    }
+
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    // Interface with sigp/discv5
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+
+    /// Exposes API of [`discv5::Discv5`].
+    pub fn with_discv5<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&Self) -> R,
+    {
+        f(self)
+    }
+
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    // Complementary
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+
+    /// Returns the [`IpMode`] of the local node.
+    pub fn ip_mode(&self) -> IpMode {
+        self.ip_mode
+    }
+
+    /// Returns the key to use to identify the [`ForkId`] kv-pair on the [`Enr`](discv5::Enr).
+    pub fn fork_id_key(&self) -> &[u8] {
+        self.fork_id_key
+    }
+}
+
+impl fmt::Debug for Discv5 {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        "{ .. }".fmt(f)
+    }
+}
+
+/// Result of successfully processing a peer discovered by [`discv5::Discv5`].
+#[derive(Debug)]
+pub struct DiscoveredPeer {
+    /// A discovery v4 backwards compatible ENR.
+    pub node_record: NodeRecord,
+    /// [`ForkId`] extracted from ENR w.r.t. configured
+    pub fork_id: Option<ForkId>,
+}
+
+/// Gets the next lookup target, based on which distance is currently being targeted.
+pub fn get_lookup_target(
+    log2_distance: usize,
+    local_node_id: discv5::enr::NodeId,
+) -> discv5::enr::NodeId {
+    let mut target = local_node_id.raw();
+    //make sure target has a 'distance'-long suffix that differs from local node id
+    if log2_distance != 0 {
+        let suffix_bit_offset = MAX_LOG2_DISTANCE.saturating_sub(log2_distance);
+        let suffix_byte_offset = suffix_bit_offset / 8;
+        // todo: flip the precise bit
+        // let rel_suffix_bit_offset = suffix_bit_offset % 8;
+        target[suffix_byte_offset] = !target[suffix_byte_offset];
+
+        if suffix_byte_offset != 31 {
+            for b in target.iter_mut().take(31).skip(suffix_byte_offset + 1) {
+                *b = rand::random::<u8>();
+            }
+        }
+    }
+
+    target.into()
+}
+
+#[cfg(test)]
+mod tests {
+    use ::enr::{CombinedKey, EnrKey};
+    use rand::Rng;
+    use secp256k1::rand::thread_rng;
+    use tracing::trace;
+
+    use super::*;
+
+    fn discv5_noop() -> Discv5 {
+        let sk = CombinedKey::generate_secp256k1();
+        Discv5 {
+            discv5: Arc::new(
+                discv5::Discv5::new(
+                    Enr::empty(&sk).unwrap(),
+                    sk,
+                    discv5::ConfigBuilder::new(ListenConfig::default()).build(),
+                )
+                .unwrap(),
+            ),
+            ip_mode: IpMode::Ip4,
+            fork_id_key: b"noop",
+            discovered_peer_filter: MustNotIncludeKeys::default(),
+            metrics: Discv5Metrics::default(),
+        }
+    }
+
+    async fn start_discovery_node(
+        udp_port_discv5: u16,
+    ) -> (Discv5, mpsc::Receiver<discv5::Event>, NodeRecord) {
+        let secret_key = SecretKey::new(&mut thread_rng());
+
+        let discv5_addr: SocketAddr = format!("127.0.0.1:{udp_port_discv5}").parse().unwrap();
+
+        let discv5_listen_config = ListenConfig::from(discv5_addr);
+        let discv5_config = Config::builder(30303)
+            .discv5_config(discv5::ConfigBuilder::new(discv5_listen_config).build())
+            .build();
+
+        Discv5::start(&secret_key, discv5_config).await.expect("should build discv5")
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn discv5() {
+        reth_tracing::init_test_tracing();
+
+        // rig test
+
+        // rig node_1
+        let (node_1, mut stream_1, _) = start_discovery_node(30344).await;
+        let node_1_enr = node_1.with_discv5(|discv5| discv5.local_enr());
+
+        // rig node_2
+        let (node_2, mut stream_2, _) = start_discovery_node(30355).await;
+        let node_2_enr = node_2.with_discv5(|discv5| discv5.local_enr());
+
+        trace!(target: "net::discovery::tests",
+            node_1_node_id=format!("{:#}", node_1_enr.node_id()),
+            node_2_node_id=format!("{:#}", node_2_enr.node_id()),
+            "started nodes"
+        );
+
+        // test
+
+        // add node_2 to discovery handle of node_1 (should add node to discv5 kbuckets)
+        let node_2_enr_reth_compatible_ty: Enr<SecretKey> =
+            EnrCombinedKeyWrapper(node_2_enr.clone()).into();
+        node_1.add_node_to_routing_table(node_2_enr_reth_compatible_ty).unwrap();
+
+        // verify node_2 is in KBuckets of node_1:discv5
+        assert!(
+            node_1.with_discv5(|discv5| discv5.table_entries_id().contains(&node_2_enr.node_id()))
+        );
+
+        // manually trigger connection from node_1 to node_2
+        node_1.with_discv5(|discv5| discv5.send_ping(node_2_enr.clone())).await.unwrap();
+
+        // verify node_1:discv5 is connected to node_2:discv5 and vv
+        let event_2_v5 = stream_2.recv().await.unwrap();
+        let event_1_v5 = stream_1.recv().await.unwrap();
+        matches!(
+            event_1_v5,
+            discv5::Event::SessionEstablished(node, socket) if node == node_2_enr && socket == node_2_enr.udp4_socket().unwrap().into()
+        );
+        matches!(
+            event_2_v5,
+            discv5::Event::SessionEstablished(node, socket) if node == node_1_enr && socket == node_1_enr.udp4_socket().unwrap().into()
+        );
+
+        // verify node_1 is in KBuckets of node_2:discv5
+        let event_2_v5 = stream_2.recv().await.unwrap();
+        matches!(
+            event_2_v5,
+            discv5::Event::NodeInserted { node_id, replaced } if node_id == node_1_enr.node_id() && replaced.is_none()
+        );
+    }
+
+    #[test]
+    fn discovered_enr_disc_socket_missing() {
+        reth_tracing::init_test_tracing();
+
+        // rig test
+        const REMOTE_RLPX_PORT: u16 = 30303;
+        let remote_socket = "104.28.44.25:9000".parse().unwrap();
+        let remote_key = CombinedKey::generate_secp256k1();
+        let remote_enr = Enr::builder().tcp4(REMOTE_RLPX_PORT).build(&remote_key).unwrap();
+
+        let mut discv5 = discv5_noop();
+
+        // test
+        let filtered_peer = discv5.on_discovered_peer(&remote_enr, remote_socket);
+
+        assert_eq!(
+            NodeRecord {
+                address: remote_socket.ip(),
+                udp_port: remote_socket.port(),
+                tcp_port: REMOTE_RLPX_PORT,
+                id: enr_to_discv4_id(&remote_enr).unwrap(),
+            },
+            filtered_peer.unwrap().node_record
+        )
+    }
+
+    // Copied from sigp/discv5 with slight modification (U256 type)
+    // <https://github.com/sigp/discv5/blob/master/src/kbucket/key.rs#L89-L101>
+    #[allow(unreachable_pub)]
+    #[allow(unused)]
+    #[allow(clippy::assign_op_pattern)]
+    mod sigp {
+        use enr::{
+            k256::sha2::digest::generic_array::{typenum::U32, GenericArray},
+            NodeId,
+        };
+        use reth_primitives::U256;
+
+        /// A `Key` is a cryptographic hash, identifying both the nodes participating in
+        /// the Kademlia DHT, as well as records stored in the DHT.
+        ///
+        /// The set of all `Key`s defines the Kademlia keyspace.
+        ///
+        /// `Key`s have an XOR metric as defined in the Kademlia paper, i.e. the bitwise XOR of
+        /// the hash digests, interpreted as an integer. See [`Key::distance`].
+        ///
+        /// A `Key` preserves the preimage of type `T` of the hash function. See [`Key::preimage`].
+        #[derive(Clone, Debug)]
+        pub struct Key<T> {
+            preimage: T,
+            hash: GenericArray<u8, U32>,
+        }
+
+        impl<T> PartialEq for Key<T> {
+            fn eq(&self, other: &Key<T>) -> bool {
+                self.hash == other.hash
+            }
+        }
+
+        impl<T> Eq for Key<T> {}
+
+        impl<TPeerId> AsRef<Key<TPeerId>> for Key<TPeerId> {
+            fn as_ref(&self) -> &Key<TPeerId> {
+                self
+            }
+        }
+
+        impl<T> Key<T> {
+            /// Construct a new `Key` by providing the raw 32 byte hash.
+            pub fn new_raw(preimage: T, hash: GenericArray<u8, U32>) -> Key<T> {
+                Key { preimage, hash }
+            }
+
+            /// Borrows the preimage of the key.
+            pub fn preimage(&self) -> &T {
+                &self.preimage
+            }
+
+            /// Converts the key into its preimage.
+            pub fn into_preimage(self) -> T {
+                self.preimage
+            }
+
+            /// Computes the distance of the keys according to the XOR metric.
+            pub fn distance<U>(&self, other: &Key<U>) -> Distance {
+                let a = U256::from_be_slice(self.hash.as_slice());
+                let b = U256::from_be_slice(other.hash.as_slice());
+                Distance(a ^ b)
+            }
+
+            // Used in the FINDNODE query outside of the k-bucket implementation.
+            /// Computes the integer log-2 distance between two keys, assuming a 256-bit
+            /// key. The output returns None if the key's are identical. The range is 1-256.
+            pub fn log2_distance<U>(&self, other: &Key<U>) -> Option<u64> {
+                let xor_dist = self.distance(other);
+                let log_dist = (256 - xor_dist.0.leading_zeros() as u64);
+                if log_dist == 0 {
+                    None
+                } else {
+                    Some(log_dist)
+                }
+            }
+        }
+
+        impl From<NodeId> for Key<NodeId> {
+            fn from(node_id: NodeId) -> Self {
+                Key { preimage: node_id, hash: *GenericArray::from_slice(&node_id.raw()) }
+            }
+        }
+
+        /// A distance between two `Key`s.
+        #[derive(Copy, Clone, PartialEq, Eq, Default, PartialOrd, Ord, Debug)]
+        pub struct Distance(pub(super) U256);
+    }
+
+    #[test]
+    fn select_lookup_target() {
+        // distance ceiled to the next byte
+        const fn expected_log2_distance(log2_distance: usize) -> u64 {
+            let log2_distance = log2_distance / 8;
+            ((log2_distance + 1) * 8) as u64
+        }
+
+        let log2_distance = rand::thread_rng().gen_range(0..=MAX_LOG2_DISTANCE);
+
+        let sk = CombinedKey::generate_secp256k1();
+        let local_node_id = discv5::enr::NodeId::from(sk.public());
+        let target = get_lookup_target(log2_distance, local_node_id);
+
+        let local_node_id = sigp::Key::from(local_node_id);
+        let target = sigp::Key::from(target);
+
+        assert_eq!(
+            expected_log2_distance(log2_distance),
+            local_node_id.log2_distance(&target).unwrap()
+        );
+    }
+}

crates/net/discv5/src/metrics.rs

@@ -0,0 +1,117 @@
+//! Tracks peer discovery for [`Discv5`](crate::Discv5).
+use metrics::{Counter, Gauge};
+use reth_metrics::Metrics;
+
+use crate::config::{ETH, ETH2, OPSTACK};
+
+/// Information tracked by [`Discv5`](crate::Discv5).
+#[derive(Debug, Default, Clone)]
+pub struct Discv5Metrics {
+    /// Frequency of networks advertised in discovered peers' node records.
+    pub discovered_peers_advertised_networks: AdvertisedChainMetrics,
+    /// Tracks discovered peers.
+    pub discovered_peers: DiscoveredPeersMetrics,
+}
+
+/// Tracks discovered peers.
+#[derive(Metrics, Clone)]
+#[metrics(scope = "discv5")]
+pub struct DiscoveredPeersMetrics {
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    // Kbuckets
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    /// Total peers currently in [`discv5::Discv5`]'s kbuckets.
+    total_kbucket_peers_raw: Gauge,
+    /// Total discovered peers that are inserted into [`discv5::Discv5`]'s kbuckets.
+    ///
+    /// This is a subset of the total established sessions, in which all peers advertise a udp
+    /// socket in their node record which is reachable from the local node. Only these peers make
+    /// it into [`discv5::Discv5`]'s kbuckets and will hence be included in queries.
+    ///
+    /// Note: the definition of 'discovered' is not exactly synonymous in `reth_discv4::Discv4`.
+    total_inserted_kbucket_peers_raw: Counter,
+
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    // Sessions
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    /// Total peers currently connected to [`discv5::Discv5`].
+    total_sessions_raw: Gauge,
+    /// Total number of sessions established by [`discv5::Discv5`].
+    total_established_sessions_raw: Counter,
+    /// Total number of sessions established by [`discv5::Discv5`], with peers that don't advertise
+    /// a socket which is reachable from the local node in their node record.
+    ///
+    /// These peers can't make it into [`discv5::Discv5`]'s kbuckets, and hence won't be part of
+    /// queries (neither shared with peers in NODES responses, nor queried for peers with FINDNODE
+    /// requests).
+    total_established_sessions_unreachable_enr: Counter,
+    /// Total number of sessions established by [`discv5::Discv5`], that pass configured
+    /// [`filter`](crate::filter) rules.
+    total_established_sessions_custom_filtered: Counter,
+}
+
+impl DiscoveredPeersMetrics {
+    /// Sets current total number of peers in [`discv5::Discv5`]'s kbuckets.
+    pub fn set_total_kbucket_peers(&mut self, num: usize) {
+        self.total_kbucket_peers_raw.set(num as f64)
+    }
+
+    /// Increments the number of kbucket insertions in [`discv5::Discv5`].
+    pub fn increment_kbucket_insertions(&mut self, num: u64) {
+        self.total_inserted_kbucket_peers_raw.increment(num)
+    }
+
+    /// Sets current total number of peers connected to [`discv5::Discv5`].
+    pub fn set_total_sessions(&mut self, num: usize) {
+        self.total_sessions_raw.set(num as f64)
+    }
+
+    /// Increments number of sessions established by [`discv5::Discv5`].
+    pub fn increment_established_sessions_raw(&mut self, num: u64) {
+        self.total_established_sessions_raw.increment(num)
+    }
+
+    /// Increments number of sessions established by [`discv5::Discv5`], with peers that don't have
+    /// a reachable node record.
+    pub fn increment_established_sessions_unreachable_enr(&mut self, num: u64) {
+        self.total_established_sessions_unreachable_enr.increment(num)
+    }
+
+    /// Increments number of sessions established by [`discv5::Discv5`], that pass configured
+    /// [`filter`](crate::filter) rules.
+    pub fn increment_established_sessions_filtered(&mut self, num: u64) {
+        self.total_established_sessions_custom_filtered.increment(num)
+    }
+}
+
+/// Tracks frequency of networks that are advertised by discovered peers.
+///
+/// Peers advertise the chain they belong to as a kv-pair in their node record, using the network
+/// as key.
+#[derive(Metrics, Clone)]
+#[metrics(scope = "discv5")]
+pub struct AdvertisedChainMetrics {
+    /// Frequency of node records with a kv-pair with [`OPSTACK`] as key.
+    opstack: Counter,
+
+    /// Frequency of node records with a kv-pair with [`ETH`] as key.
+    eth: Counter,
+
+    /// Frequency of node records with a kv-pair with [`ETH2`] as key.
+    eth2: Counter,
+}
+
+impl AdvertisedChainMetrics {
+    /// Counts each recognised network type that is advertised on node record, once.
+    pub fn increment_once_by_network_type(&mut self, enr: &discv5::Enr) {
+        if enr.get_raw_rlp(OPSTACK).is_some() {
+            self.opstack.increment(1u64)
+        }
+        if enr.get_raw_rlp(ETH).is_some() {
+            self.eth.increment(1u64)
+        }
+        if enr.get_raw_rlp(ETH2).is_some() {
+            self.eth2.increment(1u64)
+        }
+    }
+}

crates/primitives/src/chain/spec.rs

@@ -738,6 +738,13 @@ impl ChainSpec {
         self.hardfork_fork_id(Hardfork::Cancun)
     }
 
+    /// Convenience method to get the latest fork id from the chainspec. Panics if chainspec has no
+    /// hardforks.
+    #[inline]
+    pub fn latest_fork_id(&self) -> ForkId {
+        self.hardfork_fork_id(*self.hardforks().last_key_value().unwrap().0).unwrap()
+    }
+
     /// Get the fork condition for the given fork.
     pub fn fork(&self, fork: Hardfork) -> ForkCondition {
         self.hardforks.get(&fork).copied().unwrap_or(ForkCondition::Never)
@@ -3158,4 +3165,21 @@ Post-merge hard forks (timestamp based):
         // <https://optimism-sepolia.blockscout.com/block/1>
         assert_eq!(base_fee, 980000000);
     }
+
+    #[test]
+    fn latest_eth_mainnet_fork_id() {
+        assert_eq!(
+            ForkId { hash: ForkHash([0x9f, 0x3d, 0x22, 0x54]), next: 0 },
+            MAINNET.latest_fork_id()
+        )
+    }
+
+    #[cfg(feature = "optimism")]
+    #[test]
+    fn latest_op_mainnet_fork_id() {
+        assert_eq!(
+            ForkId { hash: ForkHash([0x51, 0xcc, 0x98, 0xb3]), next: 0 },
+            BASE_MAINNET.latest_fork_id()
+        )
+    }
 }

crates/primitives/src/lib.rs

@@ -71,8 +71,9 @@ pub use header::{Header, HeaderValidationError, HeadersDirection, SealedHeader};
 pub use integer_list::IntegerList;
 pub use log::{logs_bloom, Log};
 pub use net::{
-    goerli_nodes, holesky_nodes, mainnet_nodes, parse_nodes, sepolia_nodes, NodeRecord,
-    GOERLI_BOOTNODES, HOLESKY_BOOTNODES, MAINNET_BOOTNODES, SEPOLIA_BOOTNODES,
+    goerli_nodes, holesky_nodes, mainnet_nodes, parse_nodes, pk_to_id, sepolia_nodes, NodeRecord,
+    NodeRecordParseError, GOERLI_BOOTNODES, HOLESKY_BOOTNODES, MAINNET_BOOTNODES,
+    SEPOLIA_BOOTNODES,
 };
 pub use peer::{id2pk, pk2id, AnyNode, PeerId, WithPeerId};
 pub use prune::{

crates/primitives/src/net.rs

@@ -1,4 +1,4 @@
-pub use reth_rpc_types::NodeRecord;
+pub use reth_rpc_types::{pk_to_id, NodeRecord, NodeRecordParseError};
 
 // <https://github.com/ledgerwatch/erigon/blob/610e648dc43ec8cd6563313e28f06f534a9091b3/params/bootnodes.go>
 

crates/rpc/rpc-types/src/net.rs

@@ -1,5 +1,6 @@
-use crate::PeerId;
+use crate::{pk_to_id, PeerId};
 use alloy_rlp::{RlpDecodable, RlpEncodable};
+use enr::Enr;
 use secp256k1::{SecretKey, SECP256K1};
 use serde_with::{DeserializeFromStr, SerializeDisplay};
 use std::{
@@ -9,6 +10,7 @@ use std::{
     num::ParseIntError,
     str::FromStr,
 };
+use thiserror::Error;
 use url::{Host, Url};
 
 /// Represents a ENR in discovery.
@@ -114,8 +116,8 @@ impl fmt::Display for NodeRecord {
     }
 }
 
-/// Possible error types when parsing a `NodeRecord`
-#[derive(Debug, thiserror::Error)]
+/// Possible error types when parsing a [`NodeRecord`]
+#[derive(Debug, Error)]
 pub enum NodeRecordParseError {
     /// Invalid url
     #[error("Failed to parse url: {0}")]
@@ -165,6 +167,29 @@ impl FromStr for NodeRecord {
     }
 }
 
+impl TryFrom<&Enr<SecretKey>> for NodeRecord {
+    type Error = NodeRecordParseError;
+
+    fn try_from(enr: &Enr<SecretKey>) -> Result<Self, Self::Error> {
+        let Some(address) = enr.ip4().map(IpAddr::from).or_else(|| enr.ip6().map(IpAddr::from))
+        else {
+            return Err(NodeRecordParseError::InvalidUrl("ip missing".to_string()))
+        };
+
+        let Some(udp_port) = enr.udp4().or_else(|| enr.udp6()) else {
+            return Err(NodeRecordParseError::InvalidUrl("udp port missing".to_string()))
+        };
+
+        let Some(tcp_port) = enr.tcp4().or_else(|| enr.tcp6()) else {
+            return Err(NodeRecordParseError::InvalidUrl("tcp port missing".to_string()))
+        };
+
+        let id = pk_to_id(&enr.public_key());
+
+        Ok(NodeRecord { address, tcp_port, udp_port, id }.into_ipv4_mapped())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/rpc/rpc-types/src/peer.rs

@@ -2,3 +2,8 @@ use alloy_primitives::B512;
 
 /// Alias for a peer identifier
 pub type PeerId = B512;
+
+/// Converts a [`secp256k1::PublicKey`] to a [`PeerId`].
+pub fn pk_to_id(pk: &secp256k1::PublicKey) -> PeerId {
+    PeerId::from_slice(&pk.serialize_uncompressed()[1..])
+}