diff --git a/Cargo.lock b/Cargo.lock index c18802de01f..9a7f4d4bc61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -151,7 +151,7 @@ dependencies = [ "alloy-eips", "alloy-primitives", "alloy-rlp", - "sha2", + "sha2 0.10.8", ] [[package]] @@ -612,12 +612,24 @@ dependencies = [ "rand 0.8.5", ] +[[package]] +name = "arrayref" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b4930d2cb77ce62f89ee5d5289b4ac049559b1c45539271f5ed4fdc7db34545" + [[package]] name = "arrayvec" version = "0.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96d30a06541fbafbc7f82ed10c06164cfbd2c401138f6addd8404629c4b16711" +[[package]] +name = "asn1_der" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "155a5a185e42c6b77ac7b88a15143d930a9e9727a5b7b77eed417404ab15c247" + [[package]] name = "assert_matches" version = "1.5.0" @@ -783,6 +795,12 @@ dependencies = [ "rustc-demangle", ] +[[package]] +name = "base-x" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cbbc9d0964165b47557570cce6c952866c2678457aca742aafc9fb771d30270" + [[package]] name = "base16ct" version = "0.2.0" @@ -1110,6 +1128,15 @@ dependencies = [ "alloc-stdlib", ] +[[package]] +name = "bs58" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf88ba1141d185c399bee5288d850d63b8369520c1eafc32a0430b5b6c287bf4" +dependencies = [ + "tinyvec", +] + [[package]] name = "bstr" version = "0.2.17" @@ -1487,6 +1514,15 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f" +[[package]] +name = "core2" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b49ba7ef1ad6107f8824dbe97de947cbaac53c44e7f9756a1fba0d37c1eec505" +dependencies = [ + "memchr", +] + [[package]] name = "cpp_demangle" version = "0.4.3" @@ -1654,6 +1690,16 @@ dependencies = [ "typenum", ] +[[package]] +name = "crypto-mac" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b584a330336237c1eecd3e94266efb216c56ed91225d634cb2991c5f3fd1aeab" +dependencies = [ + "generic-array", + "subtle", +] + [[package]] name = "ctr" version = "0.7.0" @@ -1934,6 +1980,26 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e962a19be5cfc3f3bf6dd8f61eb50107f356ad6270fbb3ed41476571db78be5" +[[package]] +name = "data-encoding-macro" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20c01c06f5f429efdf2bae21eb67c28b3df3cf85b7dd2d8ef09c0838dac5d33e" +dependencies = [ + "data-encoding", + "data-encoding-macro-internal", +] + +[[package]] +name = "data-encoding-macro-internal" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0047d07f2c89b17dd631c80450d69841a6b5d7fb17278cbc43d7e4cfcf2576f3" +dependencies = [ + "data-encoding", + "syn 1.0.109", +] + [[package]] name = "debug-helper" version = "0.3.13" @@ -2114,8 +2180,7 @@ dependencies = [ [[package]] name = "discv5" version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bac33cb3f99889a57e56a8c6ccb77aaf0cfc7787602b7af09783f736d77314e1" +source = "git+https://github.com/sigp/discv5?rev=04ac004#04ac0042a345a9edf93b090007e5d31c008261ed" dependencies = [ "aes 0.7.5", "aes-gcm", @@ -2128,6 +2193,7 @@ dependencies = [ "hex", "hkdf", "lazy_static", + "libp2p", "lru", "more-asserts", "parking_lot 0.11.2", @@ -2222,7 +2288,7 @@ dependencies = [ "ed25519", "rand_core 0.6.4", "serde", - "sha2", + "sha2 0.10.8", "subtle", "zeroize", ] @@ -2724,6 +2790,7 @@ dependencies = [ "futures-core", "futures-task", "futures-util", + "num_cpus", ] [[package]] @@ -3064,7 +3131,17 @@ version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" dependencies = [ - "hmac", + "hmac 0.12.1", +] + +[[package]] +name = "hmac" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "126888268dcc288495a26bf004b38c5fdbb31682f992c84ceb046a1f0fe38840" +dependencies = [ + "crypto-mac", + "digest 0.9.0", ] [[package]] @@ -3076,6 +3153,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "hmac-drbg" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17ea0a1394df5b6574da6e0c1ade9e78868c9fb0a4e5ef4428e32da4676b85b1" +dependencies = [ + "digest 0.9.0", + "generic-array", + "hmac 0.8.1", +] + [[package]] name = "hostname" version = "0.3.1" @@ -3835,7 +3923,7 @@ dependencies = [ "ecdsa", "elliptic-curve", "once_cell", - "sha2", + "sha2 0.10.8", "signature", ] @@ -3914,6 +4002,122 @@ version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058" +[[package]] +name = "libp2p" +version = "0.53.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "681fb3f183edfbedd7a57d32ebe5dcdc0b9f94061185acf3c30249349cc6fc99" +dependencies = [ + "bytes", + "either", + "futures", + "futures-timer", + "getrandom 0.2.12", + "instant", + "libp2p-allow-block-list", + "libp2p-connection-limits", + "libp2p-core", + "libp2p-identity", + "libp2p-swarm", + "multiaddr", + "pin-project", + "rw-stream-sink", + "thiserror", +] + +[[package]] +name = "libp2p-allow-block-list" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "107b238b794cb83ab53b74ad5dcf7cca3200899b72fe662840cfb52f5b0a32e6" +dependencies = [ + "libp2p-core", + "libp2p-identity", + "libp2p-swarm", + "void", +] + +[[package]] +name = "libp2p-connection-limits" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7cd50a78ccfada14de94cbacd3ce4b0138157f376870f13d3a8422cd075b4fd" +dependencies = [ + "libp2p-core", + "libp2p-identity", + "libp2p-swarm", + "void", +] + +[[package]] +name = "libp2p-core" +version = "0.41.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8130a8269e65a2554d55131c770bdf4bcd94d2b8d4efb24ca23699be65066c05" +dependencies = [ + "either", + "fnv", + "futures", + "futures-timer", + "instant", + "libp2p-identity", + "multiaddr", + "multihash", + "multistream-select", + "once_cell", + "parking_lot 0.12.1", + "pin-project", + "quick-protobuf", + "rand 0.8.5", + "rw-stream-sink", + "smallvec", + "thiserror", + "tracing", + "unsigned-varint 0.8.0", + "void", +] + +[[package]] +name = "libp2p-identity" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "999ec70441b2fb35355076726a6bc466c932e9bdc66f6a11c6c0aa17c7ab9be0" +dependencies = [ + "asn1_der", + "bs58", + "ed25519-dalek", + "hkdf", + "libsecp256k1", + "multihash", + "quick-protobuf", + "rand 0.8.5", + "sha2 0.10.8", + "thiserror", + "tracing", + "zeroize", +] + +[[package]] +name = "libp2p-swarm" +version = "0.44.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e92532fc3c4fb292ae30c371815c9b10103718777726ea5497abc268a4761866" +dependencies = [ + "either", + "fnv", + "futures", + "futures-timer", + "instant", + "libp2p-core", + "libp2p-identity", + "multistream-select", + "once_cell", + "rand 0.8.5", + "smallvec", + "tracing", + "void", +] + [[package]] name = "libproc" version = "0.14.6" @@ -3936,6 +4140,54 @@ dependencies = [ "redox_syscall 0.4.1", ] +[[package]] +name = "libsecp256k1" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95b09eff1b35ed3b33b877ced3a691fc7a481919c7e29c53c906226fcf55e2a1" +dependencies = [ + "arrayref", + "base64 0.13.1", + "digest 0.9.0", + "hmac-drbg", + "libsecp256k1-core", + "libsecp256k1-gen-ecmult", + "libsecp256k1-gen-genmult", + "rand 0.8.5", + "serde", + "sha2 0.9.9", + "typenum", +] + +[[package]] +name = "libsecp256k1-core" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5be9b9bb642d8522a44d533eab56c16c738301965504753b03ad1de3425d5451" +dependencies = [ + "crunchy", + "digest 0.9.0", + "subtle", +] + +[[package]] +name = "libsecp256k1-gen-ecmult" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3038c808c55c87e8a172643a7d87187fc6c4174468159cb3090659d55bcb4809" +dependencies = [ + "libsecp256k1-core", +] + +[[package]] +name = "libsecp256k1-gen-genmult" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3db8d6ba2cec9eacc40e6e8ccc98931840301f1006e95647ceb2dd5c3aa06f7c" +dependencies = [ + "libsecp256k1-core", +] + [[package]] name = "linked-hash-map" version = "0.5.6" @@ -4273,6 +4525,60 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fafa6961cabd9c63bcd77a45d7e3b7f3b552b70417831fb0f56db717e72407e" +[[package]] +name = "multiaddr" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b852bc02a2da5feed68cd14fa50d0774b92790a5bdbfa932a813926c8472070" +dependencies = [ + "arrayref", + "byteorder", + "data-encoding", + "libp2p-identity", + "multibase", + "multihash", + "percent-encoding", + "serde", + "static_assertions", + "unsigned-varint 0.7.2", + "url", +] + +[[package]] +name = "multibase" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b3539ec3c1f04ac9748a260728e855f261b4977f5c3406612c884564f329404" +dependencies = [ + "base-x", + "data-encoding", + "data-encoding-macro", +] + +[[package]] +name = "multihash" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "076d548d76a0e2a0d4ab471d0b1c36c577786dfc4471242035d97a12a735c492" +dependencies = [ + "core2", + "unsigned-varint 0.7.2", +] + +[[package]] +name = "multistream-select" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea0df8e5eec2298a62b326ee4f0d7fe1a6b90a09dfcf9df37b38f947a8c42f19" +dependencies = [ + "bytes", + "futures", + "log", + "pin-project", + "smallvec", + "unsigned-varint 0.7.2", +] + [[package]] name = "nibble_vec" version = "0.1.0" @@ -5132,6 +5438,15 @@ version = "1.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" +[[package]] +name = "quick-protobuf" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d6da84cc204722a989e01ba2f6e1e276e190f22263d0cb6ce8526fcdb0d2e1f" +dependencies = [ + "byteorder", +] + [[package]] name = "quick-xml" version = "0.26.0" @@ -5724,6 +6039,30 @@ dependencies = [ "tracing", ] +[[package]] +name = "reth-discv5" +version = "0.2.0-beta.4" +dependencies = [ + "alloy-rlp", + "derive_more", + "discv5", + "enr", + "futures", + "itertools 0.12.1", + "libp2p-identity", + "metrics", + "multiaddr", + "rand 0.8.5", + "reth-metrics", + "reth-primitives", + "reth-tracing", + "rlp", + "secp256k1 0.27.0", + "thiserror", + "tokio", + "tracing", +] + [[package]] name = "reth-dns-discovery" version = "0.2.0-beta.4" @@ -5791,13 +6130,13 @@ dependencies = [ "educe", "futures", "generic-array", - "hmac", + "hmac 0.12.1", "pin-project", "rand 0.8.5", "reth-net-common", "reth-primitives", "secp256k1 0.27.0", - "sha2", + "sha2 0.10.8", "sha3", "thiserror", "tokio", @@ -6293,7 +6632,7 @@ dependencies = [ "reth-rpc-types-compat", "reth-transaction-pool", "revm", - "sha2", + "sha2 0.10.8", "thiserror", "tracing", ] @@ -6316,7 +6655,7 @@ dependencies = [ "revm", "revm-primitives", "serde_json", - "sha2", + "sha2 0.10.8", "thiserror", "tokio", "tokio-stream", @@ -6374,7 +6713,7 @@ dependencies = [ "serde", "serde_json", "serde_with", - "sha2", + "sha2 0.10.8", "strum 0.26.2", "sucds", "tempfile", @@ -6881,7 +7220,7 @@ dependencies = [ "revm-primitives", "ripemd", "secp256k1 0.28.2", - "sha2", + "sha2 0.10.8", "substrate-bn", ] @@ -6912,7 +7251,7 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" dependencies = [ - "hmac", + "hmac 0.12.1", "subtle", ] @@ -7175,6 +7514,17 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "rw-stream-sink" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8c9026ff5d2f23da5e45bbc283f156383001bfb09c4e44256d02c1a685fe9a1" +dependencies = [ + "futures", + "pin-project", + "static_assertions", +] + [[package]] name = "ryu" version = "1.0.17" @@ -7523,6 +7873,19 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha2" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d58a1e1bf39749807d89cf2d98ac2dfa0ff1cb3faa38fbb64dd88ac8013d800" +dependencies = [ + "block-buffer 0.9.0", + "cfg-if", + "cpufeatures", + "digest 0.9.0", + "opaque-debug", +] + [[package]] name = "sha2" version = "0.10.8" @@ -8775,6 +9138,18 @@ dependencies = [ "subtle", ] +[[package]] +name = "unsigned-varint" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105" + +[[package]] +name = "unsigned-varint" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06" + [[package]] name = "untrusted" version = "0.7.1" @@ -8858,6 +9233,12 @@ version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" +[[package]] +name = "void" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d" + [[package]] name = "wait-timeout" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 492a5f85b8d..52251dff289 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ members = [ "crates/metrics/metrics-derive/", "crates/net/common/", "crates/net/discv4/", + "crates/net/discv5/", "crates/net/dns/", "crates/net/downloaders/", "crates/net/ecies/", @@ -199,6 +200,7 @@ reth-config = { path = "crates/config" } reth-consensus-common = { path = "crates/consensus/common" } reth-db = { path = "crates/storage/db" } reth-discv4 = { path = "crates/net/discv4" } +reth-discv5 = { path = "crates/net/discv5" } reth-dns-discovery = { path = "crates/net/dns" } reth-node-builder = { path = "crates/node-builder" } reth-node-ethereum = { path = "crates/node-ethereum" } @@ -320,7 +322,7 @@ tower = "0.4" tower-http = "0.4" # p2p -discv5 = "0.4" +discv5 = { git = "https://github.com/sigp/discv5", rev = "04ac004" } igd-next = "0.14.3" # rpc diff --git a/crates/net/discv5/Cargo.toml b/crates/net/discv5/Cargo.toml new file mode 100644 index 00000000000..03b856be9a0 --- /dev/null +++ b/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"] } diff --git a/crates/net/discv5/README.md b/crates/net/discv5/README.md new file mode 100644 index 00000000000..03c031e7924 --- /dev/null +++ b/crates/net/discv5/README.md @@ -0,0 +1,3 @@ +# Discv5 + +Thin wrapper around sigp/discv5. \ No newline at end of file diff --git a/crates/net/discv5/src/config.rs b/crates/net/discv5/src/config.rs new file mode 100644 index 00000000000..1fe25c482da --- /dev/null +++ b/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, + /// Nodes to boot from. + bootstrap_nodes: HashSet, + /// [`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, + /// 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, +} + +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) -> 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::()) { + 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) -> 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, + /// [`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 { + 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::().unwrap(); + let socket_2 = "107.21.251.55:9222".parse::().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::>(); + + for node in MULTI_ADDRESSES.split(&[',']) { + assert!(bootstrap_nodes.contains(&node.to_string())); + } + } +} diff --git a/crates/net/discv5/src/enr.rs b/crates/net/discv5/src/enr.rs new file mode 100644 index 00000000000..51323b040a9 --- /dev/null +++ b/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 { + 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 { + Ok(id2pk(peer_id)?.into()) +} + +/// Converts a [`PeerId`] to a [`libp2p_identity::PeerId `]. +pub fn discv4_id_to_multiaddr_id( + peer_id: PeerId, +) -> Result { + 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`]). +#[derive(Debug, Clone)] +pub struct EnrCombinedKeyWrapper(pub discv5::Enr); + +impl From> for EnrCombinedKeyWrapper { + fn from(value: Enr) -> Self { + let encoded_enr = rlp::encode(&value); + let enr = rlp::decode::(&encoded_enr).unwrap(); + + Self(enr) + } +} + +impl From for Enr { + fn from(val: EnrCombinedKeyWrapper) -> Self { + let EnrCombinedKeyWrapper(enr) = val; + let encoded_enr = rlp::encode(&enr); + + rlp::decode::>(&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 + ) + } +} diff --git a/crates/net/discv5/src/error.rs b/crates/net/discv5/src/error.rs new file mode 100644 index 00000000000..96929b79363 --- /dev/null +++ b/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), +} diff --git a/crates/net/discv5/src/filter.rs b/crates/net/discv5/src/filter.rs new file mode 100644 index 00000000000..5cb7be18c60 --- /dev/null +++ b/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, +} + +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 { .. })); + } +} diff --git a/crates/net/discv5/src/lib.rs b/crates/net/discv5/src/lib.rs new file mode 100644 index 00000000000..360bed68bb9 --- /dev/null +++ b/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, + /// [`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) -> 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, 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, + 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, 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, + discv5: &Arc, + ) -> 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::>() + ).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, + ) { + // 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 { + 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 { + 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 { + 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( + &self, + enr: &discv5::enr::Enr, + ) -> Result { + 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(&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, +} + +/// 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::(); + } + } + } + + 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, 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 = + 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) + // + #[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 { + preimage: T, + hash: GenericArray, + } + + impl PartialEq for Key { + fn eq(&self, other: &Key) -> bool { + self.hash == other.hash + } + } + + impl Eq for Key {} + + impl AsRef> for Key { + fn as_ref(&self) -> &Key { + self + } + } + + impl Key { + /// Construct a new `Key` by providing the raw 32 byte hash. + pub fn new_raw(preimage: T, hash: GenericArray) -> Key { + 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(&self, other: &Key) -> 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(&self, other: &Key) -> Option { + 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 for Key { + 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() + ); + } +} diff --git a/crates/net/discv5/src/metrics.rs b/crates/net/discv5/src/metrics.rs new file mode 100644 index 00000000000..e38fa0fae17 --- /dev/null +++ b/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) + } + } +} diff --git a/crates/primitives/src/chain/spec.rs b/crates/primitives/src/chain/spec.rs index c583b7c212e..cb7ef8522b4 100644 --- a/crates/primitives/src/chain/spec.rs +++ b/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): // 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() + ) + } } diff --git a/crates/primitives/src/lib.rs b/crates/primitives/src/lib.rs index 7cb3d054c83..8e548a233f3 100644 --- a/crates/primitives/src/lib.rs +++ b/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::{ diff --git a/crates/primitives/src/net.rs b/crates/primitives/src/net.rs index 7d122f71a18..2e0b77d5099 100644 --- a/crates/primitives/src/net.rs +++ b/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}; // diff --git a/crates/rpc/rpc-types/src/net.rs b/crates/rpc/rpc-types/src/net.rs index c5d2f72e392..3fc3d74991a 100644 --- a/crates/rpc/rpc-types/src/net.rs +++ b/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> for NodeRecord { + type Error = NodeRecordParseError; + + fn try_from(enr: &Enr) -> Result { + 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::*; diff --git a/crates/rpc/rpc-types/src/peer.rs b/crates/rpc/rpc-types/src/peer.rs index a07e61d0028..44dbe5d71f2 100644 --- a/crates/rpc/rpc-types/src/peer.rs +++ b/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..]) +} diff --git a/deny.toml b/deny.toml index c0fc5392151..347b609651f 100644 --- a/deny.toml +++ b/deny.toml @@ -90,4 +90,5 @@ allow-git = [ # TODO: remove, see ./Cargo.toml "https://github.com/alloy-rs/alloy", "https://github.com/paradigmxyz/evm-inspectors", + "https://github.com/sigp/discv5", ]