diff options
Diffstat (limited to 'src/transport/mycelium.rs.orig')
| -rw-r--r-- | src/transport/mycelium.rs.orig | 975 |
1 files changed, 0 insertions, 975 deletions
diff --git a/src/transport/mycelium.rs.orig b/src/transport/mycelium.rs.orig deleted file mode 100644 index 9d1fb90..0000000 --- a/src/transport/mycelium.rs.orig +++ /dev/null @@ -1,975 +0,0 @@ -use anyhow::{Context, Result, anyhow, bail}; -use async_trait::async_trait; -use automerge::ChangeHash; -use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; -use reqwest::StatusCode; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use serde_bytes::{ByteBuf, Bytes}; -use std::borrow::Cow; -use std::boxed::Box; -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 std::time::Duration; -use thiserror::Error; -use tokio::process::{Child, Command}; -use url::Url; - -use crate::transport::EncodedMessage; -use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; -use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS}; -use crate::transport::{MessageHandler, ParsedMessage, Transport}; - -const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; - -const TOPIC_PREFIX: &str = "continuum.v1."; -const TOPIC_HELLO: &str = "continuum.v1.hello"; -const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply"; -const TOPIC_SYNC: &str = "continuum.v1.sync"; -const TOPIC_PING: &str = "continuum.v1.ping"; -const TOPIC_PONG: &str = "continuum.v1.pong"; - -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, -} - -/// 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<Url>, - pub key_file_opt: Option<PathBuf>, - pub network_key_file: Option<PathBuf>, - pub network_name: String, - pub tcp_port: Option<u16>, - pub api_port: Option<u16>, - pub rpc_port: Option<u16>, - pub discovery_port: Option<u16>, - pub peers: Vec<String>, -} - -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<SupervisedMycelium>), - 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<Self> { - // 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<std::process::ExitStatus> { - match &mut self.runtime { - MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, - MyceliumRuntime::External => { - std::future::pending::<std::io::Result<std::process::ExitStatus>>().await - } - } - } -} - -/// 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, -} - -#[derive(Debug, Clone)] -pub struct MyceliumNodeConfig { - pub tcp_port: u16, - pub api_port: u16, - pub rpc_port: u16, - pub discovery_port: Option<u16>, - pub private_network: Option<String>, -} - -pub struct SupervisedMycelium { - child: Child, - pub config: MyceliumNodeConfig, - _key_file_handle: Option<File>, - _net_key_file_handle: Option<File>, -} - -impl SupervisedMycelium { - async fn spawn( - node_key: KeySource, - private_network: Option<PrivateNetworkConfig>, - explicit_tcp: Option<u16>, - explicit_api: Option<u16>, - explicit_rpc: Option<u16>, - explicit_discovery_port: Option<u16>, - peers: &[String], - ) -> Result<Self> { - // 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<i32> = [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<File>, PathBuf, Option<i32>)> { - 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<u16> { - 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<std::process::ExitStatus> { - 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(); - } -} - -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<MyceliumSetup>, - 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<Self> { - 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")?; - - Ok(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, - }) - } - - async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> { - #[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<String> { - if is_pubkey(hint) { - return Ok(hint.to_owned()); - } - - if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() { - 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<EncodedMessage> { - let outbound = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - }) - } - - 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::<serde_json::Value>(&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::<OutboundMessage>(&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_PREFIX))]) - .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::<InboundMessage>(&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" - ); - if let Err(error) = handler(sender, parsed).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<std::process::ExitStatus> { - let mut guard = self.setup.lock().await; - guard.wait_for_exit().await - } -} - -fn encode_topic(topic: &str) -> String { - BASE64.encode(topic.as_bytes()) -} - -fn decode_topic(topic: &str) -> Result<String, TransportError> { - let bytes = BASE64.decode(topic)?; - Ok(String::from_utf8(bytes)?) -} - -fn encode_payload(payload: &Vec<u8>) -> String { - BASE64.encode(payload) -} - -fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> { - let topic: &str = &decode_topic(topic_encoded)?; - let bytes = BASE64.decode(payload)?; - match topic { - TOPIC_HELLO => { - let msg = bytes_to_hello_request(&bytes)?; - Ok(ParsedMessage::Hello(msg)) - } - TOPIC_HELLO_REPLY => { - let msg = bytes_to_hello_reply(&bytes)?; - Ok(ParsedMessage::HelloReply(msg)) - } - TOPIC_SYNC => { - let msg = bytes_to_sync_envelope(&bytes)?; - Ok(ParsedMessage::Sync(msg)) - } - TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)), - TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)), - t => Err(TransportError::UnknownTopic(t.to_string())), - } -} - -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)) -} - -fn format_message( - recipient: &str, - message: &ParsedMessage, -) -> Result<OutboundMessage, TransportError> { - let (topic, payload_bytes) = match message { - ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?), - ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?), - ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?), - ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?), - ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?), - }; - Ok(OutboundMessage { - dst: Destination { - pk: recipient.to_owned(), - }, - topic: encode_topic(topic), - payload: encode_payload(&payload_bytes), - }) -} - -// ===================================================================== -// 1. WIRE SHADOWS -// ===================================================================== -// -// rmp_serde encodes structs as arrays. Fields added later are appended at the end and -// marked #[serde(default)], so a message from a node that predates them still decodes -// (an absent epoch is the genesis epoch). - -/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch. -#[derive(Serialize, Deserialize, Default)] -struct EpochShadow { - generation: u64, - id: ByteBuf, -} - -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { - EpochShadow { - generation: epoch.generation, - id: ByteBuf::from(epoch.id.to_vec()), - } -} - -fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { - if shadow.generation == 0 && shadow.id.is_empty() { - return Ok(Epoch::GENESIS); - } - let id = shadow - .id - .as_slice() - .try_into() - .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?; - Ok(Epoch { - generation: shadow.generation, - id, - }) -} - -/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set. -#[derive(Serialize, Deserialize)] -struct KeepaliveShadow { - heads: Vec<ByteBuf>, - digest: ByteBuf, - epoch: EpochShadow, - ready: bool, - syncs_with_you: bool, -} - -fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> { - let (heads, digest) = match &msg.heads { - HeadsSummary::Heads(heads) => ( - heads - .iter() - .map(|head| ByteBuf::from(head.0.to_vec())) - .collect(), - ByteBuf::new(), - ), - HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())), - }; - rmp_serde::to_vec(&KeepaliveShadow { - heads, - digest, - epoch: epoch_to_shadow(&msg.epoch), - ready: msg.ready, - syncs_with_you: msg.syncs_with_you, - }) -} - -fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { - let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?; - let heads = if shadow.digest.is_empty() { - if shadow.heads.len() > MAX_LISTED_HEADS { - return Err(TransportError::InvalidPayload(format!( - "more than {MAX_LISTED_HEADS} heads listed" - ))); - } - let heads = shadow - .heads - .iter() - .map(|head| ChangeHash::try_from(head.as_slice())) - .collect::<Result<Vec<_>, _>>() - .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?; - HeadsSummary::Heads(heads) - } else { - let digest = - shadow.digest.as_slice().try_into().map_err(|_| { - TransportError::InvalidPayload("heads digest must be 32 bytes".into()) - })?; - HeadsSummary::Digest(digest) - }; - Ok(Keepalive { - heads, - epoch: epoch_from_shadow(shadow.epoch)?, - ready: shadow.ready, - syncs_with_you: shadow.syncs_with_you, - }) -} - -#[derive(Serialize, Deserialize)] -pub struct HelloRequestShadow { - #[serde(default)] - epoch: EpochShadow, - #[serde(default)] - auth: ByteBuf, -} - -#[derive(Serialize, Deserialize)] -pub struct HelloReplyShadow<'a> { - // Cow allows zero-copy references when saving, but creates an owned String when loading - pub swarm_id: Cow<'a, str>, - #[serde(default)] - pub peers: Vec<String>, - #[serde(default)] - epoch: EpochShadow, - #[serde(default)] - auth: ByteBuf, -} - -#[derive(Serialize, Deserialize)] -pub struct SyncEnvelopeShadow<'a> { - pub swarm_id: Cow<'a, str>, - #[serde(with = "automerge_msg_codec")] - pub message: automerge::sync::Message, - #[serde(default)] - epoch: EpochShadow, - /// Absent from nodes that predate it: never counts as ready. - #[serde(default)] - ready: bool, -} - -// ===================================================================== -// 2. CUSTOM CODEC -// ===================================================================== -mod automerge_msg_codec { - use super::*; - use automerge::sync::Message; - - pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error> - where - S: Serializer, - { - let bytes = msg.clone().encode(); - Bytes::new(&bytes).serialize(serializer) - } - - pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error> - where - D: Deserializer<'de>, - { - let byte_buf = ByteBuf::deserialize(deserializer)?; - Message::decode(&byte_buf) - .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e))) - } -} - -// ===================================================================== -// 3. CLEAN CONVERSION INTERFACE -// ===================================================================== - -// --- HelloRequest --- -pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { - rmp_serde::to_vec(&HelloRequestShadow { - epoch: epoch_to_shadow(&msg.epoch), - auth: ByteBuf::from(msg.auth.clone()), - }) -} -pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { - let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; - Ok(HelloRequest { - epoch: epoch_from_shadow(shadow.epoch)?, - auth: shadow.auth.into_vec(), - }) -} - -// --- HelloReply --- -pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> { - let shadow = HelloReplyShadow { - swarm_id: Cow::Borrowed(&msg.swarm_id), - peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), - auth: ByteBuf::from(msg.auth.clone()), - }; - rmp_serde::to_vec(&shadow) -} -pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> { - let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?; - Ok(HelloReply { - swarm_id: shadow.swarm_id.into_owned(), - peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, - auth: shadow.auth.into_vec(), - }) -} - -// --- SyncEnvelope --- -pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> { - let shadow = SyncEnvelopeShadow { - swarm_id: Cow::Borrowed(&msg.swarm_id), - message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), - ready: msg.ready, - }; - rmp_serde::to_vec(&shadow) -} -pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> { - let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; - Ok(SyncEnvelope { - swarm_id: shadow.swarm_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, - message: shadow.message, - ready: shadow.ready, - }) -} - -#[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("MessagePack serialization failed: {0}")] - MessagePackEncode(#[from] rmp_serde::encode::Error), - - #[error("MessagePack deserialization failed: {0}")] - MessagePackDecode(#[from] rmp_serde::decode::Error), - - #[error("Unknown topic: {0}")] - UnknownTopic(String), - - #[error("Invalid payload: {0}")] - InvalidPayload(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, - }, -} |