From 2646e19cf29b410f542097d75e620116c96da96b Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 04:43:09 +0100 Subject: cleanup mycelium daemon --- src/main.rs | 3 +- src/transport/daemon.rs | 371 -------------------------- src/transport/mod.rs | 1 - src/transport/mycelium.rs | 548 -------------------------------------- src/transport/mycelium/daemon.rs | 337 ++++++++++++++++++++++++ src/transport/mycelium/mod.rs | 552 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 890 insertions(+), 922 deletions(-) delete mode 100644 src/transport/daemon.rs delete mode 100644 src/transport/mycelium.rs create mode 100644 src/transport/mycelium/daemon.rs create mode 100644 src/transport/mycelium/mod.rs diff --git a/src/main.rs b/src/main.rs index ece72ac..a79c6a2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -35,8 +35,7 @@ use api::auth::AuthState; use document::replica::Epoch; use swarm::bootstrap; use transport::Transport; -use transport::daemon::{MyceliumSetup, MyceliumSpawnConfig}; -use transport::mycelium; +use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig}; #[tokio::main] async fn main() -> Result<()> { diff --git a/src/transport/daemon.rs b/src/transport/daemon.rs deleted file mode 100644 index eceeb4d..0000000 --- a/src/transport/daemon.rs +++ /dev/null @@ -1,371 +0,0 @@ -//! Running Mycelium for this node: either a daemon the operator runs (its REST URL), or -//! one we start and supervise (`SupervisedMycelium`): its keys handed over in memory, -//! never on disk, and it dies with us. - -use anyhow::{Context, Result, anyhow, bail}; -use std::fs::File; -use std::io::Write; -use std::net::TcpListener; -use std::os::unix::io::FromRawFd; -use std::path::PathBuf; -use std::process::Stdio; -use tokio::process::{Child, Command}; -use url::Url; - -/// Represents where a 32-byte cryptographic key originates. -#[derive(Debug, Clone)] -pub enum KeySource { - Raw([u8; 32]), - Path(PathBuf), -} - -/// Atomic structure ensuring private network credentials are always handled together. -#[derive(Debug, Clone)] -pub struct PrivateNetworkConfig { - pub name: String, - pub key: KeySource, -} - -/// Comprehensive parameters required to spawn or target an instance. -#[derive(Debug, Clone)] -pub struct MyceliumSpawnConfig { - pub rest_url_opt: Option, - pub key_file_opt: Option, - pub network_key_file: Option, - pub network_name: String, - pub tcp_port: Option, - pub api_port: Option, - pub rpc_port: Option, - pub discovery_port: Option, - pub peers: Vec, -} - -impl Default for MyceliumSpawnConfig { - fn default() -> Self { - Self { - rest_url_opt: None, - key_file_opt: None, - network_key_file: None, - network_name: "something".to_string(), - tcp_port: None, - api_port: None, - rpc_port: None, - discovery_port: None, - peers: vec!["tls://127.0.0.1:9651".to_string()], - } - } -} - -pub enum MyceliumRuntime { - Supervised(Box), - External, -} - -pub struct MyceliumSetup { - pub runtime: MyceliumRuntime, - pub rest_url: Url, -} - -impl MyceliumSetup { - /// Evaluates the configuration payload and initializes the chosen execution strategy. - pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result { - // TODO warn for all ignored options if rest_url_opt is set - if let Some(url) = config.rest_url_opt { - tracing::info!( - "Configured to interface with an external standalone Mycelium instance at: {}", - url - ); - return Ok(Self { - runtime: MyceliumRuntime::External, - rest_url: url, - }); - } - - tracing::info!( - "Configured to initialize an embedded user-space Mycelium background worker" - ); - - let key = match config.key_file_opt { - Some(key_file) => KeySource::Path(key_file), - None => { - let mut bytes = [0u8; 32]; - getrandom::fill(&mut bytes)?; - KeySource::Raw(bytes) - } - }; - - let private_net_config = config - .network_key_file - .map(|net_key_path| PrivateNetworkConfig { - name: config.network_name.clone(), - key: KeySource::Path(net_key_path), - }); - - let supervised_node = SupervisedMycelium::spawn( - key, - private_net_config, - config.tcp_port, - config.api_port, - config.rpc_port, - config.discovery_port, - &config.peers, - ) - .await - .context("Failed to securely bootstrap embedded Mycelium background worker")?; - - let resolved_url = Url::parse(&format!( - "http://127.0.0.1:{}", - supervised_node.config.api_port - ))?; - - Ok(Self { - runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)), - rest_url: resolved_url, - }) - } - - /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely. - pub async fn wait_for_exit(&mut self) -> std::io::Result { - match &mut self.runtime { - MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, - MyceliumRuntime::External => { - std::future::pending::>().await - } - } - } -} - -#[derive(Debug, Clone)] -pub struct MyceliumNodeConfig { - pub tcp_port: u16, - pub api_port: u16, - pub rpc_port: u16, - pub discovery_port: Option, - pub private_network: Option, -} - -pub struct SupervisedMycelium { - child: Child, - pub config: MyceliumNodeConfig, - _key_file_handle: Option, - _net_key_file_handle: Option, -} - -impl SupervisedMycelium { - async fn spawn( - node_key: KeySource, - private_network: Option, - explicit_tcp: Option, - explicit_api: Option, - explicit_rpc: Option, - explicit_discovery_port: Option, - peers: &[String], - ) -> Result { - // Linux only: keys are handed over in memory (memfd, never on disk), and the - // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself. - if cfg!(not(target_os = "linux")) { - bail!( - "Supervising Mycelium is only supported on Linux; run myceliumd yourself \ - and pass --mycelium-rest-url" - ); - } - - let tcp_port = match explicit_tcp { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate TCP wire port")?, - }; - let api_port = match explicit_api { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate API port")?, - }; - let rpc_port = match explicit_rpc { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?, - }; - let discovery_port = match explicit_discovery_port { - Some(0) => Some( - Self::find_available_port().context("Failed to allocate peer discovery port")?, - ), - p => p, - }; - - let config = MyceliumNodeConfig { - tcp_port, - api_port, - rpc_port, - discovery_port, - private_network: private_network.as_ref().map(|pn| pn.name.clone()), - }; - - let parent_pid = std::process::id(); - - let (key_file_handle, key_file_path, node_fd) = - Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; - - let (net_key_file_handle, net_key_file_path, net_fd) = match private_network { - Some(net_cfg) => { - let (h, p, fd) = - Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; - (h, Some(p), fd) - } - None => (None, None, None), - }; - - let api_addr = format!("127.0.0.1:{}", config.api_port); - let rpc_addr = format!("127.0.0.1:{}", config.rpc_port); - - let mut cmd = Command::new("mycelium"); - - let fds_to_inherit: Vec = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); - // The child must never outlive us. An orphaned Mycelium keeps running with our - // key; the next one we start then shares its address, and replies (including - // Mycelium's own message ACKs) go to whichever instance the routers picked, - // leaving our messages pending forever. Drop only runs on a clean shutdown, so - // let the kernel signal the child when we die, however we die. - let parent_pid = std::process::id() as libc::pid_t; - unsafe { - cmd.pre_exec(move || { - for &fd in &fds_to_inherit { - libc::fcntl(fd, libc::F_SETFD, 0); - } - // Own process group: a Ctrl-C in the terminal goes to continuum only, - // which then stops Mycelium itself, instead of both exiting at once (which - // shows up as "Mycelium runtime process exited"). - if libc::setpgid(0, 0) != 0 { - return Err(std::io::Error::last_os_error()); - } - #[cfg(target_os = "linux")] - { - if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { - return Err(std::io::Error::last_os_error()); - } - // We may have died between fork and prctl. - if libc::getppid() != parent_pid { - libc::_exit(1); - } - } - Ok(()) - }); - } - - cmd.arg("--key-file") - .arg(&key_file_path) - .arg("--tcp-listen-port") - .arg(config.tcp_port.to_string()) - .arg("--disable-quic") - .arg("--api-addr") - .arg(&api_addr) - .arg("--jsonrpc-addr") - .arg(&rpc_addr) - .arg("--no-tun"); - - if let Some(ref discovery_port) = config.discovery_port { - cmd.arg("--peer-discovery-port") - .arg(discovery_port.to_string()); - } else { - cmd.arg("--disable-peer-discovery"); - } - - if let Some(ref net_name) = config.private_network { - cmd.arg("--network-name").arg(net_name); - } - if let Some(ref net_key_path) = net_key_file_path { - cmd.arg("--network-key-file").arg(net_key_path); - } - - if !peers.is_empty() { - cmd.arg("--peers").arg(peers.join(" ")); - } - - cmd.env("RUST_LOG", "info"); - - cmd.stdout(Stdio::piped()); - cmd.stderr(Stdio::piped()); - - let mut child = cmd - .spawn() - .context("Failed to execute mycelium process daemon")?; - - // 1. FIXED: Pass-through streaming for child process STDOUT - if let Some(mut stdout) = child.stdout.take() { - tokio::spawn(async move { - let mut host_stdout = tokio::io::stdout(); - let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await; - }); - } - - // 2. FIXED: Pass-through streaming for child process STDERR - if let Some(mut stderr) = child.stderr.take() { - tokio::spawn(async move { - let mut host_stderr = tokio::io::stderr(); - let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await; - }); - } - - let node = Self { - child, - config, - _key_file_handle: key_file_handle, - _net_key_file_handle: net_key_file_handle, - }; - - Ok(node) - } - - fn resolve_key_strategy( - desc_tag: &str, - source: KeySource, - parent_pid: u32, - ) -> Result<(Option, PathBuf, Option)> { - match source { - KeySource::Raw(raw_bytes) => { - let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?; - let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd)); - Ok((Some(handle), pseudo_path, Some(fd))) - } - KeySource::Path(existing_path) => Ok((None, existing_path, None)), - } - } - - fn find_available_port() -> Result { - let listener = TcpListener::bind("127.0.0.1:0")?; - let addr = listener.local_addr()?; - Ok(addr.port()) - } - - fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> { - #[cfg(target_os = "linux")] - { - let memfd_name = std::ffi::CString::new(desc_tag)?; - let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) }; - if fd < 0 { - return Err(anyhow!( - "memfd_create failed natively inside Linux kernel for tag: {}", - desc_tag - )); - } - let mut file = unsafe { File::from_raw_fd(fd) }; - file.write_all(&raw_key)?; - file.flush()?; - Ok((file, fd)) - } - - // Unreachable: `spawn` refuses to run elsewhere. - #[cfg(not(target_os = "linux"))] - { - let _ = (desc_tag, raw_key); - bail!("In-memory key files need Linux") - } - } - - pub async fn wait_for_exit(&mut self) -> std::io::Result { - self.child.wait().await - } -} - -impl Drop for SupervisedMycelium { - fn drop(&mut self) { - // The memfd key files go away with their handles. - let _ = self.child.start_kill(); - } -} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c45d117..14c5ee4 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -10,7 +10,6 @@ use std::pin::Pin; pub use crate::document::replica::Epoch; -pub mod daemon; pub mod firewall; pub mod mycelium; #[cfg(test)] diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs deleted file mode 100644 index d8d8689..0000000 --- a/src/transport/mycelium.rs +++ /dev/null @@ -1,548 +0,0 @@ -use anyhow::{Context, Result, anyhow}; -use async_trait::async_trait; -use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; -use reqwest::StatusCode; -use serde::{Deserialize, Serialize}; -use std::time::Duration; -use thiserror::Error; -use url::Url; - -use super::daemon::MyceliumSetup; -use crate::transport::wire; -use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; - -const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; - -/// The one Mycelium topic all of continuum's messages travel on; what a message is, its -/// payload says (`wire`). -const TOPIC: &str = "continuum.v1"; - -const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25); - -#[derive(Debug, Deserialize)] -#[serde(rename_all = "camelCase")] -struct InboundMessage { - id: String, - src_pk: String, - topic: String, - payload: String, -} - -// Deserialize only for debugging -#[derive(Debug, Serialize, Deserialize)] -struct OutboundMessage { - dst: Destination, - topic: String, - payload: String, -} - -// Deserialize only for debugging -#[derive(Debug, Serialize, Deserialize)] -struct Destination { - pk: String, -} - -/// The part of Mycelium's `/api/v1/admin` answer we use. -#[derive(Deserialize, Debug, Clone)] -#[serde(rename_all = "camelCase")] -pub struct MyceliumAdminInfo { - pub node_pubkey: String, -} - -fn is_pubkey(s: &str) -> bool { - s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) -} - -pub struct MyceliumTransport { - // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane - pub setup: tokio::sync::Mutex, - pub client: reqwest::Client, - pub poll_client: reqwest::Client, - pub messages_endpoint: Url, - pubkey_basepath: Url, - local_id: String, -} - -impl MyceliumTransport { - pub async fn connect(setup: MyceliumSetup) -> Result { - let rest_url = setup.rest_url.clone(); - - let client = reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build()?; - let poll_client = reqwest::Client::builder() - .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30)) - .build()?; - - let admin: MyceliumAdminInfo = client - .get(rest_url.join("/api/v1/admin")?) - .send() - .await? - .error_for_status()? - .json() - .await - .context("Querying local Mycelium identity")?; - - let transport = Self { - setup: tokio::sync::Mutex::new(setup), - client, - poll_client, - messages_endpoint: rest_url.join("/api/v1/messages")?, - pubkey_basepath: rest_url.join("/api/v1/pubkey/")?, - local_id: admin.node_pubkey, - }; - transport.discard_backlog().await; - Ok(transport) - } - - /// Drops what Mycelium queued for us before we started. A Mycelium that outlives us - /// (an external one, or across our restart) keeps accepting messages meant for us: - /// HELLO retries, keepalives, sync messages of sessions that ended with our previous - /// run. Processing them would only replay stale handshakes. Everything that still - /// matters is resent: peers retry their HELLOs, and sync starts over after the - /// handshake. Runs before we send anything, so no answer to us is dropped. - /// - /// Bounded in time, not count: the backlog is whatever piled up while we were away, - /// which a local loop drains in well under a second even for large swarms. Whatever - /// is still arriving after that is current traffic (peers that don't know we're - /// back), and it is processed normally, exactly as without this step. - async fn discard_backlog(&self) { - const DRAIN_FOR: Duration = Duration::from_secs(2); - let deadline = std::time::Instant::now() + DRAIN_FOR; - let mut discarded = 0u64; - while std::time::Instant::now() < deadline { - let response = self - .client - .get(self.messages_endpoint.clone()) - .query(&[("timeout", 0)]) - .query(&[("topic", encode_topic(TOPIC))]) - .send() - .await; - match response { - Ok(response) if response.status() == StatusCode::NO_CONTENT => break, - Ok(response) if response.status().is_success() => discarded += 1, - Ok(response) => { - tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox"); - break; - } - Err(error) => { - tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox"); - break; - } - } - } - if discarded > 0 { - tracing::info!( - discarded, - "Discarded messages Mycelium queued for us before we started" - ); - } - } - - async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result { - #[derive(Deserialize)] - struct PublicKeyResponse { - #[serde(rename = "publicKey")] - public_key: String, - } - - let response: PublicKeyResponse = self - .client - .get(self.pubkey_basepath.join(&ip.to_string())?) - .send() - .await? - .error_for_status()? - .json() - .await?; - Ok(response.public_key) - } -} - -#[async_trait] -impl Transport for MyceliumTransport { - fn local_id(&self) -> &str { - &self.local_id - } - - async fn resolve_seed(&self, hint: &str) -> Result { - if is_pubkey(hint) { - return Ok(hint.to_owned()); - } - - if let Ok(ip) = hint.parse::() { - return self.lookup_pubkey(&ip).await; - } - - let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0))) - .await - .context("DNS lookup timed out")?? - .find_map(|addr| match addr.ip() { - std::net::IpAddr::V6(ip) => Some(ip), - _ => None, - }) - .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?; - - self.lookup_pubkey(&ip).await - } - - fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { - encode_message(peer, message) - } - - async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - tracing::trace!(payload_bytes = message.body.len(), "Sending message"); - let response = self - .client - .post(self.messages_endpoint.clone()) - .header(reqwest::header::CONTENT_TYPE, "application/json") - .body(message.body.clone()) - .send() - .await?; - - let status = response.status(); - if !status.is_success() { - tracing::error!(status = %status, "Mycelium rejected outbound message"); - return Err(TransportError::MyceliumRejected { - peer: message.peer.clone(), - status, - } - .into()); - } - - // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing - // a message that never arrived. - if tracing::enabled!(tracing::Level::DEBUG) { - let Ok(body) = response - .bytes() - .await - .map_err(anyhow::Error::from) - .and_then(|bytes| Ok(serde_json::from_slice::(&bytes)?)) - else { - tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); - return Ok(()); - }; - let Ok(outbound) = serde_json::from_slice::(&message.body) else { - tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); - return Ok(()); - }; - let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { - tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); - return Ok(()); - }; - let Some(id) = body.get("id").and_then(|id| id.as_str()) else { - tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); - return Ok(()); - }; - tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); - } - Ok(()) - } - - async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { - tracing::info!( - endpoint = self.messages_endpoint.as_str(), - "Starting Mycelium message consumer" - ); - - loop { - tracing::trace!("Polling Mycelium inbox"); - - let response = match self - .poll_client - .get(self.messages_endpoint.clone()) - .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)]) - .query(&[("topic", encode_topic(TOPIC))]) - .send() - .await - { - Ok(resp) => resp, - Err(error) => { - tracing::warn!(error = %error, "Mycelium inbox poll failed"); - tokio::time::sleep(Duration::from_millis(250)).await; - continue; - } - }; - - let status = response.status(); - - if status == StatusCode::NO_CONTENT { - tracing::trace!("Mycelium inbox is empty"); - tokio::time::sleep(EMPTY_QUEUE_DELAY).await; - continue; - } - - if !status.is_success() { - tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status"); - tokio::time::sleep(Duration::from_millis(500)).await; - continue; - } - - let raw_bytes = match response.bytes().await { - Ok(bytes) => bytes, - Err(error) => { - tracing::warn!(error = %error, "Failed to read raw Mycelium response stream"); - tokio::time::sleep(Duration::from_millis(100)).await; - continue; - } - }; - - let inbound_message = match serde_json::from_slice::(&raw_bytes) { - Ok(msg) => msg, - Err(error) => { - tracing::error!( - error = %error, - raw_payload = %String::from_utf8_lossy(&raw_bytes), - "Failed to decode inbound Mycelium message structure" - ); - tokio::time::sleep(Duration::from_millis(500)).await; - continue; - } - }; - - match parse_message(&inbound_message) { - Ok((sender, parsed)) => { - tracing::debug!( - from = %inbound_message.src_pk, - message = ?parsed, - id = %inbound_message.id, - //payload_bytes = raw_bytes.len(), - "Received message" - ); - let payload_len = base64_decoded_len(&inbound_message.payload); - if let Err(error) = handler(sender, parsed, payload_len).await { - tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); - } - } - Err(transport_error) => { - tracing::error!( - error = %transport_error, - topic = %inbound_message.topic, - "Dropping poisoned packet: parsing failed gracefully" - ); - // No sleep required: we successfully popped a malformed packet and want to process the next one immediately - } - } - } - } - - async fn wait_for_exit(&self) -> std::io::Result { - let mut guard = self.setup.lock().await; - guard.wait_for_exit().await - } -} - -/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). -/// Also the simulator's wire format (`transport::sim`), so tests go through the same -/// encoding. -pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result { - let (outbound, payload_len) = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - kind: message.kind(), - payload_len, - }) -} - -/// The message in a body made by [`encode_message`]. -#[cfg(test)] -pub(crate) fn decode_message(body: &[u8]) -> Result { - let outbound: OutboundMessage = serde_json::from_slice(body)?; - Ok(decode_payload(&outbound.topic, &outbound.payload)?) -} - -fn encode_topic(topic: &str) -> String { - BASE64.encode(topic.as_bytes()) -} - -fn decode_topic(topic: &str) -> Result { - let bytes = BASE64.decode(topic)?; - Ok(String::from_utf8(bytes)?) -} - -fn encode_payload(payload: &Vec) -> String { - BASE64.encode(payload) -} - -/// Length of what padded base64 `encoded` decodes to. -fn base64_decoded_len(encoded: &str) -> usize { - let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count(); - (encoded.len() / 4 * 3).saturating_sub(padding) -} - -fn decode_payload(topic_encoded: &str, payload: &str) -> Result { - let topic = decode_topic(topic_encoded)?; - if topic != TOPIC { - return Err(TransportError::UnknownTopic(topic)); - } - Ok(wire::decode(&BASE64.decode(payload)?)?) -} - -fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> { - let sender = message.src_pk.clone(); - - if sender.is_empty() { - return Err(TransportError::NoSender); - } - - let parsed = decode_payload(&message.topic, &message.payload)?; - - Ok((sender, parsed)) -} - -/// The message for Mycelium's API, and the length of its payload. -fn format_message( - recipient: &str, - message: &ParsedMessage, -) -> Result<(OutboundMessage, usize), TransportError> { - let payload_bytes = wire::encode(message)?; - let outbound = OutboundMessage { - dst: Destination { - pk: recipient.to_owned(), - }, - topic: encode_topic(TOPIC), - payload: encode_payload(&payload_bytes), - }; - Ok((outbound, payload_bytes.len())) -} - -#[derive(Error, Debug)] -pub enum TransportError { - #[error("Mycelium delivered a message without a sender public key")] - NoSender, - - #[error("Base64 decoding failed: {0}")] - Base64Decode(#[from] base64::DecodeError), - - #[error("UTF-8 decoding failed: {0}")] - Utf8Decode(#[from] std::string::FromUtf8Error), - - #[error(transparent)] - Wire(#[from] wire::WireError), - - #[error("Unknown topic: {0}")] - UnknownTopic(String), - - #[error("Network HTTP request failed: {0}")] - Http(#[from] reqwest::Error), - - #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")] - MyceliumRejected { - peer: String, - status: reqwest::StatusCode, - }, -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::transport::{ - BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS, - }; - use automerge::ChangeHash; - - /// Mycelium splits messages into chunks of this size. - const MYCELIUM_CHUNK: usize = 1300; - - #[test] - fn largest_keepalive_fits_in_one_chunk() { - let keepalive = Keepalive { - heads: HeadsSummary::Heads( - (0..MAX_LISTED_HEADS as u8) - .map(|n| ChangeHash([n; 32])) - .collect(), - ), - epoch: Epoch { - generation: u64::MAX, - id: [0xff; 32], - }, - epoch_seq: u32::MAX, - ready: true, - syncs_with_you: true, - }; - let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap(); - // Mycelium's message data: a length byte, the topic, then our payload. - let size = 1 + TOPIC.len() + payload.len(); - assert!( - size + 50 <= MYCELIUM_CHUNK, - "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes" - ); - let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else { - panic!("not a PONG"); - }; - assert_eq!(decoded.heads, keepalive.heads); - } - - #[test] - fn blob_messages_round_trip() { - let epoch = Epoch { - generation: 3, - id: [7; 32], - }; - let id = BlobId::Change(ChangeHash([9; 32])); - let decode_blob = |payload: &[u8]| match wire::decode(payload).unwrap() { - ParsedMessage::Blob(blob) => blob, - other => panic!("not a blob: {other:?}"), - }; - let chunk = ParsedMessage::Blob(BlobMessage { - epoch, - epoch_seq: 4, - id, - body: BlobBody::Chunk { - index: 5, - data: vec![1u8; 64 * 1024].into(), - }, - }); - let bytes = wire::encode(&chunk).unwrap(); - // A 64 KiB chunk costs only a few bytes of framing. - assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len()); - let decoded = decode_blob(&bytes); - assert_eq!(decoded.id, id); - assert_eq!(decoded.epoch, epoch); - assert_eq!(decoded.epoch_seq, 4); - assert!( - matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) - ); - - let manifest = ParsedMessage::Blob(BlobMessage { - epoch, - epoch_seq: 4, - id, - body: BlobBody::Manifest { - size: 100, - chunk_size: 64, - hashes: vec![[1; 32], [2; 32]], - }, - }); - let decoded = decode_blob(&wire::encode(&manifest).unwrap()); - assert!(matches!( - decoded.body, - BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2 - )); - } - - #[test] - fn messages_round_trip_through_the_api_body() { - let keepalive = ParsedMessage::Ping(Keepalive { - heads: HeadsSummary::of(vec![ChangeHash([1; 32])]), - epoch: Epoch::GENESIS, - epoch_seq: 0, - ready: true, - syncs_with_you: true, - }); - let encoded = encode_message("peer", &keepalive).unwrap(); - assert!(matches!( - decode_message(&encoded.body).unwrap(), - ParsedMessage::Ping(_) - )); - } - - #[test] - fn listed_heads_equal_their_digest() { - let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect(); - let listed = HeadsSummary::of(heads.clone()); - let digest = HeadsSummary::Digest(listed.digest()); - assert_eq!(listed, digest); - assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec())); - } -} diff --git a/src/transport/mycelium/daemon.rs b/src/transport/mycelium/daemon.rs new file mode 100644 index 0000000..80b6be5 --- /dev/null +++ b/src/transport/mycelium/daemon.rs @@ -0,0 +1,337 @@ +//! Running Mycelium for this node: either a daemon the operator runs (its REST URL), or +//! one we start and supervise (`SupervisedMycelium`): its keys handed over in memory, +//! never on disk, and it dies with us. + +use anyhow::{Context, Result, anyhow, bail}; +use std::fs::File; +use std::io::Write; +use std::net::TcpListener; +use std::os::unix::io::FromRawFd; +use std::path::PathBuf; +use std::process::Stdio; +use tokio::process::{Child, Command}; +use url::Url; + +/// Represents where a 32-byte cryptographic key originates. +#[derive(Debug, Clone)] +pub enum KeySource { + Raw([u8; 32]), + Path(PathBuf), +} + +/// Atomic structure ensuring private network credentials are always handled together. +#[derive(Debug, Clone)] +pub struct PrivateNetworkConfig { + pub name: String, + pub key: KeySource, +} + +/// Comprehensive parameters required to spawn or target an instance. +#[derive(Debug, Clone)] +pub struct MyceliumSpawnConfig { + pub rest_url_opt: Option, + pub key_file_opt: Option, + pub network_key_file: Option, + pub network_name: String, + pub tcp_port: Option, + pub peers: Vec, +} + +impl Default for MyceliumSpawnConfig { + fn default() -> Self { + Self { + rest_url_opt: None, + key_file_opt: None, + network_key_file: None, + network_name: "something".to_string(), + tcp_port: None, + peers: vec!["tls://127.0.0.1:9651".to_string()], + } + } +} + +pub enum MyceliumRuntime { + Supervised(Box), + External, +} + +pub struct MyceliumSetup { + pub runtime: MyceliumRuntime, + pub rest_url: Url, +} + +impl MyceliumSetup { + /// Evaluates the configuration payload and initializes the chosen execution strategy. + pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result { + // TODO warn for all ignored options if rest_url_opt is set + if let Some(url) = config.rest_url_opt { + tracing::info!( + "Configured to interface with an external standalone Mycelium instance at: {}", + url + ); + return Ok(Self { + runtime: MyceliumRuntime::External, + rest_url: url, + }); + } + + tracing::info!( + "Configured to initialize an embedded user-space Mycelium background worker" + ); + + let key = match config.key_file_opt { + Some(key_file) => KeySource::Path(key_file), + None => { + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes)?; + KeySource::Raw(bytes) + } + }; + + let private_net_config = config + .network_key_file + .map(|net_key_path| PrivateNetworkConfig { + name: config.network_name.clone(), + key: KeySource::Path(net_key_path), + }); + + let supervised_node = + SupervisedMycelium::spawn(key, private_net_config, config.tcp_port, &config.peers) + .await + .context("Failed to securely bootstrap embedded Mycelium background worker")?; + + let resolved_url = Url::parse(&format!( + "http://127.0.0.1:{}", + supervised_node.config.api_port + ))?; + + Ok(Self { + runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)), + rest_url: resolved_url, + }) + } + + /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely. + pub async fn wait_for_exit(&mut self) -> std::io::Result { + match &mut self.runtime { + MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, + MyceliumRuntime::External => { + std::future::pending::>().await + } + } + } +} + +#[derive(Debug, Clone)] +pub struct MyceliumNodeConfig { + pub tcp_port: u16, + pub api_port: u16, + pub rpc_port: u16, + pub private_network: Option, +} + +pub struct SupervisedMycelium { + child: Child, + pub config: MyceliumNodeConfig, + _key_file_handle: Option, + _net_key_file_handle: Option, +} + +impl SupervisedMycelium { + async fn spawn( + node_key: KeySource, + private_network: Option, + explicit_tcp: Option, + peers: &[String], + ) -> Result { + // Linux only: keys are handed over in memory (memfd, never on disk), and the + // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself. + if cfg!(not(target_os = "linux")) { + bail!( + "Supervising Mycelium is only supported on Linux; run myceliumd yourself \ + and pass --mycelium-rest-url" + ); + } + + let tcp_port = match explicit_tcp { + Some(p) => p, + None => Self::find_available_port().context("Failed to allocate TCP wire port")?, + }; + // Local only, so any free port will do. + let api_port = Self::find_available_port().context("Failed to allocate API port")?; + let rpc_port = Self::find_available_port().context("Failed to allocate JSON-RPC port")?; + + let config = MyceliumNodeConfig { + tcp_port, + api_port, + rpc_port, + private_network: private_network.as_ref().map(|pn| pn.name.clone()), + }; + + let parent_pid = std::process::id(); + + let (key_file_handle, key_file_path, node_fd) = + Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; + + let (net_key_file_handle, net_key_file_path, net_fd) = match private_network { + Some(net_cfg) => { + let (h, p, fd) = + Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; + (h, Some(p), fd) + } + None => (None, None, None), + }; + + let api_addr = format!("127.0.0.1:{}", config.api_port); + let rpc_addr = format!("127.0.0.1:{}", config.rpc_port); + + let mut cmd = Command::new("mycelium"); + + let fds_to_inherit: Vec = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); + // The child must never outlive us. An orphaned Mycelium keeps running with our + // key; the next one we start then shares its address, and replies (including + // Mycelium's own message ACKs) go to whichever instance the routers picked, + // leaving our messages pending forever. Drop only runs on a clean shutdown, so + // let the kernel signal the child when we die, however we die. + let parent_pid = std::process::id() as libc::pid_t; + unsafe { + cmd.pre_exec(move || { + for &fd in &fds_to_inherit { + libc::fcntl(fd, libc::F_SETFD, 0); + } + // Own process group: a Ctrl-C in the terminal goes to continuum only, + // which then stops Mycelium itself, instead of both exiting at once (which + // shows up as "Mycelium runtime process exited"). + if libc::setpgid(0, 0) != 0 { + return Err(std::io::Error::last_os_error()); + } + #[cfg(target_os = "linux")] + { + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { + return Err(std::io::Error::last_os_error()); + } + // We may have died between fork and prctl. + if libc::getppid() != parent_pid { + libc::_exit(1); + } + } + Ok(()) + }); + } + + cmd.arg("--key-file") + .arg(&key_file_path) + .arg("--tcp-listen-port") + .arg(config.tcp_port.to_string()) + .arg("--disable-quic") + .arg("--api-addr") + .arg(&api_addr) + .arg("--jsonrpc-addr") + .arg(&rpc_addr) + .arg("--no-tun"); + + // Peers come from `--mycelium-peers`; no local multicast discovery. + cmd.arg("--disable-peer-discovery"); + + if let Some(ref net_name) = config.private_network { + cmd.arg("--network-name").arg(net_name); + } + if let Some(ref net_key_path) = net_key_file_path { + cmd.arg("--network-key-file").arg(net_key_path); + } + + if !peers.is_empty() { + cmd.arg("--peers").arg(peers.join(" ")); + } + + cmd.env("RUST_LOG", "info"); + + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + + let mut child = cmd + .spawn() + .context("Failed to execute mycelium process daemon")?; + + // Its output goes where ours does. + if let Some(mut stdout) = child.stdout.take() { + tokio::spawn(async move { + let mut host_stdout = tokio::io::stdout(); + let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await; + }); + } + + if let Some(mut stderr) = child.stderr.take() { + tokio::spawn(async move { + let mut host_stderr = tokio::io::stderr(); + let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await; + }); + } + + let node = Self { + child, + config, + _key_file_handle: key_file_handle, + _net_key_file_handle: net_key_file_handle, + }; + + Ok(node) + } + + fn resolve_key_strategy( + desc_tag: &str, + source: KeySource, + parent_pid: u32, + ) -> Result<(Option, PathBuf, Option)> { + match source { + KeySource::Raw(raw_bytes) => { + let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?; + let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd)); + Ok((Some(handle), pseudo_path, Some(fd))) + } + KeySource::Path(existing_path) => Ok((None, existing_path, None)), + } + } + + fn find_available_port() -> Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + let addr = listener.local_addr()?; + Ok(addr.port()) + } + + fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> { + #[cfg(target_os = "linux")] + { + let memfd_name = std::ffi::CString::new(desc_tag)?; + let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) }; + if fd < 0 { + return Err(anyhow!( + "memfd_create failed natively inside Linux kernel for tag: {}", + desc_tag + )); + } + let mut file = unsafe { File::from_raw_fd(fd) }; + file.write_all(&raw_key)?; + file.flush()?; + Ok((file, fd)) + } + + // Unreachable: `spawn` refuses to run elsewhere. + #[cfg(not(target_os = "linux"))] + { + let _ = (desc_tag, raw_key); + bail!("In-memory key files need Linux") + } + } + + pub async fn wait_for_exit(&mut self) -> std::io::Result { + self.child.wait().await + } +} + +impl Drop for SupervisedMycelium { + fn drop(&mut self) { + // The memfd key files go away with their handles. + let _ = self.child.start_kill(); + } +} diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs new file mode 100644 index 0000000..cbae567 --- /dev/null +++ b/src/transport/mycelium/mod.rs @@ -0,0 +1,552 @@ +use anyhow::{Context, Result, anyhow}; +use async_trait::async_trait; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; +use reqwest::StatusCode; +use serde::{Deserialize, Serialize}; +use std::time::Duration; +use thiserror::Error; +use url::Url; + +use crate::transport::wire; +use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; + +pub mod daemon; + +pub use daemon::{MyceliumSetup, MyceliumSpawnConfig}; + +const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; + +/// The one Mycelium topic all of continuum's messages travel on; what a message is, its +/// payload says (`wire`). +const TOPIC: &str = "continuum.v1"; + +const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25); + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct InboundMessage { + id: String, + src_pk: String, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct OutboundMessage { + dst: Destination, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct Destination { + pk: String, +} + +/// The part of Mycelium's `/api/v1/admin` answer we use. +#[derive(Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct MyceliumAdminInfo { + pub node_pubkey: String, +} + +fn is_pubkey(s: &str) -> bool { + s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) +} + +pub struct MyceliumTransport { + /// The daemon, kept to wait for its exit. + setup: tokio::sync::Mutex, + client: reqwest::Client, + /// For the long-polling inbox (its timeout is longer). + poll_client: reqwest::Client, + messages_endpoint: Url, + pubkey_basepath: Url, + local_id: String, +} + +impl MyceliumTransport { + pub async fn connect(setup: MyceliumSetup) -> Result { + let rest_url = setup.rest_url.clone(); + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()?; + let poll_client = reqwest::Client::builder() + .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30)) + .build()?; + + let admin: MyceliumAdminInfo = client + .get(rest_url.join("/api/v1/admin")?) + .send() + .await? + .error_for_status()? + .json() + .await + .context("Querying local Mycelium identity")?; + + let transport = Self { + setup: tokio::sync::Mutex::new(setup), + client, + poll_client, + messages_endpoint: rest_url.join("/api/v1/messages")?, + pubkey_basepath: rest_url.join("/api/v1/pubkey/")?, + local_id: admin.node_pubkey, + }; + transport.discard_backlog().await; + Ok(transport) + } + + /// Drops what Mycelium queued for us before we started. A Mycelium that outlives us + /// (an external one, or across our restart) keeps accepting messages meant for us: + /// HELLO retries, keepalives, sync messages of sessions that ended with our previous + /// run. Processing them would only replay stale handshakes. Everything that still + /// matters is resent: peers retry their HELLOs, and sync starts over after the + /// handshake. Runs before we send anything, so no answer to us is dropped. + /// + /// Bounded in time, not count: the backlog is whatever piled up while we were away, + /// which a local loop drains in well under a second even for large swarms. Whatever + /// is still arriving after that is current traffic (peers that don't know we're + /// back), and it is processed normally, exactly as without this step. + async fn discard_backlog(&self) { + const DRAIN_FOR: Duration = Duration::from_secs(2); + let deadline = std::time::Instant::now() + DRAIN_FOR; + let mut discarded = 0u64; + while std::time::Instant::now() < deadline { + let response = self + .client + .get(self.messages_endpoint.clone()) + .query(&[("timeout", 0)]) + .query(&[("topic", encode_topic(TOPIC))]) + .send() + .await; + match response { + Ok(response) if response.status() == StatusCode::NO_CONTENT => break, + Ok(response) if response.status().is_success() => discarded += 1, + Ok(response) => { + tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox"); + break; + } + Err(error) => { + tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox"); + break; + } + } + } + if discarded > 0 { + tracing::info!( + discarded, + "Discarded messages Mycelium queued for us before we started" + ); + } + } + + async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result { + #[derive(Deserialize)] + struct PublicKeyResponse { + #[serde(rename = "publicKey")] + public_key: String, + } + + let response: PublicKeyResponse = self + .client + .get(self.pubkey_basepath.join(&ip.to_string())?) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(response.public_key) + } +} + +#[async_trait] +impl Transport for MyceliumTransport { + fn local_id(&self) -> &str { + &self.local_id + } + + async fn resolve_seed(&self, hint: &str) -> Result { + if is_pubkey(hint) { + return Ok(hint.to_owned()); + } + + if let Ok(ip) = hint.parse::() { + return self.lookup_pubkey(&ip).await; + } + + let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0))) + .await + .context("DNS lookup timed out")?? + .find_map(|addr| match addr.ip() { + std::net::IpAddr::V6(ip) => Some(ip), + _ => None, + }) + .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?; + + self.lookup_pubkey(&ip).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { + encode_message(peer, message) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + tracing::trace!(payload_bytes = message.body.len(), "Sending message"); + let response = self + .client + .post(self.messages_endpoint.clone()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(message.body.clone()) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + tracing::error!(status = %status, "Mycelium rejected outbound message"); + return Err(TransportError::MyceliumRejected { + peer: message.peer.clone(), + status, + } + .into()); + } + + // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing + // a message that never arrived. + if tracing::enabled!(tracing::Level::DEBUG) { + let Ok(body) = response + .bytes() + .await + .map_err(anyhow::Error::from) + .and_then(|bytes| Ok(serde_json::from_slice::(&bytes)?)) + else { + tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); + return Ok(()); + }; + let Ok(outbound) = serde_json::from_slice::(&message.body) else { + tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); + return Ok(()); + }; + let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { + tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); + return Ok(()); + }; + let Some(id) = body.get("id").and_then(|id| id.as_str()) else { + tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); + return Ok(()); + }; + tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); + } + Ok(()) + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + tracing::info!( + endpoint = self.messages_endpoint.as_str(), + "Starting Mycelium message consumer" + ); + + loop { + tracing::trace!("Polling Mycelium inbox"); + + let response = match self + .poll_client + .get(self.messages_endpoint.clone()) + .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)]) + .query(&[("topic", encode_topic(TOPIC))]) + .send() + .await + { + Ok(resp) => resp, + Err(error) => { + tracing::warn!(error = %error, "Mycelium inbox poll failed"); + tokio::time::sleep(Duration::from_millis(250)).await; + continue; + } + }; + + let status = response.status(); + + if status == StatusCode::NO_CONTENT { + tracing::trace!("Mycelium inbox is empty"); + tokio::time::sleep(EMPTY_QUEUE_DELAY).await; + continue; + } + + if !status.is_success() { + tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status"); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + + let raw_bytes = match response.bytes().await { + Ok(bytes) => bytes, + Err(error) => { + tracing::warn!(error = %error, "Failed to read raw Mycelium response stream"); + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + }; + + let inbound_message = match serde_json::from_slice::(&raw_bytes) { + Ok(msg) => msg, + Err(error) => { + tracing::error!( + error = %error, + raw_payload = %String::from_utf8_lossy(&raw_bytes), + "Failed to decode inbound Mycelium message structure" + ); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + }; + + match parse_message(&inbound_message) { + Ok((sender, parsed)) => { + tracing::debug!( + from = %inbound_message.src_pk, + message = ?parsed, + id = %inbound_message.id, + //payload_bytes = raw_bytes.len(), + "Received message" + ); + let payload_len = base64_decoded_len(&inbound_message.payload); + if let Err(error) = handler(sender, parsed, payload_len).await { + tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Err(transport_error) => { + tracing::error!( + error = %transport_error, + topic = %inbound_message.topic, + "Dropping poisoned packet: parsing failed gracefully" + ); + // No sleep required: we successfully popped a malformed packet and want to process the next one immediately + } + } + } + } + + async fn wait_for_exit(&self) -> std::io::Result { + let mut guard = self.setup.lock().await; + guard.wait_for_exit().await + } +} + +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result { + let (outbound, payload_len) = format_message(peer, message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + body: serde_json::to_vec(&outbound)?.into(), + kind: message.kind(), + payload_len, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + +fn encode_topic(topic: &str) -> String { + BASE64.encode(topic.as_bytes()) +} + +fn decode_topic(topic: &str) -> Result { + let bytes = BASE64.decode(topic)?; + Ok(String::from_utf8(bytes)?) +} + +fn encode_payload(payload: &Vec) -> String { + BASE64.encode(payload) +} + +/// Length of what padded base64 `encoded` decodes to. +fn base64_decoded_len(encoded: &str) -> usize { + let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count(); + (encoded.len() / 4 * 3).saturating_sub(padding) +} + +fn decode_payload(topic_encoded: &str, payload: &str) -> Result { + let topic = decode_topic(topic_encoded)?; + if topic != TOPIC { + return Err(TransportError::UnknownTopic(topic)); + } + Ok(wire::decode(&BASE64.decode(payload)?)?) +} + +fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> { + let sender = message.src_pk.clone(); + + if sender.is_empty() { + return Err(TransportError::NoSender); + } + + let parsed = decode_payload(&message.topic, &message.payload)?; + + Ok((sender, parsed)) +} + +/// The message for Mycelium's API, and the length of its payload. +fn format_message( + recipient: &str, + message: &ParsedMessage, +) -> Result<(OutboundMessage, usize), TransportError> { + let payload_bytes = wire::encode(message)?; + let outbound = OutboundMessage { + dst: Destination { + pk: recipient.to_owned(), + }, + topic: encode_topic(TOPIC), + payload: encode_payload(&payload_bytes), + }; + Ok((outbound, payload_bytes.len())) +} + +#[derive(Error, Debug)] +pub enum TransportError { + #[error("Mycelium delivered a message without a sender public key")] + NoSender, + + #[error("Base64 decoding failed: {0}")] + Base64Decode(#[from] base64::DecodeError), + + #[error("UTF-8 decoding failed: {0}")] + Utf8Decode(#[from] std::string::FromUtf8Error), + + #[error(transparent)] + Wire(#[from] wire::WireError), + + #[error("Unknown topic: {0}")] + UnknownTopic(String), + + #[error("Network HTTP request failed: {0}")] + Http(#[from] reqwest::Error), + + #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")] + MyceliumRejected { + peer: String, + status: reqwest::StatusCode, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::{ + BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS, + }; + use automerge::ChangeHash; + + /// Mycelium splits messages into chunks of this size. + const MYCELIUM_CHUNK: usize = 1300; + + #[test] + fn largest_keepalive_fits_in_one_chunk() { + let keepalive = Keepalive { + heads: HeadsSummary::Heads( + (0..MAX_LISTED_HEADS as u8) + .map(|n| ChangeHash([n; 32])) + .collect(), + ), + epoch: Epoch { + generation: u64::MAX, + id: [0xff; 32], + }, + epoch_seq: u32::MAX, + ready: true, + syncs_with_you: true, + }; + let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap(); + // Mycelium's message data: a length byte, the topic, then our payload. + let size = 1 + TOPIC.len() + payload.len(); + assert!( + size + 50 <= MYCELIUM_CHUNK, + "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes" + ); + let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else { + panic!("not a PONG"); + }; + assert_eq!(decoded.heads, keepalive.heads); + } + + #[test] + fn blob_messages_round_trip() { + let epoch = Epoch { + generation: 3, + id: [7; 32], + }; + let id = BlobId::Change(ChangeHash([9; 32])); + let decode_blob = |payload: &[u8]| match wire::decode(payload).unwrap() { + ParsedMessage::Blob(blob) => blob, + other => panic!("not a blob: {other:?}"), + }; + let chunk = ParsedMessage::Blob(BlobMessage { + epoch, + epoch_seq: 4, + id, + body: BlobBody::Chunk { + index: 5, + data: vec![1u8; 64 * 1024].into(), + }, + }); + let bytes = wire::encode(&chunk).unwrap(); + // A 64 KiB chunk costs only a few bytes of framing. + assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len()); + let decoded = decode_blob(&bytes); + assert_eq!(decoded.id, id); + assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); + assert!( + matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) + ); + + let manifest = ParsedMessage::Blob(BlobMessage { + epoch, + epoch_seq: 4, + id, + body: BlobBody::Manifest { + size: 100, + chunk_size: 64, + hashes: vec![[1; 32], [2; 32]], + }, + }); + let decoded = decode_blob(&wire::encode(&manifest).unwrap()); + assert!(matches!( + decoded.body, + BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2 + )); + } + + #[test] + fn messages_round_trip_through_the_api_body() { + let keepalive = ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(vec![ChangeHash([1; 32])]), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: true, + syncs_with_you: true, + }); + let encoded = encode_message("peer", &keepalive).unwrap(); + assert!(matches!( + decode_message(&encoded.body).unwrap(), + ParsedMessage::Ping(_) + )); + } + + #[test] + fn listed_heads_equal_their_digest() { + let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect(); + let listed = HeadsSummary::of(heads.clone()); + let digest = HeadsSummary::Digest(listed.digest()); + assert_eq!(listed, digest); + assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec())); + } +} -- cgit v1.3.1