summaryrefslogtreecommitdiffstats
path: root/src/transport/mycelium
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/mycelium')
-rw-r--r--src/transport/mycelium/daemon.rs337
-rw-r--r--src/transport/mycelium/mod.rs552
2 files changed, 889 insertions, 0 deletions
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<Url>,
+ pub key_file_opt: Option<PathBuf>,
+ pub network_key_file: Option<PathBuf>,
+ pub network_name: String,
+ pub tcp_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,
+ 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.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
+ }
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct MyceliumNodeConfig {
+ pub tcp_port: u16,
+ pub api_port: u16,
+ pub rpc_port: 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>,
+ 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")?,
+ };
+ // 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<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");
+
+ // 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<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();
+ }
+}
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<MyceliumSetup>,
+ 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<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))])
+ .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))])
+ .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 = 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()));
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 01:41:11 +0000