Add broker-gated iroh transport for cmux-tui

cmux-tui/Cargo.toml

@@ -29,6 +29,7 @@ members = [
     "crates/cmux-relay",
     "crates/cmux-tui-machine-agent-protocol",
     "crates/cmux-tui-machine-protocol",
+    "crates/cmux-tui-iroh",
     "crates/cmux-tui",
 ]
 default-members = [
@@ -44,6 +45,7 @@ default-members = [
     "crates/cmux-relay",
     "crates/cmux-tui-machine-agent-protocol",
     "crates/cmux-tui-machine-protocol",
+    "crates/cmux-tui-iroh",
     "crates/cmux-tui",
 ]
 exclude = ["vendor/crossterm"]
@@ -74,6 +76,7 @@ cmux-remote-protocol = { path = "crates/cmux-remote-protocol" }
 cmux-remote = { path = "crates/cmux-remote" }
 cmux-tui-machine-agent-protocol = { path = "crates/cmux-tui-machine-agent-protocol" }
 cmux-tui-machine-protocol = { path = "crates/cmux-tui-machine-protocol" }
+cmux-tui-iroh = { path = "crates/cmux-tui-iroh" }
 anyhow = "1"
 async-trait = "0.1"
 axum = { version = "0.8", features = ["ws"] }
@@ -92,6 +95,7 @@ tungstenite = { version = "0.29", default-features = false, features = ["handsha
 uds_windows = "1.2"
 windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security", "Win32_System_Diagnostics_ToolHelp", "Win32_System_JobObjects", "Win32_System_Threading"] }
 regex = "1"
+reqwest = { version = "0.13", default-features = false, features = ["json", "query", "rustls"] }
 jsonschema = { version = "0.30", default-features = false }
 flate2 = "1"
 bindgen = "0.72"
@@ -100,6 +104,7 @@ getrandom = "0.3"
 glob = "0.3"
 bytes = "1"
 diffy = "0.4"
+ed25519-dalek = "2"
 fs4 = "1.1.0"
 terminput = "=0.5.11"
 terminput-crossterm = { version = "=0.4.7", default-features = false, features = ["crossterm_0_29"] }
@@ -120,12 +125,14 @@ socket2 = { version = "0.5", features = ["all"] }
 snow = "0.10"
 tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] }
 tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "handshake", "rustls-tls-native-roots"] }
+tokio-util = { version = "0.7", features = ["rt"] }
 tower = { version = "0.5", features = ["util"] }
 tower-http = { version = "0.6", features = ["timeout"] }
 url = "2"
 uuid = { version = "1", features = ["serde", "v4"] }
 x25519-dalek = { version = "2", features = ["static_secrets"] }
 zeroize = { version = "1", features = ["derive"] }
+time = { version = "0.3", features = ["formatting", "parsing"] }
 
 [patch.crates-io]
 # Crossterm 0.29 discards Kitty's shifted key, base-layout key, and associated

cmux-tui/crates/cmux-remote/src/identity.rs

@@ -1999,6 +1999,27 @@ fn atomic_json(path: &Path, value: &impl Serialize) -> Result<(), IdentityError>
     result
 }
 
+/// Atomically persists JSON in a managed owner-only directory.
+///
+/// This is the same durability path used by the remote identity database:
+/// create an owner-only temporary file, sync it, rename it, then sync the
+/// parent directory. Callers should use [`crate::secret_file::read_owner_only`]
+/// when loading the result so the read side retains the same ownership and
+/// symlink checks.
+pub fn write_owner_only_json(path: &Path, value: &impl Serialize) -> Result<(), IdentityError> {
+    atomic_json(path, value)
+}
+
+/// Loads bounded JSON through the hardened owner-only secret-file reader.
+pub fn read_owner_only_json<T: serde::de::DeserializeOwned>(
+    path: &Path,
+    maximum_bytes: usize,
+) -> Result<T, IdentityError> {
+    let bytes =
+        crate::secret_file::read_owner_only(path, maximum_bytes).map_err(IdentityError::Io)?;
+    serde_json::from_slice(&bytes).map_err(IdentityError::Json)
+}
+
 fn sync_parent_directory(path: &Path) -> Result<(), IdentityError> {
     #[cfg(unix)]
     File::open(path).and_then(|directory| directory.sync_all()).map_err(IdentityError::Io)?;

cmux-tui/crates/cmux-remote/src/lib.rs

@@ -24,7 +24,7 @@ mod mux_codec;
 mod mux_input;
 mod mux_lanes;
 pub mod observability;
-mod owner_lock;
+pub mod owner_lock;
 pub mod provider;
 pub mod secret_file;
 pub mod secure_directory;

cmux-tui/crates/cmux-remote/src/owner_lock.rs

@@ -5,24 +5,24 @@ use std::path::{Path, PathBuf};
 use fs4::FileExt;
 
 #[derive(Debug)]
-pub(crate) struct OwnerFileLock {
+pub struct OwnerFileLock {
     file: File,
 }
 
 impl OwnerFileLock {
-    pub(crate) fn acquire(path: &Path) -> io::Result<Self> {
+    pub fn acquire(path: &Path) -> io::Result<Self> {
         let file = open_private_lock(path)?;
         FileExt::lock(&file)?;
         Ok(Self { file })
     }
 
-    pub(crate) fn try_acquire(path: &Path) -> io::Result<Self> {
+    pub fn try_acquire(path: &Path) -> io::Result<Self> {
         let file = open_private_lock(path)?;
         FileExt::try_lock(&file).map_err(io::Error::from)?;
         Ok(Self { file })
     }
 
-    pub(crate) async fn acquire_async(path: PathBuf) -> io::Result<Self> {
+    pub async fn acquire_async(path: PathBuf) -> io::Result<Self> {
         tokio::task::spawn_blocking(move || Self::acquire(&path))
             .await
             .map_err(|error| io::Error::other(format!("owner-file lock task failed: {error}")))?
@@ -35,7 +35,7 @@ impl Drop for OwnerFileLock {
     }
 }
 
-pub(crate) fn sibling_lock_path(path: &Path) -> io::Result<PathBuf> {
+pub fn sibling_lock_path(path: &Path) -> io::Result<PathBuf> {
     let file_name = path.file_name().ok_or_else(|| {
         io::Error::new(io::ErrorKind::InvalidInput, "lock target has no file name")
     })?;

cmux-tui/crates/cmux-remote/src/provider/iroh.rs

@@ -335,16 +335,16 @@ impl IrohProviderConfig {
 }
 
 #[derive(Debug, Clone, Copy)]
-struct IrohListenerLimits {
-    maximum_connections: usize,
-    maximum_connection_overflow: usize,
-    maximum_pending_streams: usize,
-    maximum_pending_stream_overflow: usize,
-    maximum_pending_streams_per_connection: usize,
-    connection_handshake_timeout: Duration,
-    first_stream_timeout: Duration,
-    unauthenticated_timeout: Duration,
-    pre_auth_timeout: Duration,
+pub struct IrohListenerLimits {
+    pub maximum_connections: usize,
+    pub maximum_connection_overflow: usize,
+    pub maximum_pending_streams: usize,
+    pub maximum_pending_stream_overflow: usize,
+    pub maximum_pending_streams_per_connection: usize,
+    pub connection_handshake_timeout: Duration,
+    pub first_stream_timeout: Duration,
+    pub unauthenticated_timeout: Duration,
+    pub pre_auth_timeout: Duration,
 }
 
 impl Default for IrohListenerLimits {
@@ -364,7 +364,7 @@ impl Default for IrohListenerLimits {
 }
 
 impl IrohListenerLimits {
-    fn validate(self) -> Result<Self, ProviderError> {
+    pub fn validate(self) -> Result<Self, ProviderError> {
         if self.maximum_connections == 0
             || self.maximum_pending_streams == 0
             || self.maximum_pending_streams_per_connection == 0
@@ -381,7 +381,7 @@ impl IrohListenerLimits {
     }
 }
 
-struct IrohAdmission {
+pub struct IrohAdmission {
     limits: IrohListenerLimits,
     connections: Arc<Semaphore>,
     connection_overflow: Arc<Semaphore>,
@@ -390,7 +390,7 @@ struct IrohAdmission {
 }
 
 impl IrohAdmission {
-    fn new(limits: IrohListenerLimits) -> Self {
+    pub fn new(limits: IrohListenerLimits) -> Self {
         Self {
             limits,
             connections: Arc::new(Semaphore::new(limits.maximum_connections)),
@@ -401,14 +401,40 @@ impl IrohAdmission {
             )),
         }
     }
+
+    pub fn limits(&self) -> IrohListenerLimits {
+        self.limits
+    }
+
+    pub fn try_reserve_connection(&self) -> Option<IrohPreAuthAdmission> {
+        try_pre_auth_admission(&self.connections, &self.connection_overflow)
+    }
+
+    pub async fn acquire_connection(
+        &self,
+        reservation: IrohPreAuthAdmission,
+    ) -> OwnedSemaphorePermit {
+        reservation.acquire(self.connections.clone()).await
+    }
+
+    pub fn try_reserve_pending_stream(&self) -> Option<IrohPreAuthAdmission> {
+        try_pre_auth_admission(&self.pending_streams, &self.pending_stream_overflow)
+    }
+
+    pub async fn acquire_pending_stream(
+        &self,
+        reservation: IrohPreAuthAdmission,
+    ) -> OwnedSemaphorePermit {
+        reservation.acquire(self.pending_streams.clone()).await
+    }
 }
 
-enum PreAuthAdmission {
+pub enum IrohPreAuthAdmission {
     Ready(OwnedSemaphorePermit),
     Queued(OwnedSemaphorePermit),
 }
 
-impl PreAuthAdmission {
+impl IrohPreAuthAdmission {
     async fn acquire(self, capacity: Arc<Semaphore>) -> OwnedSemaphorePermit {
         match self {
             Self::Ready(permit) => permit,
@@ -455,10 +481,10 @@ impl PreAuthAdmission {
 fn try_pre_auth_admission(
     capacity: &Arc<Semaphore>,
     overflow: &Arc<Semaphore>,
-) -> Option<PreAuthAdmission> {
+) -> Option<IrohPreAuthAdmission> {
     match capacity.clone().try_acquire_owned() {
-        Ok(permit) => Some(PreAuthAdmission::Ready(permit)),
-        Err(_) => overflow.clone().try_acquire_owned().ok().map(PreAuthAdmission::Queued),
+        Ok(permit) => Some(IrohPreAuthAdmission::Ready(permit)),
+        Err(_) => overflow.clone().try_acquire_owned().ok().map(IrohPreAuthAdmission::Queued),
     }
 }
 
@@ -484,7 +510,7 @@ impl IrohProvider {
     }
 
     async fn endpoint(&self) -> Result<&Endpoint, ProviderError> {
-        self.endpoint.get_or_try_init(|| bind_endpoint(&self.config)).await
+        self.endpoint.get_or_try_init(|| bind_iroh_endpoint(&self.config)).await
     }
 
     pub async fn local_node_id(&self) -> Result<NodeId, ProviderError> {
@@ -531,7 +557,8 @@ fn validate_config(config: &IrohProviderConfig) -> Result<(), ProviderError> {
     Ok(())
 }
 
-async fn bind_endpoint(config: &IrohProviderConfig) -> Result<Endpoint, ProviderError> {
+pub async fn bind_iroh_endpoint(config: &IrohProviderConfig) -> Result<Endpoint, ProviderError> {
+    validate_config(config)?;
     use ::iroh::endpoint::presets;
 
     let builder = if config.discovery_n0 {
@@ -553,7 +580,7 @@ async fn bind_endpoint(config: &IrohProviderConfig) -> Result<Endpoint, Provider
         .map_err(|_| ProviderError::Transport("could not bind Iroh endpoint".into()))
 }
 
-async fn connect_iroh_connection(
+pub async fn connect_iroh_endpoint(
     endpoint: &Endpoint,
     node_addr: &NodeAddr,
     alpn: &[u8],
@@ -597,7 +624,7 @@ async fn connect_iroh_for_path_mode(
         && node_addr.ip_addrs().next().is_some()
         && dial_addr.relay_urls().next().is_some();
     if !relay_assisted {
-        return connect_iroh_connection(endpoint, &dial_addr, alpn).await;
+        return connect_iroh_endpoint(endpoint, &dial_addr, alpn).await;
     }
 
     // A relay-first handshake avoids letting an explicitly advertised but
@@ -606,12 +633,12 @@ async fn connect_iroh_for_path_mode(
     // without relay access still falls back to the complete address set.
     match tokio::time::timeout(
         AUTO_RELAY_BOOTSTRAP_TIMEOUT,
-        connect_iroh_connection(endpoint, &dial_addr, alpn),
+        connect_iroh_endpoint(endpoint, &dial_addr, alpn),
     )
     .await
     {
         Ok(Ok(connection)) => Ok(connection),
-        Ok(Err(_)) | Err(_) => connect_iroh_connection(endpoint, node_addr, alpn).await,
+        Ok(Err(_)) | Err(_) => connect_iroh_endpoint(endpoint, node_addr, alpn).await,
     }
 }
 
@@ -722,7 +749,7 @@ impl IrohListener {
         validate_config(&config)?;
         let admission = Arc::new(IrohAdmission::new(limits.validate()?));
         let relay_enabled = !matches!(&config.relay_mode, RelayMode::Disabled);
-        let endpoint = bind_endpoint(&config).await?;
+        let endpoint = bind_iroh_endpoint(&config).await?;
         let (shutdown_tx, shutdown_rx) = oneshot::channel();
         let task = tokio::spawn(run_iroh_listener(
             endpoint.clone(),
@@ -1245,10 +1272,10 @@ mod tests {
         listener: &IrohListener,
         secret_key: SecretKey,
     ) -> (Endpoint, ::iroh::endpoint::Connection) {
-        let endpoint = bind_endpoint(&local_config(secret_key)).await.unwrap();
+        let endpoint = bind_iroh_endpoint(&local_config(secret_key)).await.unwrap();
         let route = listener.route().await.unwrap();
         let connection =
-            connect_iroh_connection(&endpoint, route.node_addr(), CMUX_IROH_ALPN).await.unwrap();
+            connect_iroh_endpoint(&endpoint, route.node_addr(), CMUX_IROH_ALPN).await.unwrap();
         (endpoint, connection)
     }
 
@@ -1659,10 +1686,10 @@ mod tests {
             alpn: CMUX_IROH_ALPN.to_vec(),
             maximum_frame_bytes: 32,
         };
-        let endpoint = bind_endpoint(&config).await.unwrap();
+        let endpoint = bind_iroh_endpoint(&config).await.unwrap();
         let node_addr = NodeAddr::new(server_key.public()).with_ip_addr(direct);
         let connection =
-            connect_iroh_connection(&endpoint, &node_addr, CMUX_IROH_ALPN).await.unwrap();
+            connect_iroh_endpoint(&endpoint, &node_addr, CMUX_IROH_ALPN).await.unwrap();
         let observed_connection = connection.clone();
         let description = format!("iroh://{}", server_key.public());
         let transport = iroh_transport_snapshot(&description, &connection);
@@ -1763,11 +1790,11 @@ mod tests {
         let (first_client, first_connection) = connect_test_client(&listener, secret(34)).await;
         wait_for_available_permits(&listener.admission.connections, 0).await;
 
-        let second_client = bind_endpoint(&local_config(secret(35))).await.unwrap();
+        let second_client = bind_iroh_endpoint(&local_config(secret(35))).await.unwrap();
         let route = listener.route().await.unwrap();
         let second = tokio::time::timeout(
             Duration::from_secs(5),
-            connect_iroh_connection(&second_client, route.node_addr(), CMUX_IROH_ALPN),
+            connect_iroh_endpoint(&second_client, route.node_addr(), CMUX_IROH_ALPN),
         )
         .await
         .expect("excess Iroh connection should be refused promptly");

cmux-tui/crates/cmux-remote/src/provider/mod.rs

@@ -27,8 +27,10 @@ use crate::observability::TransportSnapshot;
 
 #[cfg(feature = "iroh-transport")]
 pub use iroh::{
-    CMUX_IROH_ALPN, IrohListener, IrohPathMode, IrohProvider, IrohProviderConfig, IrohRoute,
-    ROUTING_DIRECT_ADDRS, ROUTING_NODE_ID, ROUTING_RELAY_URL, load_or_create_iroh_secret,
+    CMUX_IROH_ALPN, IrohAdmission, IrohListener, IrohListenerLimits, IrohPathMode,
+    IrohPreAuthAdmission, IrohProvider, IrohProviderConfig, IrohRoute, ROUTING_DIRECT_ADDRS,
+    ROUTING_NODE_ID, ROUTING_RELAY_URL, bind_iroh_endpoint, connect_iroh_endpoint,
+    load_or_create_iroh_secret,
 };
 pub use relay::{
     RelayClientConfig, RelayCredentialSource, RelayDaemonConfig, RelayDaemonRegistration,

cmux-tui/crates/cmux-tui-iroh/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "cmux-tui-iroh"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+publish.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow.workspace = true
+base64.workspace = true
+cmux-remote.workspace = true
+cmux-tui-machine-protocol.workspace = true
+ed25519-dalek.workspace = true
+getrandom.workspace = true
+iroh.workspace = true
+libc.workspace = true
+reqwest.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+sha2.workspace = true
+subtle.workspace = true
+time.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true
+url.workspace = true
+uuid.workspace = true
+zeroize.workspace = true
+
+[dev-dependencies]
+tempfile = "3"