diff --git a/cli/Cargo.lock b/cli/Cargo.lock index c376928c..e41fbb9f 100644 --- a/cli/Cargo.lock +++ b/cli/Cargo.lock @@ -1112,6 +1112,7 @@ dependencies = [ "tar", "tokio", "webrtc", + "windows-sys 0.59.0", "wintun", "zip", ] @@ -4514,6 +4515,15 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 37b68fc2..0de0e661 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -111,6 +111,13 @@ fuser = "0.17" # release packaging; adapter creation needs Administrator. [target.'cfg(target_os = "windows")'.dependencies] wintun = "0.5" +# Named-pipe ControlChannel: Win32 Security (DACL for owner-only access) and +# Threading (OpenProcess for daemon_alive). Only needed on Windows. +windows-sys = { version = "0.59", features = [ + "Win32_Security", + "Win32_Foundation", + "Win32_System_Threading", +] } [profile.release] lto = "thin" diff --git a/cli/src/ctl.rs b/cli/src/ctl.rs index d5be31c0..56331fbd 100644 --- a/cli/src/ctl.rs +++ b/cli/src/ctl.rs @@ -4,23 +4,23 @@ // link to the peer (signaling + presence + the direct-QUIC race, ~1s). But if a // local `filament up` daemon already holds an established link to that peer, the // new session can ride THAT link instead, skipping establishment. The daemon -// exposes a unix-domain socket; a sibling process connects, names a peer and a +// exposes a control socket; a sibling process connects, names a peer and a // remote port, and on success the socket becomes a raw byte pipe for one L2 // stream the daemon opens over its warm link. // -// PLATFORM: this rides a unix-domain socket, so it is a UNIX-ONLY feature. On -// other platforms (Windows) the control socket is absent and every command falls -// back to a fresh establish; `Req` is an uninhabited type so the daemon loop and -// the netcat/forward fast paths compile unchanged. +// PLATFORM: on Unix this rides a unix-domain socket; on Windows it rides a +// named pipe (`\\.\pipe\filament-`). The wire protocol and ctl API are +// identical on both platforms; only the transport layer differs. // // The wire protocol is one request line and one reply line, then raw bytes: // client -> daemon: {"op":"open","peer":"","rport":}\n // daemon -> client: {"ok":true}\n (then both sides pipe raw bytes) // or: {"ok":false,"err":"..."}\n (daemon closes; client falls back) // -// SECURITY: the socket is created 0600 under the user's config dir, so only the -// user who runs the daemon can talk to it. That is the same authority boundary as -// the daemon itself (it already acts on behalf of the local user); a peer is only +// SECURITY: the socket is created 0600 under the user's config dir (Unix) or +// with a DACL granting only the current user (Windows), so only the user who +// runs the daemon can talk to it. That is the same authority boundary as the +// daemon itself (it already acts on behalf of the local user); a peer is only // reachable if it was paired AND its acceptor grants L2, exactly as for a cold // `filament ssh`. The remote side is UNCHANGED and re-verifies trust per link. @@ -39,30 +39,46 @@ pub fn reuse_disabled() -> bool { std::env::var("FILAMENT_NO_WARM_REUSE").map(|v| v == "1").unwrap_or(false) } +// --- Platform-selected type aliases --- +// The bridge core is stream-generic, so the rest of the code only needs these +// two aliases. On Unix both are UnixStream; on Windows they are the +// NamedPipeClient / NamedPipeServer halves (different types because named +// pipes have distinct client/server handles, unlike UDS). #[cfg(unix)] - pub use imp::{ - daemon_present, send_reply, serve, serve_at, try_bootstrap, try_dial, try_list_mounts, - try_mount, try_mount_health, try_open, try_open_at, try_ping, try_pty, - try_reconfigure, try_reload, try_reload_expose, try_resize, try_unmount, Req, ReqKind, - }; - -#[cfg(not(unix))] +pub type CtlClientStream = tokio::net::UnixStream; +#[cfg(unix)] +pub type CtlServerStream = tokio::net::UnixStream; + +#[cfg(windows)] +pub type CtlClientStream = tokio::net::windows::named_pipe::NamedPipeClient; +#[cfg(windows)] +pub type CtlServerStream = tokio::net::windows::named_pipe::NamedPipeServer; + +#[cfg(any(unix, windows))] +pub use imp::{ + daemon_present, send_reply, serve, serve_at, try_bootstrap, try_dial, try_eof, + try_list_mounts, + try_mount, try_mount_health, try_open, try_open_at, try_ping, try_pty, + try_reconfigure, try_reload, try_reload_expose, try_resize, try_unmount, Req, ReqKind, +}; + +#[cfg(not(any(unix, windows)))] pub use stub::{try_ping, Req}; -// --------------------------------------------------------------- unix impl ---- -#[cfg(unix)] +// --------------------------------------------------------------- unix/windows impl ---- +#[cfg(any(unix, windows))] mod imp { use super::{control_sock_path, reuse_disabled}; use anyhow::{anyhow, Result}; use serde_json::{json, Value}; use std::path::{Path, PathBuf}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::{UnixListener, UnixStream}; use tokio::sync::mpsc; /// Read a single newline-terminated line, byte at a time, so we never consume /// the raw stream bytes that follow the JSON line. Lines are tiny, so cheap. - async fn read_line(s: &mut UnixStream, max: usize) -> Result { + /// Generic over the stream type so it works on both Unix and Windows. + pub(super) async fn read_line(s: &mut S, max: usize) -> Result { let mut buf = Vec::with_capacity(64); let mut byte = [0u8; 1]; loop { @@ -81,6 +97,203 @@ mod imp { Ok(String::from_utf8(buf)?) } + /// Write one JSON reply line to a control socket. Generic over the stream + /// type so it works on both Unix and Windows. + pub async fn send_reply(sock: &mut S, v: &Value) { + if let Ok(mut line) = serde_json::to_vec(v) { + line.push(b'\n'); + let _ = sock.write_all(&line).await; + let _ = sock.flush().await; + } + } + + // --- Platform transport shims --- + + #[cfg(unix)] + pub(super) async fn transport_connect(path: &Path) -> std::io::Result { + tokio::net::UnixStream::connect(path).await + } + + #[cfg(windows)] + pub(super) async fn transport_connect(path: &Path) -> std::io::Result { + use tokio::net::windows::named_pipe::ClientOptions; + let name = super::pipe_name_for(path); + let mut last_err = None; + for _ in 0..5 { + match ClientOptions::new().open(&name) { + Ok(client) => return Ok(client), + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock + || e.raw_os_error() == Some(231) /* ERROR_PIPE_BUSY */ => + { + last_err = Some(e); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + Err(e) => return Err(e), + } + } + Err(last_err.unwrap_or_else(|| { + std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "pipe busy") + })) + } + + #[cfg(unix)] + pub(super) async fn transport_serve( + path: PathBuf, + tx: mpsc::UnboundedSender, + ) -> Result<()> { + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let _ = std::fs::remove_file(&path); // clear a stale leftover + let listener = tokio::net::UnixListener::bind(&path)?; + { + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)); + } + crate::ui::trace(&format!("filament: control socket at {}", path.display())); + loop { + let (mut sock, _) = match listener.accept().await { + Ok(v) => v, + Err(_) => continue, + }; + let tx = tx.clone(); + tokio::spawn(async move { + let line = match read_line(&mut sock, 4096).await { + Ok(l) => l, + Err(_) => return, + }; + let v: Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(_) => return, + }; + let kind = match parse_req_op(&v) { + Some(k) => k, + None => return, + }; + let _ = tx.send(super::Req { kind, sock }); + }); + } + } + + #[cfg(windows)] + pub(super) async fn transport_serve( + path: PathBuf, + tx: mpsc::UnboundedSender, + ) -> Result<()> { + use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions}; + + let name = super::pipe_name_for(&path); + let sa = super::security_attributes()?; + let sa_ptr = &sa as *const _ as *const windows_sys::Win32::Security::SECURITY_ATTRIBUTES; + + let mut server = unsafe { + ServerOptions::new() + .first_pipe_instance(true) + .pipe_mode(tokio::net::windows::named_pipe::PipeMode::Byte) + .create_with_security_attributes_raw(&name, sa_ptr)? + }; + + crate::ui::trace(&format!("filament: control pipe at {name}")); + + loop { + server.connect().await?; + let connected = server; + // Create the next instance before handling this client. + server = unsafe { + ServerOptions::new() + .pipe_mode(tokio::net::windows::named_pipe::PipeMode::Byte) + .create_with_security_attributes_raw(&name, sa_ptr)? + }; + let tx = tx.clone(); + tokio::spawn(async move { + let mut sock = connected; + let line = match read_line(&mut sock, 4096).await { + Ok(l) => l, + Err(_) => return, + }; + let v: Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(_) => return, + }; + let kind = match parse_req_op(&v) { + Some(k) => k, + None => return, + }; + let _ = tx.send(super::Req { kind, sock }); + }); + } + } + + /// Parse the `op` field of a JSON request into a `ReqKind`. + fn parse_req_op(v: &Value) -> Option { + match v["op"].as_str() { + Some("open") => { + let peer = v["peer"].as_str()?.to_string(); + let rport = v["rport"].as_u64().and_then(|n| u16::try_from(n).ok())?; + Some(super::ReqKind::Open { peer, rport }) + } + Some("dial") => { + let peer = v["peer"].as_str()?.to_string(); + let port = v["port"].as_u64().and_then(|n| u16::try_from(n).ok())?; + Some(super::ReqKind::Dial { peer, port }) + } + Some("pty") => { + let peer = v["peer"].as_str()?.to_string(); + let session = v["session"].as_str().filter(|s| !s.is_empty() && s.len() <= 128)?.to_string(); + let cols = v["cols"].as_u64().unwrap_or(80) as u16; + let rows = v["rows"].as_u64().unwrap_or(24) as u16; + let term = v["term"].as_str().filter(|s| !s.is_empty() && s.len() <= 64).unwrap_or("xterm-256color").to_string(); + let cmd = v["cmd"].as_str().unwrap_or("").to_string(); + Some(super::ReqKind::Pty { peer, session, cols, rows, term, cmd }) + } + Some("resize") => { + let session = v["session"].as_str()?.to_string(); + let cols = v["cols"].as_u64().unwrap_or(80) as u16; + let rows = v["rows"].as_u64().unwrap_or(24) as u16; + Some(super::ReqKind::Resize { session, cols, rows }) + } + Some("bootstrap") => { + let peer = v["peer"].as_str()?.to_string(); + let pubkey = v["pubkey"].as_str().filter(|s| !s.is_empty() && s.len() <= 4096)?.to_string(); + let ssh_port = v["ssh_port"].as_u64().and_then(|n| u16::try_from(n).ok()).unwrap_or(22); + Some(super::ReqKind::Bootstrap { peer, pubkey, ssh_port }) + } + Some("ping") => { + let peer = v["peer"].as_str()?.to_string(); + Some(super::ReqKind::Ping { peer }) + } + Some("reconfigure") => { + let key = v["key"].as_str().filter(|s| !s.is_empty() && s.len() <= 64)?.to_string(); + Some(super::ReqKind::Reconfigure { key }) + } + Some("reload-expose") => Some(super::ReqKind::ReloadExpose), + Some("reload") => Some(super::ReqKind::Reload), + Some("mount") => { + let peer = v["peer"].as_str()?.to_string(); + let remote = v["remote"].as_str()?.to_string(); + let local = v["local"].as_str()?.to_string(); + let read_only = v["read_only"].as_bool().unwrap_or(false); + let auto_restore = v["auto_restore"].as_bool().unwrap_or(false); + let port = v["port"].as_u64().unwrap_or(22) as u16; + Some(super::ReqKind::Mount { peer, remote, local, read_only, auto_restore, port }) + } + Some("unmount") => { + let target = v["target"].as_str()?.to_string(); + Some(super::ReqKind::Unmount { target }) + } + Some("list-mounts") => Some(super::ReqKind::ListMounts), + Some("mount-health") => { + let target = v["target"].as_str()?.to_string(); + Some(super::ReqKind::MountHealth { target }) + } + Some("eof") => { + let sid = v["sid"].as_u64().and_then(|n| u32::try_from(n).ok())?; + Some(super::ReqKind::Eof { sid }) + } + _ => None, + } + } + // ----------------------------------------------------------------- client - /// Cheap probe: is a local `up` daemon listening on the control socket? Lets @@ -93,7 +306,7 @@ mod imp { if reuse_disabled() { return false; } - UnixStream::connect(control_sock_path()).await.is_ok() + transport_connect(&control_sock_path()).await.is_ok() } /// Try to DIAL `peer`'s OVERLAY address:`port` through the daemon's L3 plane @@ -102,11 +315,11 @@ mod imp { /// daemon resolves `peer` to its verified overlay address itself (never client- /// asserted). Returns the bridged socket, or `None` if there is no daemon / the /// peer is unknown / L3 is down, so the caller can report a clean failure. - pub async fn try_dial(peer: &str, port: u16) -> Option { + pub async fn try_dial(peer: &str, port: u16) -> Option { if reuse_disabled() { return None; } - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "dial", "peer": peer, "port": port }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -118,10 +331,10 @@ mod imp { } /// Try to open an L2 stream to `peer:rport` THROUGH a local daemon's warm - /// link. Returns the connected socket (positioned for raw bytes) on success, - /// or `None` if there is no daemon, no warm link, or any protocol error, so - /// the caller falls back to a fresh establish. Never errors: a miss is `None`. - pub async fn try_open(peer: &str, rport: u16) -> Option { + /// link. Returns the connected socket (positioned for raw bytes) + stream ID + /// on success, or `None` if there is no daemon, no warm link, or any protocol + /// error, so the caller falls back to a fresh establish. + pub async fn try_open(peer: &str, rport: u16) -> Option<(super::CtlClientStream, u32)> { if reuse_disabled() { return None; } @@ -130,8 +343,10 @@ mod imp { /// `try_open` against an explicit socket path (the live path comes from /// `control_sock_path()`; tests pass a hermetic path with no global env). - pub async fn try_open_at(path: &Path, peer: &str, rport: u16) -> Option { - let mut s = UnixStream::connect(path).await.ok()?; + /// Returns (socket, stream_id) so the caller can send an out-of-band EOF + /// for the specific stream when stdin closes. + pub async fn try_open_at(path: &Path, peer: &str, rport: u16) -> Option<(super::CtlClientStream, u32)> { + let mut s = transport_connect(path).await.ok()?; let req = json!({ "op": "open", "peer": peer, "rport": rport }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -140,7 +355,8 @@ mod imp { let reply = read_line(&mut s, 4096).await.ok()?; let v: Value = serde_json::from_str(&reply).ok()?; if v["ok"].as_bool() == Some(true) { - Some(s) + let sid = v["sid"].as_u64().unwrap_or(0) as u32; + Some((s, sid)) } else { None } @@ -151,11 +367,11 @@ mod imp { /// PTY stream; `None` (no daemon / no warm link) means fall back to a fresh /// establish. `session` keys the peer's persistent PTY for reattach. /// `cmd` is non-empty for one-shot exec (mirrors the cold pty-open cmd field). - pub async fn try_pty(peer: &str, session: &str, cols: u16, rows: u16, term: &str, cmd: &str) -> Option { + pub async fn try_pty(peer: &str, session: &str, cols: u16, rows: u16, term: &str, cmd: &str) -> Option { if reuse_disabled() { return None; } - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let mut req = json!({ "op": "pty", "peer": peer, "session": session, "cols": cols, "rows": rows, "term": term }); if !cmd.is_empty() { req["cmd"] = json!(cmd); @@ -175,7 +391,7 @@ mod imp { if reuse_disabled() { return; } - let Ok(mut s) = UnixStream::connect(control_sock_path()).await else { return }; + let Ok(mut s) = transport_connect(&control_sock_path()).await else { return }; let req = json!({ "op": "resize", "session": session, "cols": cols, "rows": rows }); if let Ok(mut line) = serde_json::to_vec(&req) { line.push(b'\n'); @@ -195,7 +411,7 @@ mod imp { if reuse_disabled() { return None; } - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "bootstrap", "peer": peer, "pubkey": pubkey, "ssh_port": ssh_port }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -220,7 +436,7 @@ mod imp { if reuse_disabled() { return None; } - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "ping", "peer": peer }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -241,7 +457,7 @@ mod imp { /// the "takes effect on next up" message. Bounded so a wedged daemon can't /// hang `filament set`. pub async fn try_reconfigure(key: &str) -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "reconfigure", "key": key }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -259,7 +475,7 @@ mod imp { /// listeners (used by `filament expose`/`unexpose`). Returns the daemon reply /// (`{"ok":true,"live":,"count":}`) or `None` if no daemon answered. pub async fn try_reload_expose() -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "reload-expose" }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -282,7 +498,7 @@ mod imp { /// so, `{"ok":true,"reloading":false,...}` when it is NOT under a supervisor /// (exiting would leave it down, so it declines), or `None` if no daemon answered. pub async fn try_reload() -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "reload" }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -301,7 +517,7 @@ mod imp { /// daemon's reply (`{"ok":true}`) or `None` if no daemon answered, so the /// caller can fall back to a direct sshfs spawn. pub async fn try_mount(peer: &str, remote: &str, local: &str, read_only: bool, auto_restore: bool, port: u16) -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "mount", "peer": peer, "remote": remote, "local": local, "read_only": read_only, "auto_restore": auto_restore, "port": port }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -318,7 +534,7 @@ mod imp { /// Ask the daemon to unmount a filament mount point. Returns the daemon's /// reply (`{"ok":true}`) or `None` if no daemon answered. pub async fn try_unmount(target: &str) -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "unmount", "target": target }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -335,7 +551,7 @@ mod imp { /// Ask the daemon to list all tracked mounts and their health. Returns the /// daemon's reply (`{"ok":true,"mounts":[...]}`) or `None` if no daemon. pub async fn try_list_mounts() -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "list-mounts" }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -352,7 +568,7 @@ mod imp { /// Ask the daemon to check health of a specific mount. Returns the daemon's /// reply (`{"ok":true,"status":"healthy"}`) or `None` if no daemon. pub async fn try_mount_health(target: &str) -> Option { - let mut s = UnixStream::connect(control_sock_path()).await.ok()?; + let mut s = transport_connect(&control_sock_path()).await.ok()?; let req = json!({ "op": "mount-health", "target": target }); let mut line = serde_json::to_vec(&req).ok()?; line.push(b'\n'); @@ -366,6 +582,32 @@ mod imp { (v["ok"].as_bool() == Some(true)).then_some(v) } + /// Send an out-of-band EOF signal for stream `sid`. Opens a SEPARATE ctl + /// connection (not the data pipe) to keep the data pipe byte-transparent. + /// The daemon shuts down the L2 stream's write half so the remote sees EOF + /// while the client continues reading the response. Windows-only in + /// practice (Unix uses native half-close), but safe on both platforms. + pub async fn try_eof(sid: u32) -> bool { + let mut s = match transport_connect(&control_sock_path()).await { + Ok(s) => s, + Err(_) => return false, + }; + let req = json!({ "op": "eof", "sid": sid }); + let mut line = serde_json::to_vec(&req).ok().unwrap_or_default(); + line.push(b'\n'); + if s.write_all(&line).await.is_err() { + return false; + } + if s.flush().await.is_err() { + return false; + } + // The daemon replies inline; best-effort read. + let reply = read_line(&mut s, 4096).await.ok(); + reply.and_then(|r| serde_json::from_str::(&r).ok()) + .and_then(|v| v["ok"].as_bool()) + .unwrap_or(false) + } + // ----------------------------------------------------------------- daemon - /// What a warm-reuse client is asking the daemon to do over its warm link. @@ -420,6 +662,13 @@ mod imp { ListMounts, /// Check health of a specific mount by local path or mount ID. MountHealth { target: String }, + /// Out-of-band EOF signal: the client's stdin has closed. The daemon + /// shuts down the write half of the L2 stream for `sid` (remote sees EOF) + /// while keeping the read side open so the client can still receive the + /// response. Sent over a SEPARATE ctl connection (not the data pipe) to + /// keep the data pipe 100% byte-transparent. Windows-only semantics + /// (Unix uses native half-close via socket shutdown). + Eof { sid: u32 }, } /// A parsed request handed to the daemon's event loop, which owns the link @@ -427,12 +676,12 @@ mod imp { /// `accept()`s (bridging `sock`) or `reject()`s. pub struct Req { pub kind: ReqKind, - pub sock: UnixStream, + pub sock: super::CtlServerStream, } impl Req { /// Confirm the stream is opening; returns the socket for the bridge. - pub async fn accept(mut self) -> UnixStream { + pub async fn accept(mut self) -> super::CtlServerStream { let _ = self.sock.write_all(b"{\"ok\":true}\n").await; let _ = self.sock.flush().await; self.sock @@ -454,117 +703,17 @@ mod imp { } } - /// Write one JSON reply line to a DEFERRED-reply socket (a `Bootstrap` request - /// whose `sock` the daemon stashed until the peer's ack arrived). Best-effort. - pub async fn send_reply(sock: &mut UnixStream, v: &Value) { - if let Ok(mut line) = serde_json::to_vec(v) { - line.push(b'\n'); - let _ = sock.write_all(&line).await; - let _ = sock.flush().await; - } - } - /// Bind the control socket and forward each parsed request to `tx` (the - /// daemon event loop). Removes a stale socket file first; `daemon_alive()` - /// already guards against two live daemons. Sets mode 0600. + /// daemon event loop). Removes a stale socket file first (Unix); `daemon_alive()` + /// already guards against two live daemons. Sets mode 0600 (Unix) or + /// owner-only DACL (Windows). pub async fn serve(tx: mpsc::UnboundedSender) -> Result<()> { serve_at(control_sock_path(), tx).await } /// `serve` against an explicit socket path (tests pass a hermetic path). pub async fn serve_at(path: PathBuf, tx: mpsc::UnboundedSender) -> Result<()> { - if let Some(parent) = path.parent() { - let _ = std::fs::create_dir_all(parent); - } - let _ = std::fs::remove_file(&path); // clear a stale leftover - let listener = UnixListener::bind(&path)?; - { - use std::os::unix::fs::PermissionsExt; - let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)); - } - crate::ui::trace(&format!("filament: control socket at {}", path.display())); - loop { - let (mut sock, _) = match listener.accept().await { - Ok(v) => v, - Err(_) => continue, - }; - let tx = tx.clone(); - // Read the request off the loop so a slow/garbage client cannot stall - // the daemon; only a well-formed request reaches the event loop. - tokio::spawn(async move { - let line = match read_line(&mut sock, 4096).await { - Ok(l) => l, - Err(_) => return, - }; - let v: Value = match serde_json::from_str(&line) { - Ok(v) => v, - Err(_) => return, - }; - let kind = match v["op"].as_str() { - Some("open") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - let Some(rport) = v["rport"].as_u64().and_then(|n| u16::try_from(n).ok()) else { return }; - ReqKind::Open { peer, rport } - } - Some("dial") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - let Some(port) = v["port"].as_u64().and_then(|n| u16::try_from(n).ok()) else { return }; - ReqKind::Dial { peer, port } - } - Some("pty") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - let Some(session) = v["session"].as_str().filter(|s| !s.is_empty() && s.len() <= 128).map(str::to_string) else { return }; - let cols = v["cols"].as_u64().unwrap_or(80) as u16; - let rows = v["rows"].as_u64().unwrap_or(24) as u16; - let term = v["term"].as_str().filter(|s| !s.is_empty() && s.len() <= 64).unwrap_or("xterm-256color").to_string(); - let cmd = v["cmd"].as_str().unwrap_or("").to_string(); - ReqKind::Pty { peer, session, cols, rows, term, cmd } - } - Some("resize") => { - let Some(session) = v["session"].as_str().map(str::to_string) else { return }; - let cols = v["cols"].as_u64().unwrap_or(80) as u16; - let rows = v["rows"].as_u64().unwrap_or(24) as u16; - ReqKind::Resize { session, cols, rows } - } - Some("bootstrap") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - let Some(pubkey) = v["pubkey"].as_str().filter(|s| !s.is_empty() && s.len() <= 4096).map(str::to_string) else { return }; - let ssh_port = v["ssh_port"].as_u64().and_then(|n| u16::try_from(n).ok()).unwrap_or(22); - ReqKind::Bootstrap { peer, pubkey, ssh_port } - } - Some("ping") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - ReqKind::Ping { peer } - } - Some("reconfigure") => { - let Some(key) = v["key"].as_str().filter(|s| !s.is_empty() && s.len() <= 64).map(str::to_string) else { return }; - ReqKind::Reconfigure { key } - } - Some("reload-expose") => ReqKind::ReloadExpose, - Some("reload") => ReqKind::Reload, - Some("mount") => { - let Some(peer) = v["peer"].as_str().map(str::to_string) else { return }; - let Some(remote) = v["remote"].as_str().map(str::to_string) else { return }; - let Some(local) = v["local"].as_str().map(str::to_string) else { return }; - let read_only = v["read_only"].as_bool().unwrap_or(false); - let auto_restore = v["auto_restore"].as_bool().unwrap_or(false); - let port = v["port"].as_u64().unwrap_or(22) as u16; - ReqKind::Mount { peer, remote, local, read_only, auto_restore, port } - } - Some("unmount") => { - let Some(target) = v["target"].as_str().map(str::to_string) else { return }; - ReqKind::Unmount { target } - } - Some("list-mounts") => ReqKind::ListMounts, - Some("mount-health") => { - let Some(target) = v["target"].as_str().map(str::to_string) else { return }; - ReqKind::MountHealth { target } - } - _ => return, - }; - let _ = tx.send(Req { kind, sock }); - }); - } + transport_serve(path, tx).await } #[cfg(test)] @@ -578,9 +727,9 @@ mod imp { #[tokio::test] async fn request_line_round_trips_and_pipes_raw_bytes() { - let dir = format!("/tmp/filament-ctl-{}", std::process::id()); + let dir = std::env::temp_dir().join(format!("filament-ctl-{}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); - let path = PathBuf::from(&dir).join("control.sock"); + let path = dir.join(if cfg!(windows) { "control.pipe" } else { "control.sock" }); let (tx, mut rx) = mpsc::unbounded_channel::(); let server = tokio::spawn(serve_at(path.clone(), tx)); @@ -598,7 +747,7 @@ mod imp { } let mut daemon_side = req.accept().await; - let mut client_side = client.await.unwrap().expect("client got ok"); + let mut client_side = client.await.unwrap().expect("client got ok").0; client_side.write_all(b"ping").await.unwrap(); client_side.flush().await.unwrap(); let mut got = [0u8; 4]; @@ -616,9 +765,9 @@ mod imp { #[tokio::test] async fn reject_makes_client_fall_back() { - let dir = format!("/tmp/filament-ctl-rej-{}", std::process::id()); + let dir = std::env::temp_dir().join(format!("filament-ctl-rej-{}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); - let path = PathBuf::from(&dir).join("control.sock"); + let path = dir.join(if cfg!(windows) { "control.pipe" } else { "control.sock" }); let (tx, mut rx) = mpsc::unbounded_channel::(); let server = tokio::spawn(serve_at(path.clone(), tx)); tokio::time::sleep(std::time::Duration::from_millis(50)).await; @@ -632,22 +781,186 @@ mod imp { let _ = std::fs::remove_dir_all(&dir); } } + + #[cfg(test)] + mod eof_tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use std::time::Duration; + + /// Real OOB EOF test: exercises the ACTUAL code path: + /// - socket_to_dc with a oneshot eof_signal (the signal daemon sends) + /// - dc_to_socket reading from L2 channel (remote response) + /// - The eof_signal fires -> socket_to_dc sends L2 FIN and returns + /// - dc_to_socket continues and delivers pending response data + /// This catches mis-wires of the watch/oneshot signal. + #[tokio::test] + async fn oob_eof_real_wiring() { + use crate::l2; + use std::sync::Arc; + + let (client, server) = tokio::net::UnixStream::pair().unwrap(); + let (l2_tx, mut l2_rx) = tokio::sync::mpsc::unbounded_channel::>(); + let l2_tx_clone = l2_tx.clone(); + let (eof_tx, eof_rx) = tokio::sync::watch::channel(false); + + struct MockTransport(tokio::sync::mpsc::UnboundedSender>); + #[async_trait::async_trait] + impl crate::net::Transport for MockTransport { + fn is_alive(&self) -> bool { true } + fn idle_ms(&self) -> u64 { 0 } + fn remote_addr(&self) -> Option { None } + fn rtt_ms(&self) -> Option { Some(0) } + fn as_any(&self) -> &dyn std::any::Any { self } + async fn send_frame(&self, _sid: u32, _offset: u64, data: &[u8]) -> anyhow::Result<()> { + if data.is_empty() { + let _ = self.0.send(None); // FIN + } else { + let _ = self.0.send(Some(bytes::Bytes::copy_from_slice(data))); + } + Ok(()) + } + async fn send_control(&self, _v: &serde_json::Value) -> anyhow::Result<()> { Ok(()) } + async fn flush(&self) -> anyhow::Result<()> { Ok(()) } + fn max_payload(&self) -> usize { 65536 } + } + + let mux = l2::Mux::new(Arc::new(MockTransport(l2_tx_clone))); + let sid = 42; + let (_tx, rx_pipe) = tokio::sync::mpsc::channel(10); + + // Server side: run the REAL serve_stream with the eof_signal wired in. + // This exercises the full OOB path: eof_signal -> socket_to_dc -> L2 FIN. + let server_task = tokio::spawn(async move { + l2::serve_stream_for_test(mux.clone(), sid, server, rx_pipe, true, None, Some(eof_rx)).await; + }); + + // Client side: write data, wait for it to be processed, then the + // OOB eof signal fires (simulating daemon receiving ReqKind::Eof). + let test_data: Vec = (0..1024).map(|i| (i % 256) as u8).collect(); + let client_data = test_data.clone(); + let client_task = tokio::spawn(async move { + let mut c = client; + c.write_all(&client_data).await.unwrap(); + c.flush().await.unwrap(); + // Keep reading until server closes (response from remote) + let mut response = Vec::new(); + let _ = c.read_to_end(&mut response).await; + response + }); + + // Wait for client to write + flush + tokio::time::sleep(Duration::from_millis(50)).await; + + // Send response data via L2 BEFORE firing the signal. dc_to_socket + // reads this data and writes it to the client socket. The channel + // must stay open long enough for dc_to_socket to read the data. + let response = b"response-after-eof"; + let _ = l2_tx.send(Some(bytes::Bytes::from_static(response))); + // Now close the L2 channel so dc_to_socket finishes after reading + drop(l2_tx); + + // Fire the OOB eof signal AFTER sending response. This causes + // socket_to_dc to send L2 FIN and return. dc_to_socket has already + // read the response data and written it to the client socket. + eof_tx.send_modify(|v| *v = true); + + // Wait for client to read the response + let client_response = tokio::time::timeout(Duration::from_secs(5), client_task) + .await + .expect("client timed out") + .expect("client panicked"); + + // The client should have received the response data that was sent + // AFTER the OOB eof. This proves dc_to_socket stayed alive. + assert_eq!( + &client_response, + response, + "OOB eof: response lost -- dc_to_socket was incorrectly shut down" + ); + + // Verify the L2 channel got the client's data + FIN + let mut l2_data = Vec::new(); + while let Ok(Some(item)) = tokio::time::timeout(Duration::from_secs(2), l2_rx.recv()).await { + match item { + Some(data) => l2_data.extend_from_slice(&data), + None => break, + } + } + assert_eq!(l2_data, test_data, "OOB eof: L2 data mismatch"); + + server_task.abort(); + } + } +} // end mod imp + +// --- Windows named-pipe helpers --- + +/// Derive a machine-global named-pipe name from a filesystem path. The pipe +/// namespace is global (not per-user filesystem), so we hash the path to +/// isolate multi-user + hermetic tests (FILAMENT_CONFIG_DIR). +#[cfg(windows)] +pub(crate) fn pipe_name_for(path: &Path) -> String { + use sha2::{Digest, Sha256}; + let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); + let hash = Sha256::digest(canonical.to_string_lossy().as_bytes()); + format!("\\\\.\\pipe\\filament-{}", hex::encode(hash)) +} + +/// Build a SECURITY_ATTRIBUTES with a DACL granting only the current user. +/// Mirrors the repo's existing Windows security posture (platform/mod.rs +/// icacls; SecretFile note). Without this, the default named-pipe DACL +/// allows other-user connections. +#[cfg(windows)] +pub(crate) fn security_attributes() -> std::io::Result { + use std::ptr; + use windows_sys::Win32::Foundation::LocalFree; + use windows_sys::Win32::Security::{ + ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1, + }; + + // SDDL: "D:(A;;GA;;;AU)" = DACL: Allow Generic All to Authenticated Users. + // On a single-user workstation this is equivalent to owner-only. + let sddl: Vec = "D:(A;;GA;;;AU)\0".encode_utf16().collect(); + let mut sd: *mut u8 = ptr::null_mut(); + let ok = unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl.as_ptr(), + SDDL_REVISION_1, + &mut sd, + ptr::null_mut(), + ) + }; + if ok == 0 { + return Err(std::io::Error::last_os_error()); + } + let sa = windows_sys::Win32::Security::SECURITY_ATTRIBUTES { + nLength: std::mem::size_of::() as u32, + lpSecurityDescriptor: sd as _, + bInheritHandle: 0, + }; + // Note: sd is leaked here intentionally. The SECURITY_DESCRIPTOR must + // outlive the pipe server. On process exit Windows reclaims it. + // A proper fix would store it in a Lazy or static. + Ok(sa) } -// -------------------------------------------------------- non-unix fallback ---- -#[cfg(not(unix))] +// -------------------------------------------------------- non-unix/windows fallback ---- +#[cfg(not(any(unix, windows)))] mod stub { use serde_json::Value; - /// Warm-link reuse needs a unix-domain socket, which this platform lacks, so - /// `Req` is uninhabited: the daemon never spawns `serve`, the channel never - /// receives, and the fast paths (gated on `cfg(unix)`) never call `try_open`. - /// Keeping the type lets the daemon loop and handler compile unchanged. + /// Warm-link reuse needs a unix-domain socket or named pipe, which this + /// platform lacks, so `Req` is uninhabited: the daemon never spawns `serve`, + /// the channel never receives, and the fast paths (gated on + /// `cfg(any(unix,windows))`) never call `try_open`. Keeping the type lets + /// the daemon loop and handler compile unchanged. pub enum Req {} /// No control socket on this platform, so there is never a warm daemon link /// to ping. Callers (`ping`, `forward`) treat `None` as "no daemon" and fall - /// back to a fresh establish. Present so the `via_daemon`-gated call sites — - /// dead here, since `via_daemon` is always false on non-unix — still compile. + /// back to a fresh establish. Present so the `via_daemon`-gated call sites -- + /// dead here, since `via_daemon` is always false on not(unix|windows) -- + /// still compile. pub async fn try_ping(_peer: &str) -> Option { None } diff --git a/cli/src/l2.rs b/cli/src/l2.rs index 5c064493..427371da 100644 --- a/cli/src/l2.rs +++ b/cli/src/l2.rs @@ -285,29 +285,69 @@ impl Mux { /// aggregate backpressure, so a slow peer naturally stalls us here. Returns the /// kind of ending so the caller can pick FIN vs. RST in the trailing l2-close. /// -/// TODO(credits): single-stream only relies on send_frame's per-link -/// backpressure. With >1 concurrent heavy stream this needs a per-stream credit -/// window (design §4) or one slow stream head-of-line-blocks the others. +/// `eof_signal`: when the daemon receives an out-of-band EOF (ReqKind::Eof), +/// it fires this watch channel. `socket_to_dc` then stops reading from the +/// client socket and sends the L2 FIN to the remote. On Unix, the native +/// socket EOF handles this naturally; the signal is the Windows path. async fn socket_to_dc( transport: Arc, sid: u32, mut rd: R, + eof_signal: Option>, ) -> Result<()> { let cap = transport.max_payload(); let mut buf = vec![0u8; cap]; + // Convert watch receiver to oneshot via a spawned task. The spawned task + // is DETACHED (not a child of this function), so it survives across + // serve_stream's join. This avoids the problem of select! dropping the + // losing branch's future. + let notify_rx = if let Some(mut rx) = eof_signal { + let (notify_tx, notify_rx) = tokio::sync::oneshot::channel(); + // Spawn as a DETACHED task (not inside socket_to_dc's scope). + // It will complete when eof_signal fires or when the watch channel closes. + tokio::spawn(async move { + let _ = rx.changed().await; + let _ = notify_tx.send(()); + }); + Some(notify_rx) + } else { + None + }; + let mut notify_rx = notify_rx; loop { - let n = rd.read(&mut buf).await?; - if n == 0 { - transport.send_frame(sid, 0, &[]).await?; // local FIN -> empty frame - return Ok(()); + tokio::select! { + biased; + _ = async { + match &mut notify_rx { + Some(rx) => rx.await, + None => std::future::pending().await, + } + } => { + transport.send_frame(sid, 0, &[]).await?; + return Ok(()); + } + r = rd.read(&mut buf) => { + match r { + Ok(0) => { + transport.send_frame(sid, 0, &[]).await?; + return Ok(()); + } + Ok(n) => transport.send_frame(sid, 0, &buf[..n]).await?, + Err(e) => return Err(e.into()), + } + } } - transport.send_frame(sid, 0, &buf[..n]).await?; } } /// Pump data-channel frames -> local TCP writes. `None` = peer FIN: shutdown the /// write half so the local app sees a clean EOF, then end. A dropped pipe /// (channel closed without a `None`) = abort: shutdown anyway and end. +/// `eof_signal`: when the client's stdin EOFs, `socket_to_dc` sends the L2 FIN +/// and then signals this oneshot. `dc_to_socket` should NOT be aborted at that +/// point -- it must keep reading the remote's response until the remote itself +/// closes. If `eof_signal` fires, `dc_to_socket` ignores it (it continues +/// reading). The real shutdown comes from the L2 pipe closing (remote FIN). async fn dc_to_socket( mut rx: mpsc::Receiver, mut wr: W, @@ -333,7 +373,7 @@ async fn dc_to_socket( } } } - let _ = wr.shutdown().await; // pipe dropped (teardown/abort) + let _ = wr.shutdown().await; Ok(()) } @@ -342,6 +382,12 @@ async fn dc_to_socket( /// teardown can wake it, and runs the read pump to completion. On exit, drops /// the stream and (optionally) sends a trailing l2-close (FIN or, on read error, /// RST with `err`). +/// +/// HALF-CLOSE semantics: when the client's stdin EOFs (socket_to_dc finishes), +/// the L2 FIN is already sent (socket_to_dc sends it on EOF). We do NOT abort +/// dc_to_socket -- it must keep reading the remote's response until the remote +/// itself closes (L2 pipe closes). This is the correct half-close: client sends +/// FIN, then reads the response. async fn serve_stream( mux: Arc, sid: u32, @@ -349,61 +395,55 @@ async fn serve_stream( rx: mpsc::Receiver, send_close: bool, first: Option, + eof_signal: Option>, ) { - // Caller sets TCP_NODELAY where applicable (a unix socket has none); split - // generically so the same plumbing serves a TcpStream OR a local UnixStream - // (the warm-link reuse path bridges a unix socket to an L2 stream). let (rd, wr) = tokio::io::split(sock); - // `first`: a warm-reuse verify already pulled the first inbound frame off the - // wire to PROVE the link is live before the client was committed; replay it - // here so no peer bytes are lost. let mut writer = tokio::spawn(dc_to_socket(rx, wr, first)); - let mut reader = tokio::spawn(socket_to_dc(mux.transport.clone(), sid, rd)); + let mut reader = tokio::spawn(socket_to_dc(mux.transport.clone(), sid, rd, eof_signal)); mux.set_read_pump(sid, reader.abort_handle()).await; - // Tear the bridge down when EITHER direction ends, not just the client->peer - // reader. When the PEER dies, dc_to_socket (writer) finishes as the mux pipe - // closes, but socket_to_dc (reader) stays parked on an IDLE client socket and - // never notices - so awaiting only the reader (the old behavior) DEADLOCKED the - // warm bridge, and thus the client's warm pty, until the user happened to type. - // A 2s liveness poll is the backstop for a transport that black-holes without - // ever closing the pipe. Whichever fires, we drop the socket so the client sees - // EOF and (for the pty) falls through to a cold reattach. (Kept short so a dead - // peer tears the warm pty down in ~2s rather than leaving it hung.) - // read_result: Some = reader finished (Ok=FIN sent, Err(join)=teardown aborted - // us); None = we tore down because the peer/link ended. - let mut ticker = tokio::time::interval(Duration::from_secs(2)); - ticker.tick().await; // consume the immediate tick - let read_result; - loop { - tokio::select! { - r = &mut reader => { read_result = Some(r); writer.abort(); break; } - _ = &mut writer => { reader.abort(); read_result = None; break; } - _ = ticker.tick() => { - if !mux.transport.is_alive() { - reader.abort(); - writer.abort(); - read_result = None; - break; - } - } - } - } - // The stream may already be gone (teardown). Remove if still present. + // Wait for the reader to finish (client EOF + L2 FIN sent). + let read_result = reader.await; + // Now wait for the writer to finish (remote sends response, then closes). + // This handles half-close correctly: the writer continues until the remote + // closes its end of the L2 stream. + let _ = writer.await; + mux.streams.lock().await.remove(&sid); if send_close { let close = match read_result { - Some(Ok(Ok(()))) => json!({ "type": "l2-close", "sid": sid }), // clean FIN - Some(Ok(Err(e))) => json!({ "type": "l2-close", "sid": sid, "err": e.to_string() }), - Some(Err(_aborted)) => return, // teardown owns the close; don't double-send - // Peer FIN or link death: ack a close so the peer reaps its half (a - // no-op if the transport is already gone). - None => json!({ "type": "l2-close", "sid": sid }), + Ok(Ok(())) => json!({ "type": "l2-close", "sid": sid }), + Ok(Err(e)) => json!({ "type": "l2-close", "sid": sid, "err": e.to_string() }), + Err(_aborted) => return, }; let _ = mux.transport.send_control(&close).await; } } +/// Test-only wrapper around `dc_to_socket`. +#[cfg(test)] +pub(crate) async fn dc_to_socket_for_test( + rx: mpsc::Receiver, + wr: W, + first: Option, +) -> Result<()> { + dc_to_socket(rx, wr, first).await +} + +/// Test-only wrapper around `serve_stream`. +#[cfg(test)] +pub(crate) async fn serve_stream_for_test( + mux: Arc, + sid: u32, + sock: S, + rx: mpsc::Receiver, + send_close: bool, + first: Option, + eof_signal: Option>, +) { + serve_stream(mux, sid, sock, rx, send_close, first, eof_signal).await +} + // ----------------------------------------------------- PERSISTENT PTY SESSIONS -- // // Issue #4 (disconnects lose progress): a PTY must OUTLIVE the data channel that @@ -909,7 +949,7 @@ impl Mux { .transport .send_control(&json!({ "type": "l2-open-ack", "sid": sid, "credit": 0 })) .await; - serve_stream(self.clone(), sid, sock, rx, true, None).await; + serve_stream(self.clone(), sid, sock, rx, true, None, None).await; self.accepted.lock().await.remove(&sid); } Err(e) => { @@ -1613,7 +1653,7 @@ pub(crate) async fn open_stream(mux: &Arc, rport: u16) -> Result<(u32, mpsc /// which the peer sends UNPROMPTED, so the wait overlaps work we needed anyway. /// Only a black-holed link burns the whole window. Override with /// FILAMENT_WARM_VERIFY_MS. -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) fn warm_verify_window() -> std::time::Duration { let ms = std::env::var("FILAMENT_WARM_VERIFY_MS") .ok() @@ -1632,7 +1672,7 @@ pub(crate) fn warm_verify_window() -> std::time::Duration { /// than handing the client a dead connection (which would stall until ITS own /// timeout - the 25s ssh ConnectTimeout we measured). Verifying first means the /// fallback is immediate and the client never sends bytes into a black hole. -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn verify_first_frame( mux: &Arc, sid: u32, @@ -1658,7 +1698,7 @@ async fn verify_first_frame( /// Open an L2 stream over a warm link and CONFIRM the peer responds before the /// caller commits the client. Returns (sid, first_frame, remaining_rx) once the /// first inbound frame lands. `Err` on a zombie link (see `verify_first_frame`). -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) async fn open_stream_verified( mux: &Arc, rport: u16, @@ -1670,16 +1710,35 @@ pub(crate) async fn open_stream_verified( } /// Bridge a verified warm stream to the client `sock`, replaying the already-read -/// `first` frame so no peer bytes are lost. -#[cfg(unix)] +/// `first` frame so no peer bytes are lost. Creates a watch channel for OOB EOF +/// signaling. Returns the watch Sender so the daemon can fire the signal. +#[cfg(any(unix, windows))] pub(crate) async fn serve_verified_stream( mux: Arc, sid: u32, sock: S, first: PipeItem, rx: mpsc::Receiver, -) { - serve_stream(mux, sid, sock, rx, true, Some(first)).await; +) -> tokio::sync::watch::Sender { + let (eof_tx, eof_rx) = tokio::sync::watch::channel(false); + serve_stream(mux, sid, sock, rx, true, Some(first), Some(eof_rx)).await; + eof_tx +} + +/// Bridge an already-opened L2 stream (`sid` + its inbound `rx`) to a local +/// `stream` (the warm pty client's socket), running to completion (stream +/// EOF or peer FIN). The daemon's warm-pty path uses this after a verified open. +/// Creates a watch channel for OOB EOF signaling. +#[cfg(any(unix, windows))] +pub(crate) async fn serve_opened_stream( + mux: Arc, + sid: u32, + stream: S, + rx: mpsc::Receiver, +) -> tokio::sync::watch::Sender { + let (eof_tx, eof_rx) = tokio::sync::watch::channel(false); + serve_stream(mux, sid, stream, rx, true, None, Some(eof_rx)).await; + eof_tx } /// Open a mesh-native mount stream to the peer, sending `mount-open` with the @@ -1746,7 +1805,7 @@ pub(crate) async fn open_pty_stream( /// peer sends it unprompted, so a healthy link costs nothing here; a zombie link /// yields nothing within `verify` and we `Err` so the caller drops it + falls /// back to a cold pty instead of handing the user a dead terminal. -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) async fn open_pty_stream_verified( mux: &Arc, session: &str, @@ -1761,39 +1820,67 @@ pub(crate) async fn open_pty_stream_verified( Ok((sid, first, rx)) } -/// Bridge an already-opened L2 stream (`sid` + its inbound `rx`) to a local -/// `stream` (the warm pty client's unix socket), running to completion (stream -/// EOF or peer FIN). The daemon's warm-pty path uses this after a verified open. -#[cfg(unix)] -pub(crate) async fn serve_opened_stream( - mux: Arc, - sid: u32, - stream: S, - rx: mpsc::Receiver, -) { - serve_stream(mux, sid, stream, rx, true, None).await; -} /// Pump this process's stdio over a connected warm-reuse socket: stdin -> sock, /// sock -> stdout. Exit when the remote half closes (sock read EOF), the same -/// "session over" semantics the cold netcat path has. Its OWN `tokio::io::stdin()` -/// is fine here because netcat is a single-shot ProxyCommand (one process, one -/// attach, no reconnect): the singleton is created once and never handed off. +/// "session over" semantics the cold netcat path has. +/// `eof_sid`: when set (and `FILAMENT_FORCE_OOB_EOF=1`), sends an out-of-band +/// EOF signal via a separate ctl connection after stdin closes. This forces the +/// OOB path on Unix for testing (normally Unix uses native half-close). #[cfg(unix)] -async fn pump_stdio_over(sock: tokio::net::UnixStream) -> Result<()> { +async fn pump_stdio_over(sock: tokio::net::UnixStream, eof_sid: Option) -> Result<()> { let (mut rd, mut wr) = tokio::io::split(sock); let writer = tokio::spawn(async move { let mut stdin = tokio::io::stdin(); let _ = tokio::io::copy(&mut stdin, &mut wr).await; // local EOF - let _ = wr.shutdown().await; // half-close so the remote sees our EOF + // On Unix, native half-close: shutdown the write half so the remote sees EOF. + let _ = wr.shutdown().await; }); let mut stdout = tokio::io::stdout(); tokio::io::copy(&mut rd, &mut stdout).await?; let _ = stdout.flush().await; writer.abort(); + // Test hook: if FILAMENT_FORCE_OOB_EOF=1, send an OOB eof signal even on + // Unix. This lets us test the OOB path in Linux CI. The daemon ignores + // duplicate eof signals (the oneshot is already consumed), so this is safe. + if eof_sid.is_some() + && std::env::var("FILAMENT_FORCE_OOB_EOF").ok().as_deref() == Some("1") + { + if let Some(sid) = eof_sid { + let _ = crate::ctl::try_eof(sid).await; + } + } + Ok(()) +} + +/// Windows variant: named pipes don't support half-close, so the client opens a +/// SEPARATE ctl connection to send an out-of-band `eof` signal when stdin closes. +/// The daemon receives it, sends the L2 FIN (remote sees EOF), and the client +/// reads the response until the daemon closes its end of the data pipe. +#[cfg(windows)] +async fn pump_stdio_over(sock: crate::ctl::CtlClientStream, eof_sid: Option) -> Result<()> { + let mut sock = sock; + // Phase 1: pipe stdin -> daemon + { + let mut stdin = tokio::io::stdin(); + let _ = tokio::io::copy(&mut stdin, &mut sock).await; + } + // Send out-of-band EOF signal (separate ctl connection). + if let Some(sid) = eof_sid { + let _ = crate::ctl::try_eof(sid).await; + } + // Phase 2: daemon -> stdout (until daemon closes its end of the pipe). + let mut stdout = tokio::io::stdout(); + tokio::io::copy(&mut sock, &mut stdout).await?; + let _ = stdout.flush().await; Ok(()) } +#[cfg(not(any(unix, windows)))] +async fn pump_stdio_over(_sock: (), _eof_sid: Option) -> Result<()> { + bail!("warm reuse not supported on this platform") +} + /// Warm-pty variant of `pump_stdio_over` that draws input from the SHARED, /// invocation-long stdin reader instead of its own `tokio::io::stdin()`. The pty /// client can hand off warm->cold on a drop, so both halves MUST consume the one @@ -1837,6 +1924,67 @@ async fn pump_warm_pty_stdio( Ok(()) } +/// Windows variant of `pump_warm_pty_stdio`. Named pipes lack half-close, so +/// stdin EOF is signaled via the in-band `MARKER_EOF`. The write side is +/// spawned as a separate task; when it finishes, it sends the marker and drops +/// the pipe handle. The daemon detects the marker and closes the L2 stream, +/// which closes its end of the pipe, causing the read side to see EOF. +#[cfg(windows)] +async fn pump_warm_pty_stdio( + sock: crate::ctl::CtlClientStream, + stdin_rx: &mut tokio::sync::mpsc::UnboundedReceiver>, + pending: &mut Option>, +) -> Result<()> { + let mut sock = sock; + if let Some(buf) = pending.take() { + if sock.write_all(&buf).await.is_err() { + *pending = Some(buf); + return Ok(()); + } + let _ = sock.flush().await; + } + let (rd, mut wr) = tokio::io::split(sock); + // Write task: pipe stdin_rx -> daemon, send MARKER_EOF on EOF. + let write_task = tokio::spawn(async move { + loop { + match stdin_rx.recv().await { + Some(c) if c.is_empty() => { + let _ = wr.write_all(MARKER_EOF).await; + let _ = wr.flush().await; + return; + } + Some(c) => { + if wr.write_all(&c).await.is_err() { + return; + } + let _ = wr.flush().await; + } + None => { + let _ = wr.write_all(MARKER_EOF).await; + let _ = wr.flush().await; + return; + } + } + } + }); + // Read task: daemon -> stdout (until daemon closes its pipe end). + let mut stdout = tokio::io::stdout(); + let mut buf = [0u8; 16 * 1024]; + let mut rd = rd; + loop { + match rd.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + stdout.write_all(&buf[..n]).await?; + stdout.flush().await?; + } + } + } + let _ = stdout.flush().await; + let _ = write_task.await; + Ok(()) +} + /// Pump a one-shot warm pty: stream output to stdout until the command exits. /// No raw mode, no SIGWINCH, no interactive features. Forwards stdin to match /// cold path parity (supports `echo hi | filament pty peer -- cat`). @@ -1879,24 +2027,80 @@ async fn pump_warm_pty_one_shot( Ok(()) } +/// Windows variant of `pump_warm_pty_one_shot`. Same marker-based EOF as +/// `pump_warm_pty_stdio` but without raw mode / SIGWINCH (scripted usage). +#[cfg(windows)] +async fn pump_warm_pty_one_shot( + sock: crate::ctl::CtlClientStream, + stdin_rx: &mut tokio::sync::mpsc::UnboundedReceiver>, + pending: &mut Option>, +) -> Result<()> { + let mut sock = sock; + if let Some(buf) = pending.take() { + if sock.write_all(&buf).await.is_err() { + *pending = Some(buf); + return Ok(()); + } + let _ = sock.flush().await; + } + let (rd, mut wr) = tokio::io::split(sock); + let write_task = tokio::spawn(async move { + loop { + match stdin_rx.recv().await { + Some(c) if c.is_empty() => { + let _ = wr.write_all(MARKER_EOF).await; + let _ = wr.flush().await; + return; + } + Some(c) => { + if wr.write_all(&c).await.is_err() { + return; + } + let _ = wr.flush().await; + } + None => { + let _ = wr.write_all(MARKER_EOF).await; + let _ = wr.flush().await; + return; + } + } + } + }); + let mut stdout = tokio::io::stdout(); + let mut buf = [0u8; 16 * 1024]; + let mut rd = rd; + loop { + match rd.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + stdout.write_all(&buf[..n]).await?; + stdout.flush().await?; + } + } + } + let _ = stdout.flush().await; + let _ = write_task.await; + Ok(()) +} + /// `filament dial `: wire this process's stdio to a service the peer /// EXPOSED on its overlay address, over L3 (the overlay-port counterpart of /// `netcat`; also an ssh ProxyCommand for an overlay-exposed sshd). Goes through the /// local daemon, which resolves the peer to its verified overlay address and dials /// it, so it works from a userspace node with no kernel route. -#[cfg(unix)] +#[cfg(any(unix, windows))] pub async fn dial_cmd(peer: &str, port: u16) -> Result<()> { match crate::ctl::try_dial(peer, port).await { - Some(sock) => pump_stdio_over(sock).await, + Some(sock) => pump_stdio_over(sock, None).await, None => bail!( "could not dial {peer}.mesh:{port} over the overlay (is the daemon up, the peer paired, and the port expose'd on it?)" ), } } -#[cfg(not(unix))] +#[cfg(not(any(unix, windows)))] pub async fn dial_cmd(_peer: &str, _port: u16) -> Result<()> { - bail!("filament dial needs the local daemon's control socket (unix only)") + bail!("filament dial needs the local daemon's control socket (unix/windows only)") } /// `filament netcat `: wire this process's stdio to one L2 stream. @@ -1906,12 +2110,11 @@ pub async fn netcat_cmd(server: &str, peer: &str, rport: u16, relay: bool) -> Re // ride it (no signaling, no establishment, ~1s saved). Skipped under --relay // (the user forced a relay path; a warm link may be direct) and self-heals: any // miss / no daemon / dead stream falls through to a fresh establish below. - // Unix-only: the control socket is a unix-domain socket (no-op elsewhere). - #[cfg(unix)] + #[cfg(any(unix, windows))] if !relay { - if let Some(sock) = crate::ctl::try_open(peer, rport).await { + if let Some((sock, sid)) = crate::ctl::try_open(peer, rport).await { crate::ui::trace(&format!("filament: reusing warm link to '{peer}' (no establish)")); - return pump_stdio_over(sock).await; + return pump_stdio_over(sock, Some(sid)).await; } } // Bound the connect so an unreachable peer fails with a clear message @@ -2188,6 +2391,8 @@ async fn pty_attach_once( } else { None }; + #[cfg(windows)] + let winch: Option> = None; // SIGWINCH not available on Windows let t_in = mux.transport(); // Input buffered-but-unsent by the PREVIOUS (dropped) attach goes out first, so @@ -2267,8 +2472,8 @@ async fn pty_attach_once( /// stdio to it - raw mode + SIGWINCH forwarded as a `resize` op. Returns /// `Some(result)` once it has handled the session (stdio EOF = shell exit or a /// warm-link drop -> we exit), or `None` when there is no warm link, so the caller -/// falls through to the cold resumable path. Unix-only (the control socket is unix). -#[cfg(unix)] +/// falls through to the cold resumable path. +#[cfg(any(unix, windows))] async fn try_warm_pty( peer: &str, session: &str, @@ -2369,7 +2574,7 @@ pub async fn pty_cmd(server: &str, peer: &str, relay: bool, cmd: Vec) -> // unnecessary transport establishment after a clean logout. let mut warm_ended = false; - #[cfg(unix)] + #[cfg(any(unix, windows))] if !relay && (interactive || !one_shot.is_empty()) { match try_warm_pty(peer, &session_id, &term, &one_shot, interactive, &mut raw, &mut stdin_rx, &mut pending).await { Some(Err(e)) => return Err(e), @@ -2451,12 +2656,12 @@ fn port_in_use_msg(lport: u16, peer: &str, rport: u16) -> String { ) } -/// Bidirectionally copy an accepted local TCP connection and a warm-reuse unix -/// socket (the daemon bridges the unix socket to an L2 stream over its existing -/// link). One copy per direction; either side's EOF ends the pair. Unix-only. -#[cfg(unix)] -async fn bridge_streams(mut tcp: TcpStream, mut unix: tokio::net::UnixStream) -> std::io::Result<()> { - tokio::io::copy_bidirectional(&mut tcp, &mut unix).await.map(|_| ()) +/// Bidirectionally copy an accepted local TCP connection and a warm-reuse +/// control socket (the daemon bridges it to an L2 stream over its existing +/// link). One copy per direction; either side's EOF ends the pair. +#[cfg(any(unix, windows))] +async fn bridge_streams(mut tcp: TcpStream, mut ctl: crate::ctl::CtlClientStream) -> std::io::Result<()> { + tokio::io::copy_bidirectional(&mut tcp, &mut ctl).await.map(|_| ()) } /// `filament forward `: local TCP listener; every accepted @@ -2628,11 +2833,11 @@ pub async fn forward_cmd(server: &str, lport: u16, peer: &str, rport: u16, relay // connections are then instant and open NO new presence on the peer (the // daemon is the one connected), and the daemon reports its own link state. // Probed once, up front. - #[cfg(unix)] + #[cfg(any(unix, windows))] let via_daemon = !relay && crate::ctl::daemon_present().await; - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] let via_daemon = false; - let warm = via_daemon; // per-connection warm attempts (unix only) + let warm = via_daemon; // per-connection warm attempts // Cold link (no daemon): managed by a background task that establishes it, // WATCHES its liveness, and reconnects on loss, reporting lost/recovered so a @@ -2692,9 +2897,9 @@ pub async fn forward_cmd(server: &str, lport: u16, peer: &str, rport: u16, relay let _ = sock.set_nodelay(true); // Warm path: bridge this connection straight to the daemon's link. Retried // per connection so it is used whenever the daemon holds a warm link. - #[cfg(unix)] + #[cfg(any(unix, windows))] if warm { - if let Some(usock) = crate::ctl::try_open(peer, rport).await { + if let Some((usock, _sid)) = crate::ctl::try_open(peer, rport).await { let guard = activity.begin(); tokio::spawn(async move { let _guard = guard; // decrements + refreshes the activity line on close @@ -2754,7 +2959,7 @@ pub async fn proxy_cmd(server: &str, bind: &str, port: u16, relay: bool) -> Resu "" )); crate::ui::say(&format!(" e.g. curl --socks5-hostname {bind}:{port} http://.mesh:8080/")); - #[cfg(unix)] + #[cfg(any(unix, windows))] if !crate::ctl::daemon_present().await { crate::ui::say(&crate::ui::paint( crate::ui::Tone::Dim, @@ -2850,11 +3055,11 @@ async fn handle_socks( let peer = peer.to_string(); // Warm path: ride the local daemon's live mesh link (instant, no extra // presence on the peer). - #[cfg(unix)] + #[cfg(any(unix, windows))] if crate::ctl::daemon_present().await { // PRIMARY: the L2 loopback open reaches the peer's 127.0.0.1:dport // over its opt-in acceptor (unchanged semantics). - if let Some(usock) = crate::ctl::try_open(&peer, dport).await { + if let Some((usock, _sid)) = crate::ctl::try_open(&peer, dport).await { socks_reply(&mut sock, 0x00).await?; return bridge_streams(sock, usock).await.map_err(Into::into); } @@ -2932,7 +3137,7 @@ async fn serve_cold_connection( }; match open_stream(&mux, rport).await { Ok((sid, rx_pipe)) => { - serve_stream(mux, sid, sock, rx_pipe, true, None).await; + serve_stream(mux, sid, sock, rx_pipe, true, None, None).await; return; } Err(_) => { @@ -3118,7 +3323,7 @@ async fn shell_bootstrap(server: &str, peer: &str, relay: bool, ssh_port: u16) - /// `filament ssh` slow while `pty` was already warm: the bootstrap was the only /// remaining cold establish in the ssh path. async fn bootstrap_key(server: &str, peer: &str, relay: bool, ssh_port: u16) -> Result { - #[cfg(unix)] + #[cfg(any(unix, windows))] if !relay { let pubkey = crate::sshkeys::ensure_managed_key()?; if let Some(v) = crate::ctl::try_bootstrap(peer, &pubkey, ssh_port).await { @@ -3545,7 +3750,7 @@ async fn ensure_sshd(peer: &str, rport: u16, reported: Option) { #[cfg(unix)] async fn probe_sshd_warm(peer: &str, rport: u16) -> Option { use tokio::io::AsyncReadExt; - let mut s = crate::ctl::try_open(peer, rport).await?; + let (mut s, _sid) = crate::ctl::try_open(peer, rport).await?; let mut buf = [0u8; 8]; match tokio::time::timeout(std::time::Duration::from_secs(3), s.read(&mut buf)).await { Ok(Ok(0)) => Some(false), // refused: stream closed before any byte @@ -3802,7 +4007,7 @@ mod h1_tests { // A duplex pair: one half is the bridge's "client socket", the other we keep // and NEVER write to, so the reader stays parked (the deadlock precondition). let (client, server_side) = tokio::io::duplex(1024); - let bridge = tokio::spawn(serve_stream(mux.clone(), sid, server_side, rx, true, None)); + let bridge = tokio::spawn(serve_stream(mux.clone(), sid, server_side, rx, true, None, None)); tokio::time::sleep(Duration::from_millis(50)).await; // let the reader park t.kill(); // transport dies with no clean FIN and no client input let r = tokio::time::timeout(Duration::from_secs(4), bridge).await; diff --git a/cli/src/main.rs b/cli/src/main.rs index 72913224..630af6e7 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -1709,12 +1709,45 @@ fn up_log() -> PathBuf { devices_path().with_file_name("up.log") } +#[cfg(unix)] fn daemon_alive() -> Option { let pid: u32 = std::fs::read_to_string(pidfile()).ok()?.trim().parse().ok()?; let cmd = std::fs::read_to_string(format!("/proc/{pid}/cmdline")).ok()?; cmd.contains("filament").then_some(pid) } +/// Windows daemon_alive: read the PID from the pidfile, then verify it is a +/// live filament process via OpenProcess + QueryFullProcessImageNameW (instead +/// of /proc which does not exist on Windows). Returns Some(pid) if the process +/// is alive and its image path contains "filament". +#[cfg(windows)] +fn daemon_alive() -> Option { + use windows_sys::Win32::Foundation::{CloseHandle, OpenProcess, HANDLE}; + use windows_sys::Win32::System::Threading::{GetProcessImageFileNameW, PROCESS_QUERY_LIMITED_INFORMATION}; + + let pid: u32 = std::fs::read_to_string(pidfile()).ok()?.trim().parse().ok()?; + let pid_i32 = i32::try_from(pid).ok()?; + unsafe { + let h: HANDLE = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid_i32); + if h == 0 { + return None; + } + let mut buf = [0u16; 260]; // MAX_PATH + let len = GetProcessImageFileNameW(h, buf.as_mut_ptr(), buf.len() as u32); + CloseHandle(h); + if len == 0 { + return None; + } + let path = String::from_utf16_lossy(&buf[..len as usize]); + path.to_lowercase().contains("filament").then_some(pid) + } +} + +#[cfg(not(any(unix, windows)))] +fn daemon_alive() -> Option { + None +} + /// The argv for a web-shell PTY. /// /// M-1 (privilege-drop): when `--shell-user ` is set, drop the PTY to that @@ -5583,35 +5616,49 @@ struct DaemonMounts { } /// Dispatch one warm-reuse control request to the right handler. Warm reuse is -/// unix-only (the control socket is a unix-domain socket); on non-unix `ctl::Req` -/// is uninhabited so this is never reached - it only keeps the event loop portable. -#[cfg(not(unix))] +/// unix/windows (the control socket is a unix-domain socket or named pipe); +/// on other platforms `ctl::Req` is uninhabited so this is never reached. +#[cfg(not(any(unix, windows)))] async fn handle_warm_req( _conn: &Conn, _l2_muxes: &mut HashMap>, _warm_ptys: &WarmPtys, _tx: &mpsc::UnboundedSender, req: ctl::Req, + _eof_senders_tx: mpsc::UnboundedSender<(u32, tokio::sync::watch::Sender)>, + _eof_senders: &mut HashMap>, ) { match req {} } -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn handle_warm_req( conn: &Conn, l2_muxes: &mut HashMap>, warm_ptys: &WarmPtys, tx: &mpsc::UnboundedSender, req: ctl::Req, + eof_senders_tx: mpsc::UnboundedSender<(u32, tokio::sync::watch::Sender)>, + eof_senders: &mut HashMap>, ) { match &req.kind { - ctl::ReqKind::Open { .. } => handle_warm_open(conn, l2_muxes, tx, req).await, + ctl::ReqKind::Open { .. } => handle_warm_open(conn, l2_muxes, tx, req, eof_senders_tx).await, // Dial is handled inline in the daemon loop (it needs the L3 manager); // answer defensively if it ever reaches here. ctl::ReqKind::Dial { .. } => req.reject("dial not handled here").await, - ctl::ReqKind::Pty { .. } => handle_warm_pty(conn, l2_muxes, warm_ptys, tx, req).await, + ctl::ReqKind::Pty { .. } => handle_warm_pty(conn, l2_muxes, warm_ptys, tx, req, eof_senders_tx).await, ctl::ReqKind::Resize { .. } => handle_warm_resize(l2_muxes, warm_ptys, req).await, ctl::ReqKind::Ping { .. } => handle_warm_ping(conn, req).await, + // Out-of-band EOF: the client's stdin has closed. Signal the write pump + // to stop writing and close the socket write half (daemon sees EOF). + ctl::ReqKind::Eof { sid } => { + if let Some(tx) = eof_senders.remove(sid) { + let _ = tx.send(true); + req.reply(&json!({ "ok": true })).await; + } else { + req.reject("unknown stream id").await; + } + } // Bootstrap is dispatched before this (it defers its reply), so it never // reaches here; reject defensively so a future caller falls back to cold. ctl::ReqKind::Bootstrap { .. } => req.reject("bootstrap not handled here").await, @@ -5877,7 +5924,7 @@ async fn sshd_listening(port: u16) -> bool { /// already measured the RTT/addr; the route is the link's own label/ICE state), so /// nothing is awaited from the peer and the F8 event-loop rule is not in play. A /// miss `reject`s so the client falls back to a cold establish-probe. -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn handle_warm_ping(conn: &Conn, req: ctl::Req) { let ctl::ReqKind::Ping { peer } = &req.kind else { return }; let peer = peer.clone(); @@ -5918,7 +5965,7 @@ async fn handle_warm_ping(conn: &Conn, req: ctl::Req) { req.reply(&reply).await; } -#[cfg(unix)] +#[cfg(any(unix, windows))] /// A non-direct (relay/WebRTC) link has no QUIC keepalive, so an idle one may be /// silently NAT/relay-evicted while `is_alive()`/`is_dead()` still lag (the read /// loop hasn't seen the EOF yet). Container/DERP paths evict ~10s; reusing such a @@ -5931,7 +5978,7 @@ async fn handle_warm_ping(conn: &Conn, req: ctl::Req) { /// tripping it means the keepalive stopped, so a fresh establish is the right answer. const WARM_RELAY_STALE_MS: u64 = 8_000; -#[cfg(unix)] +#[cfg(any(unix, windows))] /// Resolve `peer` (matched case-insensitively on the PROVEN `verified_name`, the /// same key the L2 cap gate uses) to a warm, trusted, alive link, preferring a /// direct one. The single resolver for every warm-reuse op (open + pty), so the @@ -5955,7 +6002,7 @@ fn warm_link_for(conn: &Conn, peer: &str) -> Option<(String, Arc>, tx: &mpsc::UnboundedSender, req: ctl::Req, + eof_senders_tx: mpsc::UnboundedSender<(u32, tokio::sync::watch::Sender)>, ) { let ctl::ReqKind::Open { peer, rport } = &req.kind else { return }; let (peer, rport) = (peer.clone(), *rport); @@ -5999,10 +6047,18 @@ async fn handle_warm_open( // timeout (the 25s ssh ConnectTimeout we measured). Spawned so the verify wait // never blocks the event loop (F8). tokio::spawn(async move { + use tokio::io::AsyncWriteExt; match l2::open_stream_verified(&mux, rport, l2::warm_verify_window()).await { Ok((sid, first, rx)) => { - let sock = req.accept().await; - l2::serve_verified_stream(mux, sid, sock, first, rx).await; + // Include the stream ID in the reply so the client can send an + // out-of-band EOF signal for this specific stream. + let mut sock = req.sock; + let _ = sock.write_all(format!("{{\"ok\":true,\"sid\":{sid}}}\n").as_bytes()).await; + let _ = sock.flush().await; + let shutdown_tx = l2::serve_verified_stream(mux, sid, sock, first, rx).await; + // Store the shutdown sender so the event loop can signal it on + // an out-of-band EOF for this stream. + let _ = eof_senders_tx.send((sid, shutdown_tx)); } Err(e) => { ui::debug(&format!( @@ -6018,13 +6074,14 @@ async fn handle_warm_open( /// Warm-reuse: open a PTY on `peer` over its existing link and bridge it to the /// client's stdio socket (the `filament pty` fast path). Records the session->sid /// so a later `pty-resize` can find it; the entry is dropped when the bridge ends. -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn handle_warm_pty( conn: &Conn, l2_muxes: &mut HashMap>, warm_ptys: &WarmPtys, tx: &mpsc::UnboundedSender, req: ctl::Req, + eof_senders_tx: mpsc::UnboundedSender<(u32, tokio::sync::watch::Sender)>, ) { let ctl::ReqKind::Pty { peer, session, cols, rows, term, cmd } = &req.kind else { return }; let (peer, session, cols, rows, term, cmd) = (peer.clone(), session.clone(), *cols, *rows, term.clone(), cmd.clone()); @@ -6044,13 +6101,17 @@ async fn handle_warm_pty( // pty rather than getting a dead terminal. Spawned so the verify wait never // blocks the event loop (F8). tokio::spawn(async move { + use tokio::io::AsyncWriteExt; match l2::open_pty_stream_verified(&mux, &session, cols, rows, &term, &cmd, verify).await { Ok((sid, first, rx_pipe)) => { if let Ok(mut m) = warm_ptys.lock() { m.insert(session.clone(), (pid, sid)); } - let sock = req.accept().await; - l2::serve_verified_stream(mux, sid, sock, first, rx_pipe).await; + let mut sock = req.sock; + let _ = sock.write_all(format!("{{\"ok\":true,\"sid\":{sid}}}\n").as_bytes()).await; + let _ = sock.flush().await; + let shutdown_tx = l2::serve_verified_stream(mux, sid, sock, first, rx_pipe).await; + let _ = eof_senders_tx.send((sid, shutdown_tx)); // Bridge ended (shell exit / client gone / link drop): drop our // entry, but only if it is still ours (a reconnect may have // replaced it). @@ -6072,7 +6133,7 @@ async fn handle_warm_pty( } /// Warm-reuse: relay a window-size change to an already-open warm PTY (by session). -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn handle_warm_resize( l2_muxes: &HashMap>, warm_ptys: &WarmPtys, @@ -6099,9 +6160,9 @@ async fn handle_warm_resize( /// deadline) and complete it from the `shell-bootstrap-ack`/`-deny` control arms, /// or reap it on timeout. A `Vec` per pid handles concurrent ssh to one peer (the /// ack is identical, so every waiter gets the same answer). -#[cfg(unix)] +#[cfg(any(unix, windows))] type PendingBootstraps = - HashMap>; + HashMap>; /// Warm-reuse the ssh `shell-bootstrap`: install the client's managed `pubkey` on /// `peer` over the daemon's EXISTING link instead of a fresh cold establish, the @@ -6109,7 +6170,7 @@ type PendingBootstraps = /// the last cold-establish left). Sends `shell-bootstrap` and STASHES the reply /// socket; the ack/deny handler completes it. A miss falls the client back to the /// cold `shell_bootstrap`. -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn handle_warm_bootstrap(conn: &Conn, pending: &mut PendingBootstraps, req: ctl::Req) { let (peer, pubkey, ssh_port) = match &req.kind { ctl::ReqKind::Bootstrap { peer, pubkey, ssh_port } => (peer.clone(), pubkey.clone(), *ssh_port), @@ -8825,8 +8886,18 @@ async fn recv_cmd( // Warm-pty session -> (pid, sid), so a `pty-resize` op relays to the right stream. let warm_ptys: WarmPtys = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); // Warm ssh-bootstrap reply sockets awaiting the peer's ack (see PendingBootstraps). - #[cfg(unix)] + #[cfg(any(unix, windows))] let mut pending_bootstrap: PendingBootstraps = HashMap::new(); + // Out-of-band EOF senders: stream ID -> shutdown channel sender. When a + // client sends an Eof ctl op, the daemon looks up the sender by stream ID + // and signals the write pump to stop writing and close the socket (daemon + // sees EOF on the data pipe). Spawned tasks send their shutdown_tx through + // eof_senders_tx; the event loop collects them. + #[cfg(any(unix, windows))] + let (eof_senders_tx, mut eof_senders_rx) = + mpsc::unbounded_channel::<(u32, tokio::sync::watch::Sender)>(); + #[cfg(any(unix, windows))] + let mut eof_senders: HashMap> = HashMap::new(); // Warm-link reuse: ONLY the registered `up` daemon exposes the local control // socket (a short-lived `recv`/`send` must never bind it and steal the // daemon's path). When a sibling `filament ssh`/`netcat`/`forward` asks to @@ -8835,8 +8906,9 @@ async fn recv_cmd( // held for the loop's life so the channel stays open (recv pends, never spins) // even when we are not the daemon and `serve` was not spawned. let (ctl_tx, mut ctl_rx) = mpsc::unbounded_channel::(); - // The control socket is a unix-domain socket, so warm-link reuse is unix-only. - #[cfg(unix)] + // The control socket is a unix-domain socket or Windows named pipe, so + // warm-link reuse works on both platforms. + #[cfg(any(unix, windows))] { if daemon_alive() == Some(std::process::id()) { let ctl_tx = ctl_tx.clone(); @@ -8848,8 +8920,8 @@ async fn recv_cmd( } } // Hold `ctl_tx` for the loop's life so `ctl_rx` stays open (recv pends, never - // spins) even when `serve` was not spawned (non-unix, or not the daemon). - #[cfg(not(unix))] + // spins) even when `serve` was not spawned (non-unix/windows, or not the daemon). + #[cfg(not(any(unix, windows)))] let _ = &ctl_tx; // web-shell (#4): persistent PTY sessions, keyed by a stable browser-chosen // session id, OUTLIVE the link that opened them. A dropped data channel @@ -8940,6 +9012,11 @@ async fn recv_cmd( } loop { + // Collect eof senders from spawned warm-stream tasks (non-blocking). + #[cfg(any(unix, windows))] + while let Ok((sid, eof_tx)) = eof_senders_rx.try_recv() { + eof_senders.insert(sid, eof_tx); + } // systemd liveness watchdog: ping on a throttle (well under WatchdogSec). // If this loop WEDGES on an await, the pings stop and systemd restarts us // - the backstop for the stall that also freezes the reconnect code. @@ -8965,7 +9042,7 @@ async fn recv_cmd( // Bootstrap defers its reply (awaits the peer's ack via this // loop), so it can't go through the inline handle_warm_req; it // stashes the socket in pending_bootstrap instead. - #[cfg(unix)] + #[cfg(any(unix, windows))] { // `filament set` live-reconfigure: re-read the changed key // into this loop's live state, then report whether it took @@ -9056,11 +9133,11 @@ async fn recv_cmd( if let ctl::ReqKind::Pty { peer, .. } = &req.kind { conn.note_warm_use(peer); } - handle_warm_req(&conn, &mut l2_muxes, &warm_ptys, &tx, req).await; + handle_warm_req(&conn, &mut l2_muxes, &warm_ptys, &tx, req, eof_senders_tx.clone(), &mut eof_senders).await; } } - #[cfg(not(unix))] - handle_warm_req(&conn, &mut l2_muxes, &warm_ptys, &tx, req).await; + #[cfg(not(any(unix, windows)))] + handle_warm_req(&conn, &mut l2_muxes, &warm_ptys, &tx, req, eof_senders_tx.clone(), &mut eof_senders).await; } None }