diff options
| author | tv <tv@krebsco.de> | 2026-10-28 17:40:07 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-28 17:40:07 +0100 |
| commit | 5c79174b36f7ce077b3d1b9feafe88008696fc31 (patch) | |
| tree | f0b23629719d6a0f458f1b4d3aa5e675a8f9fa55 | |
| parent | d7959009790a422ebb17d4c4ab1d9be3d48dc511 (diff) | |
introduce relay only mode
| -rw-r--r-- | src/api/mod.rs | 23 | ||||
| -rw-r--r-- | src/api/stats.rs | 46 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 19 | ||||
| -rw-r--r-- | src/config/mod.rs | 104 | ||||
| -rw-r--r-- | src/main.rs | 139 | ||||
| -rw-r--r-- | src/simtest/identity.rs | 1 | ||||
| -rw-r--r-- | src/transport/libp2p.rs | 70 |
7 files changed, 343 insertions, 59 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index 45eaec4..2cb0415 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -76,6 +76,29 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> .with_state(state) } +/// The HTTP router of a node that only relays (`--relay-only`): its figures and the log +/// filter (authenticated), and `/health`. +#[cfg_attr(not(feature = "libp2p"), allow(dead_code))] +pub(crate) fn relay_only_router( + transport: std::sync::Arc<dyn crate::transport::Transport>, + auth_state: AuthState, +) -> Router { + let relay = stats::RelayOnly { + transport, + started: tokio::time::Instant::now(), + }; + Router::new() + .route( + "/admin/log", + get(log::get_log_filter).put(log::put_log_filter), + ) + .route("/admin/stats", get(stats::get_relay_only_stats)) + .layer(middleware::from_fn_with_state(auth_state, basic_auth)) + .route("/health", get(health::health)) + .layer(middleware::from_fn(json_errors)) + .with_state(relay) +} + /// An error answer: `{"error": message}` with `details`' fields added (`docs/openapi.yaml`, /// `Error`). pub(crate) fn error_response( diff --git a/src/api/stats.rs b/src/api/stats.rs index 07521ab..889b675 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -189,6 +189,52 @@ struct StabilityStats { observed_secs: u64, } +/// What `/admin/stats` of a node that only relays (`--relay-only`) says. +#[derive(Clone)] +pub(crate) struct RelayOnly { + pub(crate) transport: std::sync::Arc<dyn crate::transport::Transport>, + pub(crate) started: tokio::time::Instant, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RelayOnlyStats { + /// The relay's libp2p peer ID. + peer_id: String, + /// Where nodes reach it, as addresses to list (`_continuum/relays`). + addresses: Vec<String>, + uptime_secs: u64, + transports: TransportStats, +} + +/// `GET /admin/stats` of a node that only relays. +#[cfg_attr(not(feature = "libp2p"), allow(dead_code))] +pub(crate) async fn get_relay_only_stats( + State(relay): State<RelayOnly>, + headers: HeaderMap, +) -> Response { + let stats = RelayOnlyStats { + peer_id: relay.transport.local_id().to_owned(), + addresses: relay + .transport + .local_addresses() + .into_iter() + .map(|address| format!("libp2p:{address}")) + .collect(), + uptime_secs: relay.started.elapsed().as_secs(), + transports: relay.transport.stats(), + }; + let accept = headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()); + if prometheus::wanted(accept) { + let body = prometheus::render_relay_only(&stats); + ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response() + } else { + Json(stats).into_response() + } +} + /// `GET /admin/stats`. pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap) -> Response { let stats = collect(&state).await; diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 8b37864..690effa 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -363,6 +363,25 @@ fn transports(out: &mut Out, stats: &TransportStats) { } } +/// `/admin/stats` of a node that only relays. +pub(super) fn render_relay_only(stats: &super::RelayOnlyStats) -> String { + let mut out = Out::default(); + out.family( + "relay_info", + "gauge", + "This relay's libp2p peer ID (always 1); it is no node.", + ); + out.sample("relay_info", &[("peer_id", &stats.peer_id)], 1.0); + out.single( + "uptime_seconds", + "gauge", + "Seconds since this relay started.", + stats.uptime_secs as f64, + ); + transports(&mut out, &stats.transports); + out.0 +} + /// A per-peer incident counter: its name, its help and how to read it. type IncidentSeries = (&'static str, &'static str, fn(&Incidents) -> f64); diff --git a/src/config/mod.rs b/src/config/mod.rs index 0890d3b..9210922 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -100,7 +100,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::Text, "PATH", None, - "Users of the HTTP API (htpasswd file with Argon2 hashes). Required.", + "Users of the HTTP API (htpasswd file with Argon2 hashes). Required, but with relay-only (then without it, no HTTP API).", ), option( "protocol-writers", @@ -191,7 +191,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::Text, "PATH", None, - "This node's key (Ed25519, 32 raw bytes): who the node is, whatever transport reaches it. Keep it with the node: a new key is a new node. Required.", + "This node's key (Ed25519, 32 raw bytes): who the node is, whatever transport reaches it. Keep it with the node: a new key is a new node. Required, but with relay-only (then not allowed).", ), option( "create-node-key", @@ -306,6 +306,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "With the tor transport: the port of this node's onion service (published with its onion address).", ), option( + "relay-only", + Kind::Flag, + "BOOL", + Some("false"), + "Only relay for nodes of swarms (libp2p circuit relay), without being a node: no document, no swarm, no node key. Needs transports = libp2p, libp2p-listen and libp2p-key-file; options about a swarm or a document are errors with it. List it for a swarm in the document (_continuum/relays), or configure it (libp2p-relays). The HTTP API (with htpasswd) has /health, /admin/stats and /admin/log only.", + ), + option( "libp2p-key-file", Kind::Text, "PATH", @@ -377,7 +384,8 @@ pub(crate) fn spec(name: &str) -> Option<&'static OptionSpec> { pub(crate) struct Settings { #[serde(default = "default_bind_addr")] pub(crate) bind_addr: SocketAddr, - pub(crate) htpasswd: String, + /// Required (`check`), but with `relay_only`. + pub(crate) htpasswd: Option<String>, #[serde(default)] pub(crate) protocol_writers: Vec<String>, #[serde(default)] @@ -404,7 +412,8 @@ pub(crate) struct Settings { pub(crate) remake: Option<String>, #[serde(default)] pub(crate) force: Vec<String>, - pub(crate) node_key_file: PathBuf, + /// Required (`check`), but not allowed with `relay_only`. + pub(crate) node_key_file: Option<PathBuf>, #[serde(default)] pub(crate) create_node_key: bool, pub(crate) swarm_key_file: Option<PathBuf>, @@ -448,6 +457,8 @@ pub(crate) struct Settings { #[serde(default = "default_tor_port")] #[cfg_attr(not(feature = "tor"), allow(dead_code))] pub(crate) tor_port: u16, + #[serde(default)] + pub(crate) relay_only: bool, pub(crate) mycelium_rest_url: Option<String>, pub(crate) mycelium_key_file: Option<PathBuf>, pub(crate) mycelium_network_key_file: Option<PathBuf>, @@ -552,6 +563,18 @@ impl Settings { if self.genesis && self.recover_from.is_some() { bail!("recover-from starts the swarm from a backup; it can't be combined with genesis"); } + if self.relay_only { + self.check_relay_only()?; + } else { + if self.htpasswd.is_none() { + bail!("htpasswd is required: the users of the HTTP API"); + } + if self.node_key_file.is_none() { + bail!( + "node-key-file is required: who this node is (create one with create-node-key)" + ); + } + } if self.transports.is_empty() { bail!("transports: a node needs at least one"); } @@ -604,6 +627,53 @@ impl Settings { } } +impl Settings { + /// A relay that is no node: libp2p alone, dialable, with a key kept, and nothing + /// about a swarm or a document. + fn check_relay_only(&self) -> Result<()> { + if !cfg!(feature = "libp2p") { + bail!("relay-only relays over libp2p; this build has no libp2p"); + } + if self.transports != ["libp2p"] { + bail!("relay-only relays over libp2p alone: set transports to libp2p"); + } + if self.libp2p_listen.is_empty() { + bail!("relay-only needs libp2p-listen: where nodes reach the relay"); + } + if self.libp2p_key_file.is_none() { + bail!( + "relay-only needs libp2p-key-file: without it, the relay's peer ID changes with every start and nodes lose it" + ); + } + let swarm_options = [ + ("node-key-file", self.node_key_file.is_some()), + ("create-node-key", self.create_node_key), + ("seeds", !self.seeds.is_empty()), + ("genesis", self.genesis), + ("state-file", self.state_file.is_some()), + ("document-file", self.document_file.is_some()), + ("recover-from", self.recover_from.is_some()), + ("remake", self.remake.is_some()), + ("swarm-key-file", self.swarm_key_file.is_some()), + ("protocol-writers", !self.protocol_writers.is_empty()), + ("alias", self.alias.is_some()), + ("debug-api", self.debug_api), + ]; + let set: Vec<&str> = swarm_options + .iter() + .filter(|(_, set)| *set) + .map(|(name, _)| *name) + .collect(); + if !set.is_empty() { + bail!( + "relay-only is no node of a swarm, so these don't apply: {}", + set.join(", ") + ); + } + Ok(()) + } +} + /// Up to 64 characters, no control characters, not blank. fn check_alias(alias: &str) -> Result<()> { if alias.trim().is_empty() { @@ -949,8 +1019,6 @@ fn culprit(merged: &Layer) -> Option<(&String, &Entry)> { return false; } let mut alone = serde_json::Map::new(); - alone.insert("htpasswd".to_owned(), JsonValue::String(String::new())); - alone.insert("node-key-file".to_owned(), JsonValue::String(String::new())); alone.insert((*name).clone(), entry.value.clone()); serde_json::from_value::<Settings>(JsonValue::Object(alone)).is_err() }) @@ -1506,6 +1574,30 @@ mod tests { ) }; assert!(transports(json!([])).is_err()); + // Without the node key: an error, but for a relay. + assert!(resolve(layer(json!({ "htpasswd": "u" })), Vec::new()).is_err()); + let relay = |extra: JsonValue| { + let mut settings = json!({ + "relay-only": true, + "transports": ["libp2p"], + "libp2p-listen": ["/ip6/::/tcp/9661"], + "libp2p-key-file": "libp2p.key", + }); + settings + .as_object_mut() + .unwrap() + .extend(extra.as_object().unwrap().clone()); + resolve(layer(settings), Vec::new()) + }; + assert_eq!(relay(json!({})).is_ok(), cfg!(feature = "libp2p")); + assert_eq!( + relay(json!({ "htpasswd": "u" })).is_ok(), + cfg!(feature = "libp2p") + ); + assert!(relay(json!({ "node-key-file": "node.key" })).is_err()); + assert!(relay(json!({ "seeds": ["a"] })).is_err()); + assert!(relay(json!({ "transports": ["mycelium"] })).is_err()); + assert!(relay(json!({ "libp2p-listen": [] })).is_err()); assert!(transports(json!(["pigeon"])).is_err()); assert!(transports(json!(["mycelium", "mycelium"])).is_err()); assert!(transports(json!(["mycelium"])).is_ok()); diff --git a/src/main.rs b/src/main.rs index e6a27f4..f16138e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -52,10 +52,17 @@ async fn main() -> Result<()> { return Ok(()); } let args = &config.settings; + if args.relay_only { + return relay_only(args).await; + } // Read before anything starts, so a bad key file fails right away. + let node_key_file = args + .node_key_file + .as_deref() + .context("node-key-file is required")?; let node_key = Arc::new(identity::NodeKey::load( - &args.node_key_file, + node_key_file, args.create_node_key, )?); info!(node_id = %node_key.id(), "Node identity"); @@ -71,7 +78,7 @@ async fn main() -> Result<()> { // the spot, and startup (spawning Mycelium, bootstrap) can take minutes. let stop = watch_for_shutdown().context("Installing signal handlers")?; // A broken htpasswd will abort startup. - let auth_state = AuthState::new(&args.htpasswd)?; + let auth_state = AuthState::new(args.htpasswd.as_deref().context("htpasswd is required")?)?; let uses = |name: &str| args.transports.iter().any(|transport| transport == name); let mut mycelium_setup = if uses("mycelium") { @@ -183,52 +190,12 @@ async fn main() -> Result<()> { ) } #[cfg(feature = "libp2p")] - "libp2p" => { - let listen = args - .libp2p_listen - .iter() - .map(|address| { - address - .parse::<libp2p::Multiaddr>() - .with_context(|| format!("libp2p-listen {address:?}")) - }) - .collect::<Result<Vec<_>>>()?; - let announce = args - .libp2p_announce - .iter() - .map(|address| { - address - .parse::<libp2p::Multiaddr>() - .with_context(|| format!("libp2p-announce {address:?}")) - }) - .collect::<Result<Vec<_>>>()?; - let relays = args - .libp2p_relays - .iter() - .map(|address| { - address - .parse::<libp2p::Multiaddr>() - .with_context(|| format!("libp2p-relays {address:?}")) - }) - .collect::<Result<Vec<_>>>()?; - let key = - transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?; - ( - transport::libp2p::SCHEME, - Arc::new(transport::libp2p::Libp2pTransport::start( - transport::libp2p::Libp2pConfig { - key, - listen, - announce, - relays, - relay: args.libp2p_relay, - mdns: args.libp2p_mdns, - public: args.libp2p_public, - memory: false, - }, - )?), - ) - } + "libp2p" => ( + transport::libp2p::SCHEME, + Arc::new(transport::libp2p::Libp2pTransport::start(libp2p_config( + args, false, + )?)?), + ), #[cfg(feature = "tor")] "tor" => { let state_dir = args @@ -397,6 +364,82 @@ async fn main() -> Result<()> { Ok(()) } +/// The libp2p transport's configuration; `relay_only`: a relay that is no node. +#[cfg(feature = "libp2p")] +fn libp2p_config( + args: &config::Settings, + relay_only: bool, +) -> Result<transport::libp2p::Libp2pConfig> { + let parse = |option: &str, addresses: &[String]| { + addresses + .iter() + .map(|address| { + address + .parse::<libp2p::Multiaddr>() + .with_context(|| format!("{option} {address:?}")) + }) + .collect::<Result<Vec<_>>>() + }; + Ok(transport::libp2p::Libp2pConfig { + key: transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?, + listen: parse("libp2p-listen", &args.libp2p_listen)?, + announce: parse("libp2p-announce", &args.libp2p_announce)?, + relays: parse("libp2p-relays", &args.libp2p_relays)?, + relay: args.libp2p_relay || relay_only, + continuum: !relay_only, + mdns: args.libp2p_mdns, + public: args.libp2p_public, + memory: false, + }) +} + +/// `--relay-only`: libp2p relaying for nodes of swarms, and the HTTP API's relay part +/// (with `--htpasswd`), until stopped. +#[cfg(feature = "libp2p")] +async fn relay_only(args: &config::Settings) -> Result<()> { + let stop = watch_for_shutdown().context("Installing signal handlers")?; + let auth_state = args.htpasswd.as_deref().map(AuthState::new).transpose()?; + let transport: Arc<dyn Transport> = Arc::new(transport::libp2p::Libp2pTransport::start( + libp2p_config(args, true)?, + )?); + info!(peer_id = transport.local_id(), "Relaying only (no node)"); + + let server = match auth_state { + Some(auth_state) => { + let app = api::relay_only_router(Arc::clone(&transport), auth_state); + let listener = tokio::net::TcpListener::bind(args.bind_addr) + .await + .with_context(|| format!("Binding local API to {}", args.bind_addr))?; + info!(address = %listener.local_addr()?, "HTTP API listening"); + let stop = stop.clone(); + Some(tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(stopped(stop)) + .await + })) + } + None => { + info!("No htpasswd: no HTTP API"); + None + } + }; + + tokio::select! { + exit = transport.wait_for_exit() => error!(status = ?exit, "The libp2p transport stopped"), + _ = stopped(stop.clone()) => {} + } + if let Some(server) = server { + let _ = server.await; + } + info!("Relay stopped"); + Ok(()) +} + +#[cfg(not(feature = "libp2p"))] +async fn relay_only(_args: &config::Settings) -> Result<()> { + anyhow::bail!("relay-only relays over libp2p; this build has no libp2p") +} + /// Turns SIGINT and SIGTERM into a shutdown request. A second signal exits at once (the /// supervised Mycelium still gets its parent-death signal then). fn watch_for_shutdown() -> Result<watch::Receiver<bool>> { diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index 789b011..9f2bff3 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -219,6 +219,7 @@ async fn nodes_over_libp2p() { announce: Vec::new(), relays: Vec::new(), relay: false, + continuum: true, mdns: false, public: false, memory: true, diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index a6778c6..7351b16 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -146,6 +146,9 @@ pub(crate) struct Libp2pConfig { pub(crate) relays: Vec<Multiaddr>, /// Relay for other nodes. pub(crate) relay: bool, + /// Speak continuum (`PROTOCOL`): not on a node that only relays (`--relay-only`), + /// which nobody should take for a node. + pub(crate) continuum: bool, /// Find peers on the LAN (mDNS). pub(crate) mdns: bool, /// Use public infrastructure: the IPFS DHT, public relays. @@ -716,9 +719,15 @@ impl Libp2pTransport { swarm.add_external_address(address.clone()); } let mut control = swarm.behaviour().stream.new_control(); - let incoming = control - .accept(PROTOCOL) - .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?; + let incoming = if config.continuum { + Some( + control + .accept(PROTOCOL) + .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?, + ) + } else { + None + }; let (commands, receiver) = mpsc::unbounded_channel(); let mut book = Book { announced: config.announce, @@ -751,7 +760,7 @@ impl Libp2pTransport { control, commands, book, - incoming: tokio::sync::Mutex::new(Some(incoming)), + incoming: tokio::sync::Mutex::new(incoming), stopped, }) } @@ -1694,7 +1703,7 @@ impl Transport for Libp2pTransport { /// time, in the order they were read; PINGs and PONGs at once (`is_probe`). async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { let Some(mut incoming) = self.incoming.lock().await.take() else { - bail!("libp2p consumer already started"); + bail!("libp2p consumer already started, or this node only relays"); }; let handler: Arc<HandlerFn> = Arc::from(handler); let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS); @@ -1781,6 +1790,7 @@ mod tests { announce: Vec::new(), relays, relay, + continuum: true, mdns: false, public: false, memory: true, @@ -1926,6 +1936,7 @@ mod tests { announce: vec![announced.clone()], relays: Vec::new(), relay: false, + continuum: true, mdns: false, public: false, memory: true, @@ -1988,6 +1999,55 @@ mod tests { assert!(a.connection(b.local_id()).is_some()); } + /// A relay that is no node (`--relay-only`) relays, and takes no messages itself. + #[tokio::test] + async fn a_relay_that_is_no_node() { + let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed); + let key = Keypair::generate_ed25519(); + let r_peer = key.public().to_peer_id(); + let r_listen: Multiaddr = format!("/memory/{port}").parse().unwrap(); + let r = Libp2pTransport::start(Libp2pConfig { + key, + listen: vec![r_listen.clone()], + announce: Vec::new(), + relays: Vec::new(), + relay: true, + continuum: false, + mdns: false, + public: false, + memory: true, + }) + .unwrap(); + let r_address = r_listen.with_p2p(r_peer).unwrap(); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let (a, _) = memory_node(); + let relayed = b.local_addresses()[0].clone(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &status()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + + // The relay itself doesn't speak continuum: it takes no messages (a send to it + // isn't refused, though: the protocol goes with the first data, `swarm_config`, + // so the relay's refusal comes too late; nodes never send to it, as it never + // proves itself a node). + assert_eq!( + a.resolve_seed(&r_address.to_string()).await.unwrap(), + r.local_id() + ); + assert!( + r.start_consumer(Box::new(|_, _, _| Box::pin(async { Ok(()) }))) + .await + .is_err() + ); + } + /// A relay configured without its peer ID is found out, and used. #[tokio::test] async fn a_relay_configured_without_its_peer_id() { |
