summaryrefslogtreecommitdiffstats
path: root/src/transport/mycelium.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/mycelium.rs')
-rw-r--r--src/transport/mycelium.rs1276
1 files changed, 0 insertions, 1276 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
deleted file mode 100644
index efab475..0000000
--- a/src/transport/mycelium.rs
+++ /dev/null
@@ -1,1276 +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::{BlobBody, BlobId, BlobMessage};
-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 TOPIC_BLOB: &str = "continuum.v1.blob";
-
-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")?;
-
- 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_PREFIX))])
- .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<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> {
- 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::<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"
- );
- 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<std::process::ExitStatus> {
- 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<EncodedMessage> {
- 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<ParsedMessage> {
- 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<String, TransportError> {
- let bytes = BASE64.decode(topic)?;
- Ok(String::from_utf8(bytes)?)
-}
-
-fn encode_payload(payload: &Vec<u8>) -> 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<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)?)),
- TOPIC_BLOB => Ok(ParsedMessage::Blob(bytes_to_blob(&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))
-}
-
-/// The message for Mycelium's API, and the length of its payload.
-fn format_message(
- recipient: &str,
- message: &ParsedMessage,
-) -> Result<(OutboundMessage, usize), 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)?),
- ParsedMessage::Blob(blob) => (TOPIC_BLOB, blob_to_bytes(blob)?),
- };
- let outbound = OutboundMessage {
- dst: Destination {
- pk: recipient.to_owned(),
- },
- topic: encode_topic(topic),
- payload: encode_payload(&payload_bytes),
- };
- Ok((outbound, payload_bytes.len()))
-}
-
-// =====================================================================
-// 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,
- /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`).
- #[serde(default)]
- seq: u32,
-}
-
-fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow {
- EpochShadow {
- generation: epoch.generation,
- id: ByteBuf::from(epoch.id.to_vec()),
- seq,
- }
-}
-
-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, msg.epoch_seq),
- 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)?,
- epoch_seq: shadow.epoch.seq,
- ready: shadow.ready,
- syncs_with_you: shadow.syncs_with_you,
- })
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct HelloRequestShadow {
- swarm_id: Option<String>,
- epoch: EpochShadow,
- auth: ByteBuf,
- ready: bool,
- run_id: 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,
- ready: bool,
- run_id: ByteBuf,
- away_ms: Option<u64>,
- sender_run_id: 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 {
- swarm_id: msg.swarm_id.clone(),
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- auth: ByteBuf::from(msg.auth.clone()),
- ready: msg.ready,
- run_id: ByteBuf::from(msg.run_id.to_vec()),
- })
-}
-pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
- let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
- Ok(HelloRequest {
- swarm_id: shadow.swarm_id,
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- auth: shadow.auth.into_vec(),
- ready: shadow.ready,
- run_id: run_id_from_shadow(&shadow.run_id)?,
- })
-}
-
-fn run_id_from_shadow(bytes: &ByteBuf) -> Result<crate::transport::RunId, TransportError> {
- bytes
- .as_slice()
- .try_into()
- .map_err(|_| TransportError::InvalidPayload("run id must be 16 bytes".into()))
-}
-
-// --- 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, msg.epoch_seq),
- auth: ByteBuf::from(msg.auth.clone()),
- ready: msg.ready,
- run_id: ByteBuf::from(msg.run_id.to_vec()),
- away_ms: msg.away_ms,
- sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()),
- };
- 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)?,
- epoch_seq: shadow.epoch.seq,
- auth: shadow.auth.into_vec(),
- ready: shadow.ready,
- run_id: run_id_from_shadow(&shadow.run_id)?,
- away_ms: shadow.away_ms,
- sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?,
- })
-}
-
-// --- 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, msg.epoch_seq),
- 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)?,
- epoch_seq: shadow.epoch.seq,
- message: shadow.message,
- ready: shadow.ready,
- })
-}
-
-// --- BlobMessage ---
-#[derive(Serialize, Deserialize)]
-enum BlobIdShadow {
- Change(ByteBuf),
- Content(ByteBuf),
-}
-
-#[derive(Serialize, Deserialize)]
-enum BlobBodyShadow {
- Request {
- chunks: Vec<u32>,
- },
- Manifest {
- size: u64,
- chunk_size: u32,
- hashes: Vec<ByteBuf>,
- },
- Chunk {
- index: u32,
- data: ByteBuf,
- },
- Missing,
-}
-
-#[derive(Serialize, Deserialize)]
-struct BlobShadow {
- epoch: EpochShadow,
- id: BlobIdShadow,
- body: BlobBodyShadow,
-}
-
-fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- let id = match msg.id {
- BlobId::Change(hash) => BlobIdShadow::Change(ByteBuf::from(hash.0.to_vec())),
- BlobId::Content(hash) => BlobIdShadow::Content(ByteBuf::from(hash.to_vec())),
- };
- let body = match &msg.body {
- BlobBody::Request { chunks } => BlobBodyShadow::Request {
- chunks: chunks.clone(),
- },
- BlobBody::Manifest {
- size,
- chunk_size,
- hashes,
- } => BlobBodyShadow::Manifest {
- size: *size,
- chunk_size: *chunk_size,
- hashes: hashes
- .iter()
- .map(|hash| ByteBuf::from(hash.to_vec()))
- .collect(),
- },
- BlobBody::Chunk { index, data } => BlobBodyShadow::Chunk {
- index: *index,
- data: ByteBuf::from(data.to_vec()),
- },
- BlobBody::Missing => BlobBodyShadow::Missing,
- };
- rmp_serde::to_vec(&BlobShadow {
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- id,
- body,
- })
-}
-
-fn hash32(bytes: &[u8], what: &str) -> Result<[u8; 32], TransportError> {
- bytes
- .try_into()
- .map_err(|_| TransportError::InvalidPayload(format!("{what} must be 32 bytes")))
-}
-
-fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> {
- let shadow: BlobShadow = rmp_serde::from_slice(bytes)?;
- let id = match shadow.id {
- BlobIdShadow::Change(hash) => BlobId::Change(ChangeHash(hash32(&hash, "a change hash")?)),
- BlobIdShadow::Content(hash) => BlobId::Content(hash32(&hash, "a content hash")?),
- };
- let body = match shadow.body {
- BlobBodyShadow::Request { chunks } => BlobBody::Request { chunks },
- BlobBodyShadow::Manifest {
- size,
- chunk_size,
- hashes,
- } => BlobBody::Manifest {
- size,
- chunk_size,
- hashes: hashes
- .iter()
- .map(|hash| hash32(hash, "a chunk hash"))
- .collect::<Result<_, _>>()?,
- },
- BlobBodyShadow::Chunk { index, data } => BlobBody::Chunk {
- index,
- data: data.into_vec().into(),
- },
- BlobBodyShadow::Missing => BlobBody::Missing,
- };
- Ok(BlobMessage {
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- id,
- body,
- })
-}
-
-#[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,
- },
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- /// 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 = keepalive_to_bytes(&keepalive).unwrap();
- // Mycelium's message data: a length byte, the topic, then our payload.
- let size = 1 + TOPIC_PONG.len().max(TOPIC_PING.len()) + payload.len();
- assert!(
- size + 50 <= MYCELIUM_CHUNK,
- "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
- );
- assert_eq!(bytes_to_keepalive(&payload).unwrap().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 chunk = BlobMessage {
- epoch,
- epoch_seq: 4,
- id,
- body: BlobBody::Chunk {
- index: 5,
- data: vec![1u8; 64 * 1024].into(),
- },
- };
- let bytes = blob_to_bytes(&chunk).unwrap();
- // A 64 KiB chunk costs only a few bytes of framing.
- assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len());
- let decoded = bytes_to_blob(&bytes).unwrap();
- 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 = BlobMessage {
- epoch,
- epoch_seq: 4,
- id,
- body: BlobBody::Manifest {
- size: 100,
- chunk_size: 64,
- hashes: vec![[1; 32], [2; 32]],
- },
- };
- let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap();
- assert!(matches!(
- decoded.body,
- BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2
- ));
- }
-
- #[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()));
- }
-}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 06:59:12 +0000