diff --git a/Cargo.lock b/Cargo.lock index 73022c62..1df5b0f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5584,6 +5584,7 @@ dependencies = [ "url", "vise", "vise-exporter", + "wiremock", ] [[package]] diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index 6e3da8db..c8345bfe 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -122,10 +122,9 @@ pub(crate) async fn wire_p2p( // TODO: also send the post-#4130 `Cluster-Uuid` header (relay-side load // balancing), pending in pluto-p2p's `new_relays`. - let relay_addrs = bootnode::relay_addrs_for_resolution(&p2p_config.relays); let relays = bootnode::new_relays( cancellation.clone(), - &relay_addrs, + &p2p_config.relays, &crate::utils::hex_7(&lock_hash), ) .await?; diff --git a/crates/cli/src/commands/common.rs b/crates/cli/src/commands/common.rs index 71280e8d..9a430943 100644 --- a/crates/cli/src/commands/common.rs +++ b/crates/cli/src/commands/common.rs @@ -1,8 +1,9 @@ //! Shared helpers for CLI commands. -use std::str::FromStr; +use pluto_p2p::config::RelayAddr; +use tracing::warn; -use libp2p::{Multiaddr, multiaddr}; +use crate::error::CliError; /// Shared license notice shown by long-running commands. pub const LICENSE: &str = concat!( @@ -51,7 +52,93 @@ pub fn build_console_tracing_config( builder.override_env_filter(level.into()).build() } -/// Parses a relay string as either a relay URL or a raw multiaddr. -pub fn parse_relay_addr(relay: &str) -> std::result::Result { - multiaddr::from_url(relay).or_else(|_| Multiaddr::from_str(relay)) +/// Parses the configured relay addresses, warning about insecure ones. +/// +/// Exactly one empty value (`--p2p-relays=""`) means "no relays". That is the +/// only accepted empty form: every other empty is an error, including interior +/// ones (`a,,b`, `,`) and an empty value repeated or mixed with real addresses +/// (`--p2p-relays="" --p2p-relays=https://x`, which flattens to the same list +/// as `--p2p-relays=,https://x` and so cannot be told apart from it). Each of +/// those is a field that was meant to hold an address; dropping them would +/// silently leave fewer relays configured than requested. +pub fn parse_relay_addrs(relays: &[String]) -> std::result::Result, CliError> { + if let [only] = relays + && only.is_empty() + { + return Ok(Vec::new()); + } + + let mut parsed = Vec::with_capacity(relays.len()); + + for relay in relays { + let addr: RelayAddr = relay.parse().map_err(|source| CliError::InvalidRelayAddr { + addr: relay.clone(), + source, + })?; + + // Warn once per plain-http relay while validating flags, before the P2P + // stack starts resolving them. + if addr.is_insecure_url() { + warn!(address = %relay, "Insecure relay address provided, not HTTPS"); + } + + parsed.push(addr); + } + + Ok(parsed) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Per-address parsing is covered by `RelayAddr`'s own tests; what is left + // to check here is the empty-value contract and the error wrapping. + + #[test] + fn treats_a_lone_empty_value_as_no_relays() { + // `--p2p-relays=""` is how relaying is turned off. + assert!( + parse_relay_addrs(&["".to_string()]) + .expect("relays") + .is_empty() + ); + assert!(parse_relay_addrs(&[]).expect("relays").is_empty()); + } + + #[test] + fn rejects_interior_empty_values() { + // `a,,b` splits to ["a", "", "b"] and `,` to ["", ""]. Dropping those + // empties would silently turn a typo'd flag into fewer relays, or none + // at all. + for relays in [ + vec!["https://relay.one".to_string(), String::new()], + vec![ + "https://relay.one".to_string(), + String::new(), + "https://relay.two".to_string(), + ], + vec![String::new(), String::new()], + ] { + let err = parse_relay_addrs(&relays).expect_err("empty entry should be rejected"); + + assert!( + err.to_string().contains("empty relay address"), + "unexpected error: {err}" + ); + } + } + + #[test] + fn rejects_invalid_relays() { + let err = parse_relay_addrs(&["not-an-address".to_string()]) + .expect_err("invalid relay should be rejected"); + + // The offending address must be named; the old error was just + // "Invalid multiaddr: invalid multiaddr". + assert!( + err.to_string().contains("not-an-address"), + "unexpected error: {err}" + ); + } } diff --git a/crates/cli/src/commands/dkg.rs b/crates/cli/src/commands/dkg.rs index 54d3f728..cc806ca1 100644 --- a/crates/cli/src/commands/dkg.rs +++ b/crates/cli/src/commands/dkg.rs @@ -3,13 +3,12 @@ use std::{future::Future, path::PathBuf}; use crate::{ - commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addr}, + commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addrs}, duration::Duration, error::{CliError, Result}, }; -use libp2p::multiaddr::Protocol; use tokio_util::sync::CancellationToken; -use tracing::{info, warn}; +use tracing::info; /// Arguments for the `dkg` command. #[derive(clap::Args, Clone, Debug)] @@ -127,17 +126,7 @@ impl TryFrom for pluto_dkg::dkg::Config { let tracing_config = build_console_tracing_config(args.log.level.clone(), &args.log.color, None); let p2p_config = { - let mut relays = Vec::new(); - - for relay in &args.p2p.relays { - let multiaddr = parse_relay_addr(relay)?; - - if multiaddr.iter().any(|protocol| protocol == Protocol::Http) { - warn!(address = %relay, "Insecure relay address provided, not HTTPS"); - } - - relays.push(multiaddr); - } + let relays = parse_relay_addrs(&args.p2p.relays)?; pluto_p2p::config::P2PConfig { relays, @@ -296,8 +285,8 @@ mod tests { use super::*; use crate::cli::{Cli, Commands}; use clap::Parser; - use libp2p::{Multiaddr, multiaddr}; - use std::{str::FromStr, sync::Arc, time::Duration as StdDuration}; + use pluto_p2p::config::RelayAddr; + use std::{sync::Arc, time::Duration as StdDuration}; #[test] fn dkg_is_registered_as_top_level_subcommand() { @@ -432,8 +421,8 @@ mod tests { assert_eq!( config.p2p.relays, vec![ - multiaddr::from_url("https://relay.one").expect("relay url"), - Multiaddr::from_str("/ip4/127.0.0.1/tcp/9000").expect("relay multiaddr") + RelayAddr::Url("https://relay.one".parse().expect("relay url")), + RelayAddr::Multiaddr("/ip4/127.0.0.1/tcp/9000".parse().expect("relay multiaddr")), ] ); assert_eq!(config.p2p.external_ip.as_deref(), Some("1.2.3.4")); diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 4ae7317d..eabc3556 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -1,8 +1,7 @@ use crate::{ - commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addr}, + commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addrs}, error::CliError, }; -use libp2p::multiaddr::Protocol; use pluto_p2p::k1; use std::{collections::HashMap, path::PathBuf, time::Duration}; use tokio_util::sync::CancellationToken; @@ -39,20 +38,7 @@ impl TryInto for RelayArgs { fn try_into(self) -> std::result::Result { let p2p_config = { - let mut relays = Vec::new(); - - for relay in &self.p2p.relays { - let multiaddr = parse_relay_addr(relay)?; - - if multiaddr.iter().any(|protocol| protocol == Protocol::Http) { - tracing::warn!( - address = %relay, - "Insecure relay address provided, not HTTPS" - ); - } - - relays.push(multiaddr); - } + let relays = parse_relay_addrs(&self.p2p.relays)?; pluto_p2p::config::P2PConfig { relays, diff --git a/crates/cli/src/commands/run.rs b/crates/cli/src/commands/run.rs index ccf6ea91..d808d9a3 100644 --- a/crates/cli/src/commands/run.rs +++ b/crates/cli/src/commands/run.rs @@ -32,14 +32,13 @@ use std::{ time::Duration as StdDuration, }; -use libp2p::multiaddr::Protocol; use pluto_eth2util::helpers::validate_http_headers; use pluto_featureset::{Feature, FeaturesetError, Status}; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use crate::{ - commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addr}, + commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addrs}, duration::Duration, error::{CliError, Result}, }; @@ -726,23 +725,7 @@ impl TryFrom for RunConfig { // --- p2p validation --- validate_hostname(p2p.external_host.as_deref())?; - let mut relays = Vec::with_capacity(p2p.relays.len()); - for relay in &p2p.relays { - // Charon treats `--p2p-relays=""` as "no relays"; clap's comma parser - // yields a single empty string, so skip empties to match (also handles - // stray empties like `a,,b`). - if relay.is_empty() { - continue; - } - - let multiaddr = parse_relay_addr(relay)?; - - if multiaddr.iter().any(|protocol| protocol == Protocol::Http) { - warn!(address = %relay, "Insecure relay address provided, not HTTPS"); - } - - relays.push(multiaddr); - } + let relays = parse_relay_addrs(&p2p.relays)?; // --- run-level validation --- if general.beacon_node_endpoints.is_empty() && !general.simnet_beacon_mock { @@ -1497,6 +1480,41 @@ mod tests { ); } + #[test] + fn run_accepts_relay_urls_with_a_path() { + // `http://relay:3640/enr` is the form the docker-compose relay serves; + // the path must reach the config intact rather than being rejected or + // truncated by a multiaddr round-trip. + let cases = [ + "http://relay:3640/enr", + "http://relay:3640", + "https://relay.example.org/enr", + "/ip4/127.0.0.1/tcp/3610/p2p/16Uiu2HAm7ULrTMdiEmQCJ2N9nsuGvfUDvfDGgHXJ4vNjrCwCzGDs", + ]; + + for case in cases { + let config = + parse_run(&[&format!("--p2p-relays={case}")]).expect("relay should be accepted"); + + assert_eq!( + config.p2p.relays, + vec![case.parse().expect("relay addr")], + "unexpected relays for {case}" + ); + } + } + + #[test] + fn run_rejects_invalid_relays() { + let err = parse_run(&["--p2p-relays=not-an-address"]) + .expect_err("invalid relay should be rejected"); + + assert!( + err.to_string().contains("not-an-address"), + "unexpected error: {err}" + ); + } + #[test] fn run_simnet_beacon_mock_satisfies_beacon_requirement() { let cli = Cli::try_parse_from(["pluto", "run", "--simnet-beacon-mock"]) diff --git a/crates/cli/src/commands/test/peers.rs b/crates/cli/src/commands/test/peers.rs index d1d41d1c..1c3c2f13 100644 --- a/crates/cli/src/commands/test/peers.rs +++ b/crates/cli/src/commands/test/peers.rs @@ -21,7 +21,7 @@ use pluto_k1util::load as load_key; use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode::new_relays, - config::{DEFAULT_RELAYS, P2PConfig}, + config::{DEFAULT_RELAYS, P2PConfig, RelayAddr}, gater::ConnGater, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -41,6 +41,7 @@ use super::{ write_result_to_writer, }; use crate::{ + commands::common::parse_relay_addrs, duration::Duration as CliDuration, error::{CliError, Result}, }; @@ -276,11 +277,13 @@ pub async fn run( disable_reuse_port: args.p2p_disable_reuseport, }; + let relay_addrs = parse_relay_addrs(&args.p2p_relays)?; + let (node, relay_peers) = setup_p2p( timeout_ct.clone(), private_key, p2p_cfg, - &args.p2p_relays, + &relay_addrs, &cluster_peers, self_peer_id, &enr_hash, @@ -308,7 +311,7 @@ pub async fn run( self_tests_clone, only_self_tests, ), - run_relay_http_tests(&args.p2p_relays, &relay_tests, timeout_ct.clone()), + run_relay_http_tests(&relay_addrs, &relay_tests, timeout_ct.clone()), ); let self_results = self_results.expect("self-test task should not panic"); let mut all_targets: HashMap> = HashMap::new(); @@ -435,8 +438,16 @@ fn peer_target_name(peer: &Peer, enr_str: &str) -> String { format!("peer {} {}", peer.name, format_enr(enr_str)) } +/// Probes every configured relay over HTTP. +/// +/// Targets are derived from the parsed addresses rather than the raw +/// `--p2p-relays` strings so that this and the P2P stack agree on what was +/// configured — probing the raw strings reported a bogus target when relaying +/// was disabled with `--p2p-relays=""`. Multiaddr relays are probed too: they +/// have no HTTP endpoint, so the probe reports a failure for them instead of +/// quietly leaving them untested. async fn run_relay_http_tests( - relay_urls: &[String], + relays: &[RelayAddr], queued: &[TestCaseName], ct: CancellationToken, ) -> HashMap> { @@ -444,10 +455,10 @@ async fn run_relay_http_tests( return HashMap::new(); } - let mut futs: FuturesUnordered<_> = relay_urls + let mut futs: FuturesUnordered<_> = relays .iter() - .map(|url| { - let url = url.clone(); + .map(|relay| { + let url = relay.to_string(); let ct = ct.clone(); let queued = queued.to_vec(); tokio::spawn(async move { @@ -1044,12 +1055,12 @@ async fn setup_p2p( cancel: CancellationToken, private_key: k256::SecretKey, p2p_cfg: P2PConfig, - relay_urls: &[String], + relay_addrs: &[RelayAddr], cluster_peers: &[Peer], self_peer_id: PeerId, enr_hash: &str, ) -> Result<(Node, Vec)> { - let relay_peers = new_relays(cancel.clone(), relay_urls, enr_hash).await?; + let relay_peers = new_relays(cancel.clone(), relay_addrs, enr_hash).await?; let mut all_peer_ids: Vec = cluster_peers.iter().map(|p| p.id).collect(); all_peer_ids.push(self_peer_id); @@ -1381,4 +1392,38 @@ mod tests { .collect(); assert_eq!(enrs, expected); } + + #[tokio::test] + async fn relay_http_tests_report_nothing_when_relaying_is_disabled() { + // `--p2p-relays=""` parses to no relays, so there is nothing to probe. + // Probing the raw flag strings instead used to key a target off the + // empty string and report it as a failing relay. + let relays = parse_relay_addrs(&["".to_string()]).expect("relays"); + let queued = [TestCaseName::new("PingRelay", 1)]; + + let results = run_relay_http_tests(&relays, &queued, CancellationToken::new()).await; + + assert!(results.is_empty(), "unexpected relay targets: {results:?}"); + } + + #[tokio::test] + async fn relay_http_tests_key_targets_by_address() { + let relays = parse_relay_addrs(&[ + "http://127.0.0.1:1/enr".to_string(), + "/ip4/127.0.0.1/tcp/3610/p2p/16Uiu2HAm7ULrTMdiEmQCJ2N9nsuGvfUDvfDGgHXJ4vNjrCwCzGDs" + .to_string(), + ]) + .expect("relays"); + let queued = [TestCaseName::new("PingRelay", 1)]; + + let results = run_relay_http_tests(&relays, &queued, CancellationToken::new()).await; + + // Both forms are probed, and the path survives into the target key. + assert_eq!(results.len(), 2); + assert!( + results.contains_key("relay http://127.0.0.1:1/enr"), + "unexpected targets: {:?}", + results.keys().collect::>() + ); + } } diff --git a/crates/cli/src/error.rs b/crates/cli/src/error.rs index 5e0ee100..c1ce261a 100644 --- a/crates/cli/src/error.rs +++ b/crates/cli/src/error.rs @@ -34,9 +34,15 @@ pub enum CliError { #[error("ENR generation failed: {0}")] EnrError(#[from] pluto_eth2util::enr::RecordError), - /// Invalid Multiaddr - #[error("Invalid multiaddr: {0}")] - InvalidMultiaddr(#[from] libp2p::multiaddr::Error), + /// Invalid relay URL or multiaddr. + #[error("parse relay address '{addr}': {source}")] + InvalidRelayAddr { + /// The offending relay address. + addr: String, + /// Why the address was rejected. + #[source] + source: pluto_p2p::config::RelayAddrError, + }, /// IO error occurred. #[error("IO error: {0}")] diff --git a/crates/consensus/examples/qbft.rs b/crates/consensus/examples/qbft.rs index ee17f925..39138c6d 100644 --- a/crates/consensus/examples/qbft.rs +++ b/crates/consensus/examples/qbft.rs @@ -98,7 +98,7 @@ use pluto_featureset::FeatureSet; use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode, - config::P2PConfig, + config::{P2PConfig, RelayAddr}, gater, k1, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -170,7 +170,7 @@ struct Args { /// Relay URLs or relay multiaddrs. #[arg(long, value_delimiter = ',')] - relays: Vec, + relays: Vec, /// TCP listen addresses. #[arg(long, value_delimiter = ',', default_value = "0.0.0.0:0")] diff --git a/crates/dkg/examples/bcast.rs b/crates/dkg/examples/bcast.rs index 02f2febc..eb717165 100644 --- a/crates/dkg/examples/bcast.rs +++ b/crates/dkg/examples/bcast.rs @@ -61,7 +61,7 @@ use pluto_dkg::bcast::{self, Component}; use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode, - config::P2PConfig, + config::{P2PConfig, RelayAddr}, gater, k1, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -88,7 +88,7 @@ struct ExampleBehaviour { struct Args { /// Relay URLs or relay multiaddrs to use. #[arg(long, value_delimiter = ',')] - relays: Vec, + relays: Vec, /// Data directory containing `charon-enr-private-key` and /// `cluster-lock.json`, typically one of the `nodeN/` directories produced diff --git a/crates/dkg/examples/sync.rs b/crates/dkg/examples/sync.rs index 98dc5087..0f95a9b9 100644 --- a/crates/dkg/examples/sync.rs +++ b/crates/dkg/examples/sync.rs @@ -61,7 +61,7 @@ use pluto_dkg::sync::{self, Client, Server}; use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode, - config::P2PConfig, + config::{P2PConfig, RelayAddr}, gater, k1, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -85,7 +85,7 @@ struct ExampleBehaviour { struct Args { /// Relay URLs or relay multiaddrs to use. #[arg(long, value_delimiter = ',')] - relays: Vec, + relays: Vec, /// Data directory containing `charon-enr-private-key` and /// `cluster-lock.json`, typically one of the `nodeN/` directories produced diff --git a/crates/dkg/src/dkg.rs b/crates/dkg/src/dkg.rs index 5e75daab..277f68f1 100644 --- a/crates/dkg/src/dkg.rs +++ b/crates/dkg/src/dkg.rs @@ -364,7 +364,7 @@ impl AppendConfig { fn default_p2p_config() -> P2PConfig { P2PConfig { - relays: pluto_p2p::config::default_relay_multiaddrs(), + relays: pluto_p2p::config::default_relays(), ..Default::default() } } @@ -1090,10 +1090,7 @@ mod tests { assert_eq!(config.def_file, DEFAULT_DEFINITION_FILE); assert!(!config.no_verify); assert_eq!(config.data_dir, path::PathBuf::from(DEFAULT_DATA_DIR)); - assert_eq!( - config.p2p.relays, - pluto_p2p::config::default_relay_multiaddrs() - ); + assert_eq!(config.p2p.relays, pluto_p2p::config::default_relays()); assert_eq!(config.log.override_env_filter.as_deref(), Some("info")); assert!(config.log.console.is_some()); assert_eq!(config.publish.address, DEFAULT_PUBLISH_ADDRESS); diff --git a/crates/dkg/src/node.rs b/crates/dkg/src/node.rs index f6d55b0e..ca20add8 100644 --- a/crates/dkg/src/node.rs +++ b/crates/dkg/src/node.rs @@ -60,8 +60,7 @@ pub(crate) async fn setup_p2p( verify_p2p_key(peers, &key)?; - let relay_addrs = bootnode::relay_addrs_for_resolution(&conf.p2p.relays); - let relays = bootnode::new_relays(ct, &relay_addrs, &hex::encode(&def_hash)).await?; + let relays = bootnode::new_relays(ct, &conf.p2p.relays, &hex::encode(&def_hash)).await?; let conn_gater = gater::ConnGater::new_conn_gater(peer_ids.clone(), relays.clone()); diff --git a/crates/p2p/Cargo.toml b/crates/p2p/Cargo.toml index 34ea6d69..1765c4a1 100644 --- a/crates/p2p/Cargo.toml +++ b/crates/p2p/Cargo.toml @@ -43,6 +43,7 @@ libp2p.workspace = true k256.workspace = true tokio = { workspace = true, features = ["test-util"] } futures.workspace = true +wiremock.workspace = true [lints] workspace = true diff --git a/crates/p2p/examples/bootnode.rs b/crates/p2p/examples/bootnode.rs index e84c36b8..5db64c8c 100644 --- a/crates/p2p/examples/bootnode.rs +++ b/crates/p2p/examples/bootnode.rs @@ -39,7 +39,7 @@ use pluto_cluster::lock::Lock; use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode, - config::P2PConfig, + config::{P2PConfig, RelayAddr}, gater, k1, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -62,7 +62,7 @@ pub struct ExampleBehaviour { pub struct Args { /// The relay URLs to use #[arg(long, value_delimiter = ',')] - relays: Vec, + relays: Vec, /// The data directory to use #[arg(long)] diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs index e07affab..ea2d2ec3 100644 --- a/crates/p2p/src/bootnode.rs +++ b/crates/p2p/src/bootnode.rs @@ -3,13 +3,15 @@ use std::time::Duration; use backon::Retryable; -use libp2p::{Multiaddr, multiaddr::Protocol}; +use libp2p::Multiaddr; use pluto_eth2util::enr::Record; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; +use url::Url; -use crate::peer::{ - AddrInfo, MutablePeer, Peer, PeerError, addr_infos_from_p2p_addrs, peer_id_from_key, +use crate::{ + config::RelayAddr, + peer::{AddrInfo, MutablePeer, Peer, PeerError, addr_infos_from_p2p_addrs, peer_id_from_key}, }; /// Polling interval for relay address updates. @@ -32,21 +34,21 @@ const RELAY_MAX_BODY: usize = 1024 * 1024; /// Bootnode error. #[derive(Debug, thiserror::Error)] pub enum BootnodeError { - /// Invalid relay multiaddr. - #[error("invalid relay multiaddr: {0}")] - InvalidRelayMultiaddr(String), - /// Failed to get peer from multiaddr. #[error("peer from multiaddr: {0}")] PeerFromMultiaddr(String), - /// Failed to parse relay URL. - #[error("parse relay url")] - ParseRelayUrl(#[from] url::ParseError), + /// The relay responded with a non-success status. + #[error("relay address query failed with status {0}")] + RelayQueryStatus(u16), + + /// The relay address response was not valid JSON. + #[error("parse relay addresses json: {0}")] + ParseRelayAddrsJson(#[source] serde_json::Error), - /// Invalid relay URL (not http/https). - #[error("invalid relay url")] - InvalidRelayUrl, + /// The relay URL scheme is neither `http` nor `https`. + #[error("invalid relay url: {0}")] + InvalidRelayUrl(String), /// HTTP request error. #[error("new request: {0}")] @@ -99,39 +101,45 @@ pub type Result = std::result::Result; /// Waits up to 1 minute for at least one ENR to resolve. pub async fn new_relays( cancel: CancellationToken, - relays: &[String], + relays: &[RelayAddr], lock_hash_hex: &str, ) -> Result> { let mut resp = Vec::new(); for relay_addr in relays { - if relay_addr.starts_with("http") { - if !relay_addr.starts_with("https") { - warn!(addr = %relay_addr, "Relay URL does not use https protocol"); - } - - let mutable = MutablePeer::default(); - let url = relay_addr.clone(); - let hash = lock_hash_hex.to_string(); - let mutable_clone = mutable.clone(); - let cancel_clone = cancel.child_token(); + match relay_addr { + RelayAddr::Url(url) => { + // Reject a scheme the resolver cannot use before spawning it. + // The task would fail on its first request and exit, leaving + // the wait below to run its full timeout and report a + // resolution timeout instead of the real problem. + if !matches!(url.scheme(), "http" | "https") { + return Err(BootnodeError::InvalidRelayUrl(url.to_string())); + } - tokio::spawn(async move { - resolve_relay(cancel_clone, url, hash, mutable_clone).await; - }); + if url.scheme() != "https" { + warn!(addr = %url, "Relay URL does not use https protocol"); + } - resp.push(mutable); - continue; - } + let mutable = MutablePeer::default(); + let url = url.clone(); + let hash = lock_hash_hex.to_string(); + let mutable_clone = mutable.clone(); + let cancel_clone = cancel.child_token(); - let addr: Multiaddr = relay_addr - .parse() - .map_err(|_| BootnodeError::InvalidRelayMultiaddr(relay_addr.clone()))?; + tokio::spawn(async move { + resolve_relay(cancel_clone, url, hash, mutable_clone).await; + }); - let info = addr_info_from_p2p_addr(&addr) - .map_err(|_| BootnodeError::PeerFromMultiaddr(relay_addr.clone()))?; + resp.push(mutable); + } + RelayAddr::Multiaddr(addr) => { + let info = addr_info_from_p2p_addr(addr) + .map_err(|_| BootnodeError::PeerFromMultiaddr(addr.to_string()))?; - resp.push(MutablePeer::new(Peer::new_relay_peer(&info))); + resp.push(MutablePeer::new(Peer::new_relay_peer(&info))); + } + } } if resp.is_empty() { @@ -165,7 +173,7 @@ pub async fn new_relays( /// Polls the URL every 2 minutes and calls the callback when peer info changes. async fn resolve_relay( cancel: CancellationToken, - raw_url: String, + relay_url: Url, lock_hash_hex: String, mutable: MutablePeer, ) { @@ -180,11 +188,12 @@ async fn resolve_relay( return; } - let addrs = match query_relay_addrs(cancel.clone(), &client, &raw_url, &lock_hash_hex).await + let addrs = match query_relay_addrs(cancel.clone(), &client, &relay_url, &lock_hash_hex) + .await { Ok(addrs) => addrs, Err(e) => { - tracing::error!(err = %e, url = %raw_url, "Failed resolving relay addresses from URL"); + tracing::error!(err = %e, url = %relay_url, "Failed resolving relay addresses from URL"); return; } }; @@ -208,7 +217,7 @@ async fn resolve_relay( let peer = Peer::new_relay_peer(&infos[0]); info!( peer = %peer.name, - url = %raw_url, + url = %relay_url, addrs = ?peer.addresses, "Resolved new relay" ); @@ -236,13 +245,15 @@ async fn resolve_relay( async fn query_relay_addrs( cancel: CancellationToken, client: &reqwest::Client, - relay_url: &str, + relay_url: &Url, lock_hash_hex: &str, ) -> Result> { - let parsed_url = url::Url::parse(relay_url)?; - let scheme = parsed_url.scheme(); - if scheme != "http" && scheme != "https" { - return Err(BootnodeError::InvalidRelayUrl); + // `RelayAddr::from_str` already enforces this, but the variant is publicly + // constructible, so re-check at the point of use — and before the retry + // loop, since an unsupported scheme is not transient and would otherwise + // retry until cancellation instead of failing fast. + if !matches!(relay_url.scheme(), "http" | "https") { + return Err(BootnodeError::InvalidRelayUrl(relay_url.to_string())); } // Retry with exponential backoff until the cancel token fires, matching @@ -255,7 +266,7 @@ async fn query_relay_addrs( } let resp = client - .get(relay_url) + .get(relay_url.clone()) .header("Charon-Cluster", lock_hash_hex) .send() .await @@ -269,7 +280,7 @@ async fn query_relay_addrs( status_code = resp.status().as_u16(), "Non-200 response querying relay addresses (will try again)" ); - return Err(BootnodeError::InvalidRelayUrl); + return Err(BootnodeError::RelayQueryStatus(resp.status().as_u16())); } let body = read_relay_body_capped(resp, RELAY_MAX_BODY).await?; @@ -286,7 +297,7 @@ async fn query_relay_addrs( let addrs: Vec = serde_json::from_str(&body).map_err(|e| { tracing::warn!(err = %e, "Failure parsing relay addresses json (will try again)"); - BootnodeError::InvalidRelayUrl + BootnodeError::ParseRelayAddrsJson(e) })?; let mut maddrs = Vec::new(); @@ -391,88 +402,148 @@ fn addr_info_from_p2p_addr(addr: &Multiaddr) -> std::result::Result Vec { - relays.iter().map(relay_addr_for_resolution).collect() -} +#[cfg(test)] +mod tests { + use k256::elliptic_curve::rand_core::OsRng; + use libp2p::PeerId; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{header, method, path}, + }; -/// Converts one relay multiaddr into the string form [`new_relays`] expects. -/// -/// The default port for the scheme (80/443) is omitted from the URL. -pub fn relay_addr_for_resolution(relay: &Multiaddr) -> String { - let mut scheme = None; - let mut host = None; - let mut port = None; - - for protocol in relay.iter() { - match protocol { - Protocol::Http => scheme = Some("http"), - Protocol::Https => scheme = Some("https"), - Protocol::Dns(name) - | Protocol::Dns4(name) - | Protocol::Dns6(name) - | Protocol::Dnsaddr(name) - if host.is_none() => - { - host = Some(name.to_string()); - } - Protocol::Ip4(ip) if host.is_none() => { - host = Some(ip.to_string()); - } - Protocol::Ip6(ip) if host.is_none() => { - host = Some(format!("[{ip}]")); - } - Protocol::Tcp(tcp_port) => port = Some(tcp_port), - _ => {} - } + use super::*; + + const LOCK_HASH: &str = "0badcafe"; + + /// Returns a random relay peer ID together with the multiaddr JSON body a + /// relay serves for it. + fn relay_fixture() -> (PeerId, String) { + let key = k256::SecretKey::random(&mut OsRng); + let peer_id = peer_id_from_key(key.public_key()).expect("peer id from key"); + let body = serde_json::to_string(&[format!("/ip4/10.0.0.1/tcp/3610/p2p/{peer_id}")]) + .expect("serialize relay addrs"); + + (peer_id, body) } - if let (Some(scheme), Some(host)) = (scheme, host) { - let default_port = match scheme { - "https" => 443, - _ => 80, - }; + /// Starts a relay stub serving `body` at `relay_path`. Any other path 404s, + /// so a request that loses the path fails the test rather than silently + /// resolving against the root. + async fn relay_server(relay_path: &str, body: String) -> MockServer { + let server = MockServer::start().await; - return match port { - Some(port) if port != default_port => format!("{scheme}://{host}:{port}"), - _ => format!("{scheme}://{host}"), - }; + Mock::given(method("GET")) + .and(path(relay_path)) + .and(header("Charon-Cluster", LOCK_HASH)) + .respond_with(ResponseTemplate::new(200).set_body_string(body)) + .mount(&server) + .await; + + server } - relay.to_string() -} + /// Resolves `relay` and returns the single relay peer ID it yields. + async fn resolve_one(relay: &str) -> PeerId { + let relay: RelayAddr = relay.parse().expect("relay addr should parse"); + let cancel = CancellationToken::new(); -#[cfg(test)] -mod tests { - use super::*; + let relays = new_relays(cancel.clone(), &[relay], LOCK_HASH) + .await + .expect("relays should resolve"); + cancel.cancel(); - #[test] - fn relay_addr_resolution_forms() { - let cases = [ - ( - "/dns/relay.example.org/tcp/443/https", - "https://relay.example.org", - ), - ( - "/dns/relay.example.org/tcp/8443/https", - "https://relay.example.org:8443", - ), - ( - "/dns4/relay.example.org/tcp/80/http", - "http://relay.example.org", + assert_eq!(relays.len(), 1); + + relays[0].peer().expect("relay should be resolved").id + } + + #[tokio::test] + async fn new_relays_resolves_url_with_path() { + let (peer_id, body) = relay_fixture(); + let server = relay_server("/enr", body).await; + + assert_eq!(resolve_one(&format!("{}/enr", server.uri())).await, peer_id); + } + + #[tokio::test] + async fn new_relays_resolves_url_without_path() { + let (peer_id, body) = relay_fixture(); + let server = relay_server("/", body).await; + + assert_eq!(resolve_one(&server.uri()).await, peer_id); + } + + #[tokio::test] + async fn new_relays_resolves_raw_multiaddr() { + let (peer_id, _) = relay_fixture(); + + assert_eq!( + resolve_one(&format!("/ip4/10.0.0.1/tcp/3610/p2p/{peer_id}")).await, + peer_id + ); + } + + #[tokio::test] + async fn new_relays_rejects_multiaddr_without_peer_id() { + let relay: RelayAddr = "/ip4/10.0.0.1/tcp/3610".parse().expect("relay addr"); + + let err = new_relays(CancellationToken::new(), &[relay], LOCK_HASH) + .await + .expect_err("multiaddr without a peer ID should be rejected"); + + assert!(matches!(err, BootnodeError::PeerFromMultiaddr(_))); + } + + #[tokio::test] + async fn new_relays_without_relays_is_empty() { + let relays = new_relays(CancellationToken::new(), &[], LOCK_HASH) + .await + .expect("no relays should resolve"); + + assert!(relays.is_empty()); + } + + #[tokio::test] + async fn new_relays_rejects_unsupported_url_scheme() { + // A hand-built `RelayAddr::Url` can carry a scheme the resolver cannot + // use. It must be rejected up front, not spawned and then waited on + // until the resolve timeout reports a misleading failure. + let relay = RelayAddr::Url("ftp://relay.example.org".parse().expect("url")); + + let err = tokio::time::timeout( + Duration::from_secs(5), + new_relays( + CancellationToken::new(), + std::slice::from_ref(&relay), + LOCK_HASH, ), - ("/ip4/10.0.0.1/tcp/3640/http", "http://10.0.0.1:3640"), - ("/ip6/::1/tcp/443/https", "https://[::1]"), - ]; - for (addr, expected) in cases { - let addr: Multiaddr = addr.parse().expect("valid multiaddr"); - assert_eq!(relay_addr_for_resolution(&addr), expected); - } + ) + .await + .expect("should fail fast, not wait out BOOTNODE_RESOLVE_TIMEOUT") + .expect_err("unsupported scheme should be rejected"); + + assert!( + matches!(err, BootnodeError::InvalidRelayUrl(_)), + "unexpected error: {err}" + ); + } - // Non-HTTP multiaddrs fall back to the multiaddr string. - let plain: Multiaddr = "/ip4/10.0.0.1/tcp/3610".parse().expect("valid multiaddr"); - assert_eq!(relay_addr_for_resolution(&plain), plain.to_string()); + #[tokio::test] + async fn query_relay_addrs_rejects_unsupported_scheme() { + // `RelayAddr::Url` is publicly constructible, so a non-HTTP(S) URL can + // reach here; it must fail fast rather than retry until cancellation. + let err = query_relay_addrs( + CancellationToken::new(), + &reqwest::Client::new(), + &"ftp://relay.example.org".parse().expect("url"), + LOCK_HASH, + ) + .await + .expect_err("unsupported scheme should not be queried"); + + assert!( + matches!(err, BootnodeError::InvalidRelayUrl(_)), + "unexpected error: {err}" + ); } } diff --git a/crates/p2p/src/config.rs b/crates/p2p/src/config.rs index fb31857b..2ebd72c1 100644 --- a/crates/p2p/src/config.rs +++ b/crates/p2p/src/config.rs @@ -1,12 +1,14 @@ //! # Charon P2P Configuration use std::{ + fmt, net::{IpAddr, SocketAddr}, str::FromStr, time::Duration, }; use libp2p::{Multiaddr, multiaddr, ping}; +use url::Url; /// Shared default relay endpoints used by commands and P2P-facing configs. pub const DEFAULT_RELAYS: [&str; 5] = [ @@ -17,6 +19,97 @@ pub const DEFAULT_RELAYS: [&str; 5] = [ "https://1.relay.obol.tech", ]; +/// Relay address parse error. +#[derive(Debug, thiserror::Error)] +pub enum RelayAddrError { + /// The address is empty. + #[error("empty relay address")] + Empty, + + /// The `http`-prefixed address is not a valid URL. + #[error("invalid relay url: {0}")] + Url(#[source] url::ParseError), + + /// The URL scheme is neither `http` nor `https`. + #[error("invalid relay url scheme {0:?}, want http or https")] + Scheme(String), + + /// The address is not a valid libp2p multiaddr. + #[error("invalid relay multiaddr: {0}")] + Multiaddr(#[source] multiaddr::Error), +} + +/// A configured libp2p relay address. +/// +/// A relay is given either as an HTTP(S) endpoint, whose ENR is resolved in the +/// background, or as a libp2p multiaddr that is dialed directly. Modelling that +/// split in the type keeps URL paths intact — a multiaddr cannot represent one, +/// so `http://relay:3640/enr` would otherwise be rejected outright or silently +/// truncated to `http://relay:3640`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelayAddr { + /// An HTTP(S) endpoint serving the relay's ENR or multiaddrs. + /// + /// [`FromStr`] guarantees an `http`/`https` scheme, but the variant is + /// publicly constructible, so consumers re-check it rather than relying on + /// the invariant. + Url(Url), + + /// A raw libp2p multiaddr, dialed directly. + Multiaddr(Multiaddr), +} + +impl RelayAddr { + /// Returns true for a plain-`http://` URL, i.e. one whose ENR is fetched + /// over an unencrypted connection. + /// + /// Always false for a [`RelayAddr::Multiaddr`]: only the HTTP resolution + /// step is at issue here, so a directly dialed relay is never reported as + /// insecure regardless of the transport it names. + pub fn is_insecure_url(&self) -> bool { + matches!(self, Self::Url(url) if url.scheme() != "https") + } +} + +impl FromStr for RelayAddr { + type Err = RelayAddrError; + + fn from_str(s: &str) -> std::result::Result { + // The multiaddr parser accepts "" as a zero-component `Multiaddr`, so + // reject it up front: an empty address must not masquerade as a + // dialable relay. + if s.is_empty() { + return Err(RelayAddrError::Empty); + } + + // Dispatch on the literal `http` prefix rather than probing both + // parsers, so classification here matches how the address is later + // consumed. + if s.starts_with("http") { + let url = Url::parse(s).map_err(RelayAddrError::Url)?; + + if !matches!(url.scheme(), "http" | "https") { + return Err(RelayAddrError::Scheme(url.scheme().to_owned())); + } + + return Ok(Self::Url(url)); + } + + s.parse() + .map(Self::Multiaddr) + .map_err(RelayAddrError::Multiaddr) + } +} + +impl fmt::Display for RelayAddr { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Url(url) => url.fmt(f), + Self::Multiaddr(addr) => addr.fmt(f), + } + } +} + /// P2P configuration error. #[derive(Debug, thiserror::Error)] pub enum P2PConfigError { @@ -61,7 +154,7 @@ type Result = std::result::Result; #[derive(Debug, Clone, Default)] pub struct P2PConfig { /// Defines the libp2p relay multiaddrs or URLs. - pub relays: Vec, + pub relays: Vec, /// The external IP address of the node. pub external_ip: Option, @@ -110,11 +203,11 @@ impl P2PConfig { } } -/// Returns the default relay endpoints parsed as [`Multiaddr`]s. -pub fn default_relay_multiaddrs() -> Vec { +/// Returns the default relay endpoints parsed as [`RelayAddr`]s. +pub fn default_relays() -> Vec { DEFAULT_RELAYS .iter() - .map(|relay| multiaddr::from_url(relay).expect("default relay should parse")) + .map(|relay| relay.parse().expect("default relay should parse")) .collect() } @@ -133,7 +226,7 @@ impl P2PConfigBuilder { } /// Sets the relay multiaddrs. - pub fn with_relays(mut self, relays: Vec) -> Self { + pub fn with_relays(mut self, relays: Vec) -> Self { self.config.relays = relays; self } @@ -315,6 +408,106 @@ mod tests { assert_eq!(merged_addrs_str, expected_addrs_str); } + #[test] + fn relay_addr_parses_url_and_multiaddr_forms() { + // A path (and query) must survive parsing: `http://relay:3640/enr` is a + // supported relay address, and no multiaddr can express it. + let cases = [ + "http://relay:3640/enr", + "https://relay.example.org/enr", + "https://relay.example.org/enr?cluster=abc", + "/ip4/10.0.0.1/tcp/3610/p2p/16Uiu2HAm7ULrTMdiEmQCJ2N9nsuGvfUDvfDGgHXJ4vNjrCwCzGDs", + "/dns/relay.example.org/tcp/443/p2p/16Uiu2HAm7ULrTMdiEmQCJ2N9nsuGvfUDvfDGgHXJ4vNjrCwCzGDs", + ]; + for case in cases { + let addr: RelayAddr = case.parse().expect("relay addr should parse"); + assert_eq!(addr.to_string(), case, "{case} should round-trip"); + } + + // A host-only URL round-trips with the root path the `url` crate + // normalises it to; the request it produces is identical. + let addr: RelayAddr = "http://relay:3640" + .parse() + .expect("relay addr should parse"); + assert_eq!(addr.to_string(), "http://relay:3640/"); + } + + #[test] + fn relay_addr_flags_insecure_urls() { + let insecure: RelayAddr = "http://relay:3640/enr".parse().expect("relay addr"); + assert!(insecure.is_insecure_url()); + + let secure: RelayAddr = "https://relay:3640/enr".parse().expect("relay addr"); + assert!(!secure.is_insecure_url()); + + let multiaddr: RelayAddr = "/ip4/10.0.0.1/tcp/3610".parse().expect("relay addr"); + assert!(!multiaddr.is_insecure_url()); + } + + #[test] + fn relay_addr_rejects_invalid_forms() { + // `http`-prefixed but not a URL: the prefix dispatch classifies it as a + // URL, so it is rejected as one. + assert!(matches!( + "httpfoo".parse::(), + Err(RelayAddrError::Url(_)) + )); + assert!(matches!( + "https://".parse::(), + Err(RelayAddrError::Url(_)) + )); + + // `http`-prefixed and a valid URL, but not an HTTP(S) one. + assert!(matches!( + "httpx://relay.example.org".parse::(), + Err(RelayAddrError::Scheme(scheme)) if scheme == "httpx" + )); + + // Everything else must be a multiaddr — including other URL schemes, + // which never reach the scheme check. + assert!(matches!( + "ftp://relay.example.org".parse::(), + Err(RelayAddrError::Multiaddr(_)) + )); + assert!(matches!( + "not-an-address".parse::(), + Err(RelayAddrError::Multiaddr(_)) + )); + + // The multiaddr parser accepts "" as a zero-component multiaddr, so an + // empty address needs its own guard. + assert!(matches!( + "".parse::(), + Err(RelayAddrError::Empty) + )); + assert!("".parse::().is_ok(), "guard is still needed"); + } + + #[test] + fn relay_addr_error_exposes_its_cause() { + let err = "not-an-address" + .parse::() + .expect_err("should not parse"); + + // The message is self-contained, and the typed cause stays reachable + // through the error chain. + assert!(err.to_string().starts_with("invalid relay multiaddr:")); + assert!( + std::error::Error::source(&err) + .expect("source") + .downcast_ref::() + .is_some() + ); + } + + #[test] + fn default_relays_parse() { + let relays = default_relays(); + + assert_eq!(relays.len(), DEFAULT_RELAYS.len()); + assert!(relays.iter().all(|relay| !relay.is_insecure_url())); + } + #[test] fn config_invalid_multiaddrs() { let config = P2PConfig { diff --git a/crates/parsigex/examples/parsigex.rs b/crates/parsigex/examples/parsigex.rs index 06663a60..0d6ff913 100644 --- a/crates/parsigex/examples/parsigex.rs +++ b/crates/parsigex/examples/parsigex.rs @@ -75,7 +75,7 @@ use pluto_core::{ use pluto_p2p::{ behaviours::pluto::PlutoBehaviourEvent, bootnode, - config::P2PConfig, + config::{P2PConfig, RelayAddr}, gater, k1, p2p::{Node, NodeType}, p2p_context::P2PContext, @@ -133,7 +133,7 @@ impl From for CombinedBehaviourEvent { struct Args { /// Relay URLs or multiaddrs. #[arg(long, value_delimiter = ',')] - relays: Vec, + relays: Vec, /// Directory holding the p2p private key and cluster lock. #[arg(long)]