split mycelium daemon in its own module

1 files changed, 371 insertions, 0 deletions

diff --git a/src/transport/daemon.rs b/src/transport/daemon.rs
new file mode 100644
index 0000000..eceeb4d
--- /dev/null
+++ b/src/transport/daemon.rs
@@ -0,0 +1,371 @@
+//! 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 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
+            }
+        }
+    }
+}
+
+#[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();
+    }
+}