diff --git a/crates/ledger-cli/src/main.rs b/crates/ledger-cli/src/main.rs index 9ea981f..b9173fb 100644 --- a/crates/ledger-cli/src/main.rs +++ b/crates/ledger-cli/src/main.rs @@ -59,10 +59,10 @@ fn gui_args(args: Vec) -> Vec { /// The viewer's owner (ledger-gui owner.rs): its Books menu, done as the /// command line does it (the books server, the agents' own commands). -#[cfg(all(feature = "gui", unix))] +#[cfg(feature = "gui")] struct ViewerOwner(ledger_host::commands::Common); -#[cfg(all(feature = "gui", unix))] +#[cfg(feature = "gui")] impl ledger_gui::owner::Owner for ViewerOwner { fn close_books(&self) -> Result { ledger_host::commands::close_books(&self.0) @@ -124,7 +124,6 @@ impl ledger_gui::owner::Owner for ViewerOwner { #[cfg(feature = "gui")] fn run_gui(args: Vec) -> i32 { let args = gui_args(args); - #[cfg(unix)] { let value = |flag: &str| { // The last given, as the viewer reads it. @@ -256,7 +255,6 @@ fn run_setup(args: Vec) -> i32 { /// After setup: the owner passphrase, asked for at a terminal if none is /// set yet; elsewhere (a script, an agent) only said how to set it. -#[cfg(unix)] fn owner_passphrase(root: &ledger_store::root::Root) -> i32 { let auth = match root.auth_dir() { Ok(dir) => dir, @@ -294,14 +292,8 @@ fn owner_passphrase(root: &ledger_store::root::Root) -> i32 { } } -#[cfg(not(unix))] -fn owner_passphrase(_root: &ledger_store::root::Root) -> i32 { - 0 -} - /// `wanderbooks server`: the books server for the data root, on its /// local sockets (plan 3.9). The host starts it. -#[cfg(unix)] fn run_server(args: Vec) -> i32 { match ledger_mcp::local::run("wanderbooks server", args) { Ok(()) => 0, @@ -317,7 +309,6 @@ fn run_server(args: Vec) -> i32 { /// The modes that open and hold the books (plan 3.9): the host, the /// agents' bridge, and the owner's commands. -#[cfg(unix)] fn run_local(mode: &str, args: Vec) -> i32 { let prog = format!("wanderbooks {mode}"); let result = match mode { @@ -345,7 +336,6 @@ fn run_local(mode: &str, args: Vec) -> i32 { /// `upload` and `download`: the agent's file commands, whose words may /// be paths that aren't text. -#[cfg(unix)] fn run_files(mode: &str, args: Vec) -> i32 { let prog = format!("wanderbooks {mode}"); let result = match mode { @@ -364,27 +354,6 @@ fn run_files(mode: &str, args: Vec) -> i32 { } } -/// The file commands, where local sockets aren't built yet. -#[cfg(not(unix))] -fn run_files(_mode: &str, _args: Vec) -> i32 { - eprintln!("error: opening the books this way comes to this system with P11"); - 2 -} - -/// The modes that hold the books, where local sockets aren't built yet. -#[cfg(not(unix))] -fn run_local(_mode: &str, _args: Vec) -> i32 { - eprintln!("error: opening the books this way comes to this system with P11"); - 2 -} - -/// `wanderbooks server` where local sockets aren't built yet. -#[cfg(not(unix))] -fn run_server(_args: Vec) -> i32 { - eprintln!("error: the books server's local sockets come to this system with P11"); - 2 -} - fn main() { // A file's path may be any name the system takes: the file commands // get the words as given. Everything else is text. diff --git a/crates/ledger-cli/tests/common/mod.rs b/crates/ledger-cli/tests/common/mod.rs index 3fb84a1..981f4d2 100644 --- a/crates/ledger-cli/tests/common/mod.rs +++ b/crates/ledger-cli/tests/common/mod.rs @@ -63,14 +63,10 @@ impl Home { .prefix(&format!("wb-{tag}-")) .tempdir() .unwrap(); - let out = Command::new(env!("CARGO_BIN_EXE_wanderbooks")) - .args(["setup", "--home"]) - .arg(dir.path()) - .env_clear() - .env("PATH", "/usr/bin:/bin") - .env("HOME", dir.path()) - .output() - .unwrap(); + let mut setup = Command::new(env!("CARGO_BIN_EXE_wanderbooks")); + setup.args(["setup", "--home"]).arg(dir.path()); + clean_env(&mut setup, dir.path()); + let out = setup.output().unwrap(); assert!(out.status.success(), "{out:?}"); Self { dir, @@ -79,7 +75,7 @@ impl Home { } pub fn root(&self) -> PathBuf { - self.dir.path().join(".local/share/wanderbooks") + ledger_store::root::default_root(self.dir.path()) } pub fn run(&self) -> PathBuf { @@ -190,14 +186,14 @@ impl Server { .append(true) .open(home.dir.path().join("server.log")) .unwrap(); - let mut child = Command::new(env!("CARGO_BIN_EXE_wanderbooks")) + let mut command = Command::new(env!("CARGO_BIN_EXE_wanderbooks")); + command .arg("server") .arg("--root") .arg(home.root()) - .current_dir(cwd) - .env_clear() - .env("PATH", "/usr/bin:/bin") - .env("HOME", home.dir.path()) + .current_dir(cwd); + clean_env(&mut command, home.dir.path()); + let mut child = command .env("XDG_CONFIG_HOME", home.dir.path().join(".config")) .envs(env.iter().map(|(k, v)| (*k, v.as_os_str()))) .stdin(Stdio::piped()) @@ -248,10 +244,10 @@ pub fn mcp_http_builder(home: &Home) -> reqwest::ClientBuilder { /// [`mcp_http_builder`], for the root's `run/` folder `run`. pub fn mcp_http_builder_at(run: &Path) -> reqwest::ClientBuilder { // The short path is good in this process only, while it lives: kept. - let short = ledger_store::root::SocketPath::new(run, "mcp.sock").unwrap(); - let path = short.path().to_path_buf(); - std::mem::forget(short); - reqwest::Client::builder().unix_socket(path) + let endpoint = ledger_mcp::ipc::Endpoint::new(run, "mcp.sock").unwrap(); + let builder = ledger_host::channel::http_builder(&endpoint); + std::mem::forget(endpoint); + builder } /// The settings a settings file's text gives, as books.set_setting changes @@ -836,20 +832,71 @@ pub fn book_files(dir: &Path) -> std::collections::BTreeMap { /// Copy the tree `from` into `to`, private as the server keeps a book /// (folders 0700, files 0600). pub fn copy_private(from: &Path, to: &Path) { - use std::os::unix::fs::PermissionsExt; std::fs::create_dir_all(to).unwrap(); - std::fs::set_permissions(to, std::fs::Permissions::from_mode(0o700)).unwrap(); + private_folder(to); for entry in std::fs::read_dir(from).unwrap().flatten() { let (src, dst) = (entry.path(), to.join(entry.file_name())); if entry.file_type().unwrap().is_dir() { copy_private(&src, &dst); } else { std::fs::copy(&src, &dst).unwrap(); - std::fs::set_permissions(&dst, std::fs::Permissions::from_mode(0o600)).unwrap(); + private_file(&dst); } } } +/// Make the folder `dir` private as the server keeps one (0700; on +/// Windows this account's alone, by an owner-only DACL). +pub fn private_folder(dir: &Path) { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + } + #[cfg(windows)] + ledger_store::winsec::set_owner_only(dir).unwrap(); +} + +/// Make the file `path` private as the server keeps one (0600; on +/// Windows this account's alone). +pub fn private_file(path: &Path) { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap(); + } + #[cfg(windows)] + ledger_store::winsec::set_owner_only(path).unwrap(); +} + +/// A child process's environment cleared down to what a program needs to +/// run and to find `home` as its own: PATH and HOME on Unix; the system +/// folder, a PATH of the system's, the profile and the temporary folder on +/// Windows (a program started with nothing at all can't load its DLLs). +pub fn clean_env(command: &mut Command, home: &Path) { + command.env_clear(); + #[cfg(unix)] + { + command.env("PATH", "/usr/bin:/bin").env("HOME", home); + } + #[cfg(windows)] + { + let root = std::env::var_os("SystemRoot").unwrap_or_else(|| r"C:\Windows".into()); + let system32 = Path::new(&root).join("System32"); + command + .env("SystemRoot", &root) + .env("windir", &root) + .env( + "PATH", + std::env::join_paths([system32, root.clone().into()]).unwrap(), + ) + .env("USERPROFILE", home) + .env("HOME", home) + .env("TEMP", std::env::temp_dir()) + .env("TMP", std::env::temp_dir()); + } +} + /// A test's place: a home (whose folder holds the test's own files, its /// settings file among them) and the books [`BOOK`] in its root, served /// by a server [`Place::start`] starts, and starts again on each next @@ -912,8 +959,7 @@ impl Place { let dir = self.books_data(); let fresh = !dir.join("journal.jsonl").exists(); if dir.exists() { - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + private_folder(&dir); } let cwd = self.cwd.borrow().clone(); let cwd = cwd.as_deref().unwrap_or(self.home.dir.path()); diff --git a/crates/ledger-cli/tests/one_executable.rs b/crates/ledger-cli/tests/one_executable.rs index f9134e9..91434b8 100644 --- a/crates/ledger-cli/tests/one_executable.rs +++ b/crates/ledger-cli/tests/one_executable.rs @@ -2,6 +2,7 @@ //! (`server`) and the viewer (`gui`) as modes of itself, each with its own //! flags, and the server-only build has no viewer. +#[cfg_attr(not(unix), allow(unused_imports))] use std::process::{Command, Stdio}; #[cfg(all(feature = "gui", target_os = "linux"))] @@ -105,13 +106,12 @@ fn a_viewer_that_cannot_open_a_window_exits_1() { let server = common::Server::start(&home); home.create(common::BOOK, "USD"); assert!(server.close().success()); - let mut child = wanderbooks() - .args(["gui", "--root"]) + let mut gui = wanderbooks(); + gui.args(["gui", "--root"]) .arg(home.root()) - .args(["--book", common::BOOK]) - .env_clear() - .env("PATH", "/usr/bin:/bin") - .env("HOME", home.dir.path()) + .args(["--book", common::BOOK]); + common::clean_env(&mut gui, home.dir.path()); + let mut child = gui .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() @@ -147,7 +147,6 @@ fn the_server_only_build_has_no_viewer() { } /// The books server needs a data root: with none, it says to run setup. -#[cfg(unix)] #[test] fn the_books_server_needs_a_root() { // No root yet: the books server says to run setup. @@ -258,52 +257,3 @@ fn the_file_commands_end_with_the_code_that_says_how() { ); assert!(!root.exists(), "the books were reached for"); } - -/// Where local sockets aren't built yet (P11, Windows): the modes that -/// hold the books say so and end with 2, never a success; setup makes -/// the folder and ends with 0, asking for no passphrase (nothing there -/// asks for one yet). -#[cfg(not(unix))] -#[test] -fn without_local_sockets_the_modes_that_hold_the_books_say_so() { - for mode in [ - "server", - "host", - "mcp", - "serve", - "close", - "books", - "doctor", - "token", - "connect", - "passphrase", - "upload", - "download", - ] { - let out = wanderbooks() - .arg(mode) - .stdin(Stdio::null()) - .output() - .unwrap(); - assert_eq!(out.status.code(), Some(2), "{mode}: {out:?}"); - assert!( - String::from_utf8_lossy(&out.stderr).contains("to this system with P11"), - "{mode}: {out:?}" - ); - } - let home = tempfile::Builder::new() - .prefix("wb-one-nosock-") - .tempdir() - .unwrap(); - let out = wanderbooks() - .args(["setup", "--home"]) - .arg(home.path()) - .env_remove("XDG_DATA_HOME") - .stdin(Stdio::null()) - .output() - .unwrap(); - assert_eq!(out.status.code(), Some(0), "{out:?}"); - let text = String::from_utf8_lossy(&out.stdout); - assert!(text.contains("The books' private folder: made"), "{text}"); - assert!(!text.contains("passphrase"), "{text}"); -} diff --git a/crates/ledger-cli/tests/startup.rs b/crates/ledger-cli/tests/startup.rs index 66310fa..cbd7dee 100644 --- a/crates/ledger-cli/tests/startup.rs +++ b/crates/ledger-cli/tests/startup.rs @@ -15,10 +15,9 @@ mod common; /// New books `id` written in `home`'s root by the store itself, before a /// server opens them; the store, still open. fn written(home: &Home, id: &str) -> ledger_store::Store { - use std::os::unix::fs::PermissionsExt; let dir = home.book_dir(id); let store = ledger_store::Store::create(&dir, ledger_core::Currency::USD).unwrap(); - std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + common::private_folder(&dir); store } diff --git a/crates/ledger-gui/src/receipts.rs b/crates/ledger-gui/src/receipts.rs index e4e3822..7bf1631 100644 --- a/crates/ledger-gui/src/receipts.rs +++ b/crates/ledger-gui/src/receipts.rs @@ -380,6 +380,10 @@ fn keep(path: &std::path::Path, png: &[u8]) { use std::os::unix::fs::PermissionsExt; let _ = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)); } + // Windows: this account's alone (an owner-only DACL, inherited by the + // thumbnails written inside). + #[cfg(windows)] + let _ = ledger_store::winsec::set_owner_only(dir); // One name per write, across viewers (the pid) and this one's workers. static WRITES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let tmp = dir.join(format!( diff --git a/crates/ledger-gui/tests/read_only.rs b/crates/ledger-gui/tests/read_only.rs index 4af8710..dd993dc 100644 --- a/crates/ledger-gui/tests/read_only.rs +++ b/crates/ledger-gui/tests/read_only.rs @@ -19,6 +19,7 @@ use ledger_gui::data::{self, Loader, Update}; use ledger_gui::nav::{ReportKind, Route}; use ledger_gui::{App, Source}; use std::collections::BTreeMap; +#[cfg(unix)] use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; use std::time::{Duration, SystemTime}; @@ -46,22 +47,36 @@ fn tree(root: &Path) -> Tree { } /// Every file 0444 and folder 0555 (`read_only`), or back to 0644/0755. +/// On Windows a file gets its read-only attribute (which stops a write +/// to it) and a folder is left as it is: a folder can't be made +/// read-only there. fn set_read_only(root: &Path, read_only: bool) { fn walk(p: &Path, ro: bool) { let m = std::fs::symlink_metadata(p).unwrap(); if m.is_dir() { + #[cfg(unix)] if !ro { std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o755)).unwrap(); } for e in std::fs::read_dir(p).unwrap() { walk(&e.unwrap().path(), ro); } + #[cfg(unix)] if ro { std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o555)).unwrap(); } } else { - let mode = if ro { 0o444 } else { 0o644 }; - std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode)).unwrap(); + #[cfg(unix)] + { + let mode = if ro { 0o444 } else { 0o644 }; + std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode)).unwrap(); + } + #[cfg(windows)] + { + let mut permissions = m.permissions(); + permissions.set_readonly(ro); + std::fs::set_permissions(p, permissions).unwrap(); + } } } walk(root, read_only); diff --git a/crates/ledger-host/src/bridge.rs b/crates/ledger-host/src/bridge.rs index cf9d013..09f08b4 100644 --- a/crates/ledger-host/src/bridge.rs +++ b/crates/ledger-host/src/bridge.rs @@ -162,8 +162,8 @@ struct Bridge { next_id: std::sync::atomic::AtomicU64, /// The agent's requests not yet answered, by id (INVARIANT 8). pending: Mutex>, - /// The server's socket, for its owner check (INVARIANT 9). - socket: PathBuf, + /// The server's channel, for its owner check (INVARIANT 9). + endpoint: ledger_mcp::ipc::Endpoint, } /// Who has the server's socket, when it isn't plainly this account. @@ -1034,9 +1034,9 @@ impl Bridge { /// Whether the server's socket belongs to this account (INVARIANT 9). fn socket_is_ours(&self) -> Result<(), Owner> { - let stream = std::os::unix::net::UnixStream::connect(&self.socket) - .map_err(|_| Owner::Unreachable)?; - if ledger_mcp::local::peer_is_us(&stream) { + let stream = + ledger_mcp::ipc::connect_std(&self.endpoint).map_err(|_| Owner::Unreachable)?; + if ledger_mcp::ipc::peer_is_us(&stream) { Ok(()) } else { Err(Owner::Other( @@ -1314,18 +1314,16 @@ pub fn run( .map_err(|e| e.to_string())? .to_string(); let opener = Opener::new(root.as_deref(), knobs, Kind::Bridge, Some(book.clone()))?; - // Kept for the whole session: the client reaches the socket by it. - let socket = - ledger_store::root::SocketPath::new(opener.run_dir(), ledger_mcp::local::MCP_SOCKET) - .map_err(|e| format!("can't reach the books server: {e}"))?; + // Kept for the whole session: the client reaches the channel by it. + let endpoint = ledger_mcp::ipc::Endpoint::new(opener.run_dir(), ledger_mcp::local::MCP_SOCKET) + .map_err(|e| format!("can't reach the books server: {e}"))?; let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build()?; let result = rt.block_on(async move { // A connection per call: one the server closes while idle can't // swallow a call (INVARIANT 4). - let http = reqwest::Client::builder() - .unix_socket(socket.path()) + let http = crate::channel::http_builder(&endpoint) .pool_max_idle_per_host(0) .no_proxy() .build() @@ -1347,7 +1345,7 @@ pub fn run( tools_seen: Mutex::new(None), next_id: std::sync::atomic::AtomicU64::new(1), pending: Mutex::new(std::collections::HashMap::new()), - socket: socket.path().to_path_buf(), + endpoint, }); let keeper = tokio::spawn(bridge.clone().keep(opener)); let writer = tokio::spawn(async move { @@ -1363,12 +1361,7 @@ pub fn run( // A client that stops the bridge with a signal (most do) still has // its session ended upstream (INVARIANT 10). let stop = Arc::new(Notify::new()); - for kind in [ - tokio::signal::unix::SignalKind::terminate(), - tokio::signal::unix::SignalKind::interrupt(), - tokio::signal::unix::SignalKind::hangup(), - ] { - let mut signals = tokio::signal::unix::signal(kind).map_err(|e| e.to_string())?; + for mut signals in ledger_mcp::ipc::stop_signals().map_err(|e| e.to_string())? { let stop = stop.clone(); tokio::spawn(async move { if signals.recv().await.is_some() { @@ -1537,7 +1530,7 @@ mod tests { /// The bridge against a scripted server on a Unix socket: every way an /// answer can come back, and the bridge's own bookkeeping. -#[cfg(test)] +#[cfg(all(test, unix))] mod against_a_server { use super::*; use std::collections::{HashMap, VecDeque}; @@ -1669,8 +1662,12 @@ mod against_a_server { } fn bridge_at(socket: &Path, knobs: Knobs) -> (Arc, mpsc::UnboundedReceiver) { - let http = reqwest::Client::builder() - .unix_socket(socket) + let endpoint = ledger_mcp::ipc::Endpoint::new( + socket.parent().unwrap(), + socket.file_name().and_then(|n| n.to_str()).unwrap(), + ) + .unwrap(); + let http = crate::channel::http_builder(&endpoint) .pool_max_idle_per_host(0) .no_proxy() .build() @@ -1693,7 +1690,7 @@ mod against_a_server { tools_seen: Mutex::new(None), next_id: std::sync::atomic::AtomicU64::new(1), pending: Mutex::new(std::collections::HashMap::new()), - socket: socket.to_path_buf(), + endpoint, }); (bridge, rx) } diff --git a/crates/ledger-host/src/channel.rs b/crates/ledger-host/src/channel.rs new file mode 100644 index 0000000..935618d --- /dev/null +++ b/crates/ledger-host/src/channel.rs @@ -0,0 +1,19 @@ +//! The HTTP client for the books server's channel (plan 3.9, D-62): a +//! Unix socket on Unix, a named pipe on Windows. The server speaks HTTP +//! on it; only how a client reaches it differs. + +use ledger_mcp::ipc::Endpoint; + +/// A client builder that connects to `endpoint` whatever URL it is given +/// (`http://localhost/...`). On Unix the endpoint must outlive the client: +/// a long socket path is reached through a handle it keeps open. +pub fn http_builder(endpoint: &Endpoint) -> reqwest::ClientBuilder { + #[cfg(unix)] + { + reqwest::Client::builder().unix_socket(endpoint.path().to_path_buf()) + } + #[cfg(windows)] + { + reqwest::Client::builder().windows_named_pipe(endpoint.pipe_name().to_string()) + } +} diff --git a/crates/ledger-host/src/commands.rs b/crates/ledger-host/src/commands.rs index e08056d..dfba42e 100644 --- a/crates/ledger-host/src/commands.rs +++ b/crates/ledger-host/src/commands.rs @@ -302,12 +302,7 @@ pub fn serve( // Ctrl-C, SIGTERM and SIGHUP are heard from the start, whatever // serve is doing (opening, waiting to try again): none is missed. let stop = std::sync::Arc::new(tokio::sync::Notify::new()); - for kind in [ - tokio::signal::unix::SignalKind::interrupt(), - tokio::signal::unix::SignalKind::terminate(), - tokio::signal::unix::SignalKind::hangup(), - ] { - let mut signals = tokio::signal::unix::signal(kind)?; + for mut signals in ledger_mcp::ipc::stop_signals()? { let stop = stop.clone(); tokio::spawn(async move { while signals.recv().await.is_some() { @@ -872,8 +867,13 @@ fn offer_upgrades(run: &Path) { /// or a job in the background, would wait unseen for an answer, and mix /// the question into its output). fn offer_upgrade(run: &Path, book: &str, waiting: &admin::FormatWaiting) { + // Windows has no foreground process group to ask about: a terminal + // is all there is to see. + #[cfg(unix)] // SAFETY: isatty, tcgetpgrp and getpgrp have no preconditions. let seen = unsafe { libc::isatty(1) == 1 && libc::tcgetpgrp(0) == libc::getpgrp() }; + #[cfg(windows)] + let seen = std::io::IsTerminal::is_terminal(&std::io::stdout()); if !(seen && can_type_a_passphrase()) { eprintln!( "note: the books \"{book}\" wait for a format upgrade (to {}): until the owner agrees, with `wanderbooks books upgrade {book}` at a terminal or in the app, they work on their old format.", @@ -1160,6 +1160,20 @@ pub fn connect( } } +/// Whether `p` is a file someone may run (Windows: a file; what it can +/// run as is told by its extension, which the caller chose). +fn is_runnable(p: &Path) -> bool { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::metadata(p).is_ok_and(|m| m.is_file() && m.permissions().mode() & 0o111 != 0) + } + #[cfg(not(unix))] + { + std::fs::metadata(p).is_ok_and(|m| m.is_file()) + } +} + /// Let the agent `agent` ("claude" or "codex") reach the books `book`: /// their token kept in its token file (made if none is), and `wanderbooks /// mcp --book ` added to the agent's own list as @@ -1176,7 +1190,7 @@ pub fn connect_agent(common: &Common, agent: &str, book: &str) -> Result Result Result>() }) + .find(|p| is_runnable(p)) .ok_or_else(|| { format!("{label} isn't installed here (no `{cli}` on the PATH, nor where its installer puts it)") })?; @@ -1610,8 +1634,7 @@ fn vault_beside(source: &Path) -> Option { /// Whether standard input is a terminal. fn at_a_terminal() -> bool { - // SAFETY: isatty has no preconditions. - unsafe { libc::isatty(0) == 1 } + std::io::IsTerminal::is_terminal(&std::io::stdin()) } /// `wanderbooks books adopt` (INVARIANT 3). @@ -2397,10 +2420,9 @@ fn on_the_socket( ) })?; let token = zeroize::Zeroizing::new(token); - let socket = ledger_store::root::SocketPath::new(&run_dir, ledger_mcp::local::MCP_SOCKET) + let socket = ledger_mcp::ipc::Endpoint::new(&run_dir, ledger_mcp::local::MCP_SOCKET) .map_err(|e| format!("can't reach the books server: {e}"))?; - let http = reqwest::Client::builder() - .unix_socket(socket.path()) + let http = crate::channel::http_builder(&socket) .no_proxy() .build() .map_err(|e| e.to_string())?; diff --git a/crates/ledger-host/src/host.rs b/crates/ledger-host/src/host.rs index 7921311..c3a3706 100644 --- a/crates/ledger-host/src/host.rs +++ b/crates/ledger-host/src/host.rs @@ -53,6 +53,7 @@ use crate::knobs::{FLAG, Knobs}; use crate::lease::{self, Ask, HostMsg, Intent, LeaseInfo, Reason}; use ledger_store::root::Root; use std::collections::BTreeMap; +#[cfg(unix)] use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -80,7 +81,23 @@ const LOG_LIMIT: u64 = 8 * 1024 * 1024; /// The exit code when another host holds the root. pub const EXIT_ANOTHER_HOST: i32 = 3; +/// File options for what the host makes: mode 0600 on Unix; on Windows +/// the file takes the owner-only DACL its folder carries (P11c). +fn private_options() -> std::fs::OpenOptions { + #[allow(unused_mut)] + let mut options = std::fs::OpenOptions::new(); + #[cfg(unix)] + options.mode(0o600); + options +} + +/// Windows passes no descriptor to a child unless asked to, so there is +/// nothing to close (INVARIANT 2). +#[cfg(not(unix))] +fn close_stray_descriptors() {} + /// Close every descriptor above 2 (INVARIANT 2). +#[cfg(unix)] fn close_stray_descriptors() { #[cfg(target_os = "linux")] { @@ -112,12 +129,7 @@ fn confined(run: &Path) -> Option<&'static str> { } } let probe = run.join(format!(".host-probe-{}", std::process::id())); - match std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o600) - .open(&probe) - { + match private_options().write(true).create_new(true).open(&probe) { Ok(_) => { let _ = std::fs::remove_file(&probe); None @@ -138,11 +150,10 @@ fn take_host_lock(run: &Path) -> Result, String> { /// [`take_host_lock`], waiting at most `wait` (tests shorten it). fn take_host_lock_within(run: &Path, wait: Duration) -> Result, String> { let path = run.join(lease::HOST_LOCK); - let file = std::fs::OpenOptions::new() + let file = private_options() .create(true) .truncate(false) .write(true) - .mode(0o600) .open(&path) .map_err(|e| format!("can't open the host lock: {}", e.kind()))?; let deadline = std::time::Instant::now() + wait; @@ -180,11 +191,7 @@ fn open_log_within(path: &Path, limit: u64) -> std::io::Result { if std::fs::metadata(path).is_ok_and(|m| m.len() > limit) { let _ = std::fs::rename(path, path.with_extension("log.1")); } - std::fs::OpenOptions::new() - .create(true) - .append(true) - .mode(0o600) - .open(path) + private_options().create(true).append(true).open(path) } /// The executable the server is started from (INVARIANT 3). @@ -196,9 +203,13 @@ fn children_executable() -> PathBuf { } } +/// A program file's identity: what changes when the file is replaced. +type Identity = (u64, u64); + /// The running executable's file identity, and its file's now: they /// differ once the file was replaced (INVARIANT 7). -fn running_identity() -> Option<(u64, u64)> { +#[cfg(unix)] +fn running_identity() -> Option { let running = if cfg!(target_os = "linux") { std::fs::metadata("/proc/self/exe").ok()? } else { @@ -207,19 +218,46 @@ fn running_identity() -> Option<(u64, u64)> { Some((running.dev(), running.ino())) } +/// Windows can't replace a program that runs, so an update puts the new +/// file at the old path (the old one moved aside): the identity is the +/// file's size and last write at the path, read when the host started. +#[cfg(windows)] +fn running_identity() -> Option { + identity_of(&std::env::current_exe().ok()?) +} + /// The identity of the file at this executable's path, if it is a /// program (a file half-written or gone doesn't count yet). -fn file_identity() -> Option<(u64, u64)> { +fn file_identity() -> Option { identity_of(&crate::start::this_executable().ok()?) } /// The identity (device, inode) of the program at `path`: a file someone /// may run; else None. -fn identity_of(path: &Path) -> Option<(u64, u64)> { +#[cfg(unix)] +fn identity_of(path: &Path) -> Option { let meta = std::fs::metadata(path).ok()?; (meta.is_file() && meta.permissions().mode() & 0o111 != 0).then(|| (meta.dev(), meta.ino())) } +/// The identity (size, last write in nanoseconds) of the program at +/// `path`: a file; else None. +#[cfg(windows)] +fn identity_of(path: &Path) -> Option { + let meta = std::fs::metadata(path).ok()?; + let written = meta + .modified() + .ok()? + .duration_since(std::time::UNIX_EPOCH) + .ok()?; + meta.is_file().then(|| { + ( + meta.len(), + u64::try_from(written.as_nanos()).unwrap_or(u64::MAX), + ) + }) +} + /// Wait at most `wait` for a host to answer on the lease socket: whether /// one did. fn wait_for_a_host(run: &Path, wait: Duration) -> bool { @@ -347,19 +385,29 @@ pub fn run( /// Start the next host from the file now on disk, with this one's /// arguments and (already clean) environment (INVARIANT 7). fn restart() -> Result<(), String> { - use std::os::unix::process::CommandExt; let exe = crate::start::this_executable()?; let mut c = std::process::Command::new(exe); c.args(std::env::args_os().skip(1)) - .current_dir("/") .stdin(std::process::Stdio::null()) .stdout(std::process::Stdio::null()); - // SAFETY: setsid is async-signal-safe and touches only the child. - unsafe { - c.pre_exec(|| { - libc::setsid(); - Ok(()) - }); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + c.current_dir("/"); + // SAFETY: setsid is async-signal-safe and touches only the child. + unsafe { + c.pre_exec(|| { + libc::setsid(); + Ok(()) + }); + } + } + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + // Detached from this console and its Ctrl+C, as a new session is + // on Unix. + c.creation_flags(crate::start::DETACHED_HOST); } c.spawn() .map(drop) @@ -456,20 +504,11 @@ impl Host { /// Bind the lease socket, start the server, and serve leases until /// the books close. async fn serve(mut self) -> Result { - let listener = bind(&self.run)?; + let mut listener = bind(&self.run)?; let (events, mut inbox) = mpsc::unbounded_channel::(); let connections = TaskTracker::new(); - let mut signals = Vec::new(); - for kind in [ - tokio::signal::unix::SignalKind::terminate(), - tokio::signal::unix::SignalKind::interrupt(), - tokio::signal::unix::SignalKind::hangup(), - ] { - signals.push( - tokio::signal::unix::signal(kind) - .map_err(|e| format!("can't watch for signals: {e}"))?, - ); - } + let signals = + ledger_mcp::ipc::stop_signals().map_err(|e| format!("can't watch for signals: {e}"))?; let (stop_tx, mut stop) = mpsc::unbounded_channel::<()>(); for mut s in signals { let stop_tx = stop_tx.clone(); @@ -513,9 +552,9 @@ impl Host { } } accepted = listener.accept() => match accepted { - Ok((stream, _)) if ledger_mcp::local::peer_is_us(&stream) => { + Ok(stream) if ledger_mcp::ipc::peer_is_us(&stream) => { // The holder's process, as the kernel knows it. - let pid = ledger_mcp::local::peer_pid(&stream); + let pid = ledger_mcp::ipc::peer_pid(&stream); connections.spawn(connection(stream, pid, events.clone())); } Ok(_) => {} @@ -674,7 +713,9 @@ impl Host { let log = open_log(&self.run.join(SERVER_LOG)) .map_err(|e| format!("can't open the server's log: {}", e.kind()))?; let mut command = tokio::process::Command::new(children_executable()); - command.arg0("wanderbooks").arg("server"); + #[cfg(unix)] + command.arg0("wanderbooks"); + command.arg("server"); if self.explicit_root { command.arg("--root").arg(self.root.path()); } @@ -815,11 +856,10 @@ fn consented(socket: &Path, book: String, passphrase: ledger_mcp::admin::Passphr /// Take run/update.lock (exclusive), if it is free now. fn take_update_lock(run: &Path) -> Result { - let file = std::fs::OpenOptions::new() + let file = private_options() .create(true) .truncate(false) .write(true) - .mode(0o600) .open(run.join(lease::UPDATE_LOCK)) .map_err(|e| format!("can't open the update lock: {}", e.kind()))?; match file.try_lock() { @@ -834,31 +874,22 @@ fn take_update_lock(run: &Path) -> Result { } /// Bind run/lease.sock, replacing a stale one (only the host holding the -/// lock gets here). -fn bind(run: &Path) -> Result { - let path = run.join(lease::LEASE_SOCKET); - match std::fs::remove_file(&path) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(format!("can't replace the lease socket: {}", e.kind())), - } - let short = ledger_store::root::SocketPath::new(run, lease::LEASE_SOCKET) +/// lock gets here); on Windows the lease pipe of the root. +fn bind(run: &Path) -> Result { + let endpoint = ledger_mcp::ipc::Endpoint::new(run, lease::LEASE_SOCKET) .map_err(|e| format!("can't make the lease socket: {e}"))?; - let listener = tokio::net::UnixListener::bind(short.path()) - .map_err(|e| format!("can't make the lease socket: {}", e.kind()))?; - std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) - .map_err(|e| format!("can't make the lease socket private: {}", e.kind()))?; - Ok(listener) + ledger_mcp::ipc::Listener::bind(&endpoint) + .map_err(|e| format!("can't make the lease socket: {}", e.kind())) } /// One connection: its first line, then a lease or one answer /// (lease.rs INVARIANT 1). async fn connection( - stream: tokio::net::UnixStream, + stream: ledger_mcp::ipc::Accepted, peer_pid: Option, events: mpsc::UnboundedSender, ) { - let (read, mut write) = stream.into_split(); + let (read, mut write) = ledger_mcp::ipc::split_accepted(stream); let mut reader = tokio::io::BufReader::new(read.take(lease::MAX_LINE as u64 + 1)); let mut first = String::new(); let answer = |msg: HostMsg| lease::line(&msg); @@ -1019,13 +1050,31 @@ mod tests { "123456" ); assert_eq!(std::fs::read_to_string(&log).unwrap(), "", "a new log"); - use std::os::unix::fs::PermissionsExt; - assert_eq!( - std::fs::metadata(&log).unwrap().permissions().mode() & 0o777, - 0o600 - ); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(&log).unwrap().permissions().mode() & 0o777, + 0o600 + ); + } } + #[cfg(windows)] + #[test] + fn a_folder_the_host_can_write_isnt_a_sandbox() { + let dir = tempfile::tempdir().unwrap(); + assert_eq!(confined(dir.path()), None); + assert_eq!( + std::fs::read_dir(dir.path()).unwrap().count(), + 0, + "its probe went" + ); + // A folder that isn't there: no sign of a sandbox. + assert_eq!(confined(&dir.path().join("gone")), None); + } + + #[cfg(unix)] #[test] fn a_folder_the_host_can_write_isnt_a_sandbox() { let seccomp = std::fs::read_to_string("/proc/self/status") @@ -1061,6 +1110,24 @@ mod tests { assert_eq!(confined(&dir.path().join("gone")), None); } + #[cfg(windows)] + #[test] + fn only_a_file_has_an_identity_and_a_replaced_one_another() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("wanderbooks.exe"); + std::fs::write(&file, "x").unwrap(); + let first = identity_of(&file).expect("a file"); + // Replaced by another file (as an update does): another identity. + let next = dir.path().join("next.exe"); + std::fs::write(&next, "yy").unwrap(); + std::fs::rename(&next, &file).unwrap(); + let second = identity_of(&file).unwrap(); + assert_ne!(first, second); + assert_eq!(identity_of(dir.path()), None, "a folder isn't one"); + assert_eq!(identity_of(&dir.path().join("gone")), None); + } + + #[cfg(unix)] #[test] fn only_a_program_file_has_an_identity() { use std::os::unix::fs::PermissionsExt; diff --git a/crates/ledger-host/src/lib.rs b/crates/ledger-host/src/lib.rs index cb4f568..94054e0 100644 --- a/crates/ledger-host/src/lib.rs +++ b/crates/ledger-host/src/lib.rs @@ -11,17 +11,13 @@ //! viewer's Books menu does). Agents' tokens are kept in files under the data root //! ([`tokens`]). -#[cfg(unix)] pub mod bridge; -#[cfg(unix)] +pub mod channel; pub mod commands; -#[cfg(unix)] pub mod host; pub mod knobs; pub mod lease; -#[cfg(unix)] pub mod opener; -#[cfg(unix)] mod start; pub mod tokens; diff --git a/crates/ledger-host/src/opener.rs b/crates/ledger-host/src/opener.rs index 53eba6e..e063856 100644 --- a/crates/ledger-host/src/opener.rs +++ b/crates/ledger-host/src/opener.rs @@ -109,8 +109,8 @@ pub struct Opener { pub struct Lease { /// Its number. pub id: u64, - lines: tokio::io::Lines>>, - _write: tokio::net::unix::OwnedWriteHalf, + lines: tokio::io::Lines>>, + _write: ledger_mcp::ipc::ConnectedWrite, } impl Lease { @@ -207,11 +207,11 @@ impl Opener { let mut started: Option = None; loop { wait_for_restart(&self.run).await; - let short = ledger_store::root::SocketPath::new(&self.run, lease::LEASE_SOCKET) + let endpoint = ledger_mcp::ipc::Endpoint::new(&self.run, lease::LEASE_SOCKET) .map_err(|e| format!("can't reach the books' host: {e}"))?; - match tokio::net::UnixStream::connect(short.path()).await { + match ledger_mcp::ipc::connect(&endpoint).await { Ok(stream) => { - if !ledger_mcp::local::peer_is_us(&stream) { + if !ledger_mcp::ipc::peer_is_us(&stream) { return Err( "the books' host socket belongs to another account: not used".into(), ); @@ -279,8 +279,8 @@ impl Opener { } /// Say hello on a connection and read the host's first word. - async fn ask(&self, stream: tokio::net::UnixStream) -> Result { - let (read, mut write) = stream.into_split(); + async fn ask(&self, stream: ledger_mcp::ipc::Connected) -> Result { + let (read, mut write) = ledger_mcp::ipc::split_connected(stream); let hello = Hello { proto: lease::PROTO, ask: Ask::Hold { @@ -402,14 +402,14 @@ pub fn ask_once(run: &Path, ask: Ask) -> Result, String> { /// [`ask_once`], waiting at most `wait` for the answer (a probe). pub fn ask_once_within(run: &Path, ask: Ask, wait: Duration) -> Result, String> { use std::io::{BufRead, Write}; - let short = ledger_store::root::SocketPath::new(run, lease::LEASE_SOCKET) + let endpoint = ledger_mcp::ipc::Endpoint::new(run, lease::LEASE_SOCKET) .map_err(|e| format!("can't reach the books' host: {e}"))?; - let stream = match std::os::unix::net::UnixStream::connect(short.path()) { + let stream = match ledger_mcp::ipc::connect_std(&endpoint) { Ok(s) => s, Err(e) if no_host(&e) => return Ok(None), Err(e) => return Err(format!("can't reach the books' host: {}", e.kind())), }; - if !ledger_mcp::local::peer_is_us(&stream) { + if !ledger_mcp::ipc::peer_is_us(&stream) { return Err("the books' host socket belongs to another account: not used".into()); } stream diff --git a/crates/ledger-host/src/start.rs b/crates/ledger-host/src/start.rs index 47f7bae..56bb78b 100644 --- a/crates/ledger-host/src/start.rs +++ b/crates/ledger-host/src/start.rs @@ -28,18 +28,34 @@ //! ``` use crate::knobs::Knobs; +#[cfg(unix)] use std::os::unix::fs::{MetadataExt, OpenOptionsExt}; +#[cfg(unix)] use std::os::unix::process::CommandExt; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; +#[cfg(unix)] use std::time::{Duration, Instant}; /// The host's PATH (INVARIANT 1). +#[cfg(unix)] const PATH: &str = "/usr/local/bin:/usr/bin:/bin"; /// Where systemd-run may be (never looked up on the opener's PATH). +#[cfg(unix)] const SYSTEMD_RUN: [&str; 2] = ["/usr/bin/systemd-run", "/bin/systemd-run"]; +/// Windows: a process of its own, without this console, its Ctrl+C or its +/// window (DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP): what `setsid` +/// is on Unix. +#[cfg(windows)] +pub(crate) const DETACHED_HOST: u32 = 0x0000_0008 | 0x0000_0200; +/// Windows: leave the job the opener runs in, if it lets its processes +/// (CREATE_BREAKAWAY_FROM_JOB): a host must not end with an agent's job, +/// as it must not end with an app's scope on Linux (INVARIANT 3). +#[cfg(windows)] +const BREAKAWAY: u32 = 0x0100_0000; + /// The host's log, in the root's `run/`. pub const HOST_LOG: &str = "host.log"; /// A log larger than this is moved aside when a host starts. @@ -60,11 +76,13 @@ pub fn this_executable() -> Result { } /// This account's user ID. +#[cfg(unix)] fn own_uid() -> u32 { // SAFETY: geteuid has no preconditions and can't fail. unsafe { libc::geteuid() } } +#[cfg(unix)] /// `XDG_RUNTIME_DIR` from `value`, if it is this account's own private /// runtime folder: absolute, a real folder (no link), owned by this /// account, and closed to everyone else. @@ -79,6 +97,7 @@ pub fn own_runtime_dir(value: Option<&std::ffi::OsStr>) -> Option { (meta.is_dir() && meta.uid() == own_uid() && meta.mode() & 0o077 == 0).then_some(dir) } +#[cfg(unix)] /// `DBUS_SESSION_BUS_ADDRESS` from `value`, if it is the bus in `runtime` /// (`unix:path=/bus`, perhaps with a guid after a comma). pub fn own_bus(value: Option<&std::ffi::OsStr>, runtime: &Path) -> Option { @@ -88,6 +107,7 @@ pub fn own_bus(value: Option<&std::ffi::OsStr>, runtime: &Path) -> Option Option { .or_else(|| own_runtime_dir(Some(format!("/run/user/{}", own_uid()).as_ref()))) } +#[cfg(unix)] /// The session bus in `runtime`: the one named in the environment if it /// is there, else its usual socket there, if one exists. fn session_bus(runtime: &Path) -> Option { @@ -115,6 +136,7 @@ fn session_bus(runtime: &Path) -> Option { } } +#[cfg(unix)] /// The ssh agent's socket named in `named`, if it is a socket this /// account owns at an absolute path (INVARIANT 1). fn own_agent(named: Option<&std::ffi::OsStr>) -> Option { @@ -126,6 +148,7 @@ fn own_agent(named: Option<&std::ffi::OsStr>) -> Option { .then(|| path.as_os_str().to_os_string()) } +#[cfg(unix)] /// The host's environment (INVARIANT 1), given the opener's /// `SSH_AUTH_SOCK`. fn host_env(home: &Path, ssh_agent: Option<&std::ffi::OsStr>) -> Vec<(String, std::ffi::OsString)> { @@ -145,6 +168,7 @@ fn host_env(home: &Path, ssh_agent: Option<&std::ffi::OsStr>) -> Vec<(String, st env } +#[cfg(unix)] /// Whether this process sits in a unit under this account's user manager /// (an app's scope or service, as uwsm makes for every app), whose stop /// would take the host with it; going by the unified cgroup line of @@ -160,6 +184,7 @@ pub fn in_app_scope(cgroup: &str, uid: u32) -> bool { }) } +#[cfg(unix)] /// systemd-run, if this process is in an app scope and the user manager /// can be reached (INVARIANT 3). fn systemd_run() -> Option<&'static str> { @@ -173,6 +198,7 @@ fn systemd_run() -> Option<&'static str> { SYSTEMD_RUN.into_iter().find(|p| Path::new(p).is_file()) } +#[cfg(unix)] /// Start a host for the root (`root`, else the one the host finds /// itself), with these knobs and these server flags (INVARIANTS 1-5). pub fn start_host( @@ -266,9 +292,11 @@ pub fn start_host( Ok(()) } +#[cfg(unix)] /// How long systemd-run has to make its scope and run the host. const SCOPE_WAIT: Duration = Duration::from_secs(2); +#[cfg(unix)] /// Whether systemd-run made its scope and ran the host (INVARIANT 3): the /// host's first line is in its log after `from`, or systemd-run is still /// running when SCOPE_WAIT is up (it became the host). Only systemd-run @@ -298,6 +326,111 @@ fn scope_made(child: &mut std::process::Child, log: &std::fs::File, from: u64) - } } +/// The host's environment on Windows (INVARIANT 1's counterpart): the +/// system's own variables, the account's profile, and a PATH fixed to the +/// system folders and Git for Windows (the backups run `git`). Nothing +/// else of the opener's passes. (CHOICE, P11c: no authority states a +/// Windows list; this is the least a Windows program and `git` need.) +#[cfg(windows)] +fn host_env(home: &Path) -> Vec<(String, std::ffi::OsString)> { + let var = |name: &str| std::env::var_os(name); + let system_root: PathBuf = var("SystemRoot") + .or_else(|| var("windir")) + .map_or_else(|| PathBuf::from(r"C:\Windows"), PathBuf::from); + let program_files: PathBuf = + var("ProgramFiles").map_or_else(|| PathBuf::from(r"C:\Program Files"), PathBuf::from); + let path = std::env::join_paths([ + system_root.join("System32"), + system_root.clone(), + program_files.join("Git").join("cmd"), + ]) + .unwrap_or_default(); + let mut env: Vec<(String, std::ffi::OsString)> = vec![ + ("PATH".into(), path), + ("SystemRoot".into(), system_root.clone().into_os_string()), + ("windir".into(), system_root.into_os_string()), + ("USERPROFILE".into(), home.as_os_str().to_os_string()), + ("HOME".into(), home.as_os_str().to_os_string()), + ("TEMP".into(), std::env::temp_dir().into_os_string()), + ("TMP".into(), std::env::temp_dir().into_os_string()), + ]; + if let Some(local) = ledger_store::winsec::local_app_data() { + env.push(("LOCALAPPDATA".into(), local.into_os_string())); + } + for name in ["APPDATA", "ComSpec", "PATHEXT", "ProgramFiles"] { + if let Some(value) = var(name) { + env.push((name.into(), value)); + } + } + env +} + +/// Start a host for the root (`root`, else the one the host finds +/// itself), with these knobs and these server flags (INVARIANTS 1, 2, 4 +/// and 5; there is no scope to leave, but a job to break away from). +#[cfg(windows)] +pub fn start_host( + root: Option<&Path>, + knobs: &Knobs, + server_flags: &[String], +) -> Result<(), String> { + use std::os::windows::process::CommandExt; + let found = ledger_store::root::Root::find(root).map_err(|e| e.to_string())?; + let run = found.run_dir().map_err(|e| e.to_string())?; + let home = ledger_store::root::account_home().map_err(|e| e.to_string())?; + let exe = this_executable()?; + let mut args: Vec = vec!["host".into()]; + if root.is_some() { + args.push("--root".into()); + args.push(found.path().as_os_str().to_os_string()); + } + args.extend(knobs.args().into_iter().map(Into::into)); + args.extend(server_flags.iter().map(Into::into)); + let log = || { + let path = run.join(HOST_LOG); + if std::fs::metadata(&path).is_ok_and(|m| m.len() > LOG_LIMIT) + && !crate::opener::host_lock_held(&run) + { + let _ = std::fs::rename(&path, path.with_extension("log.1")); + } + std::fs::OpenOptions::new() + .create(true) + .read(true) + .append(true) + .open(&path) + .map_err(|e| format!("can't open the host's log: {}", e.kind())) + }; + let spawn = |flags: u32| -> Result { + Command::new(&exe) + .args(&args) + .env_clear() + .envs(host_env(&home)) + .current_dir(&run) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(log()?) + .creation_flags(flags) + .spawn() + .map_err(|e| { + // Named for the caller below: breakaway refused. + if e.kind() == std::io::ErrorKind::PermissionDenied { + "denied".to_string() + } else { + format!("can't start the books' host: {e}") + } + }) + }; + let child = match spawn(DETACHED_HOST | BREAKAWAY) { + Ok(child) => child, + // The job the opener runs in doesn't let a process leave it: the + // host starts inside it, as a host without a scope does on Linux. + Err(why) if why == "denied" => spawn(DETACHED_HOST)?, + Err(why) => return Err(why), + }; + reap(child); + Ok(()) +} + /// Reap a started host when it ends, without waiting for it (INVARIANT 5). fn reap(mut child: std::process::Child) { let _ = std::thread::Builder::new() @@ -307,7 +440,7 @@ fn reap(mut child: std::process::Child) { }); } -#[cfg(test)] +#[cfg(all(test, unix))] mod tests { use super::*; use std::os::unix::fs::PermissionsExt; @@ -457,3 +590,44 @@ mod tests { assert!(exe.is_file(), "{exe:?}"); } } + +#[cfg(all(test, windows))] +mod windows_tests { + use super::*; + + #[test] + fn this_executable_is_a_file_on_disk() { + let exe = this_executable().unwrap(); + assert!(exe.is_file(), "{exe:?}"); + } + + #[test] + fn the_host_environment_carries_nothing_of_the_openers_choosing() { + let env = host_env(Path::new(r"C:\Users\someone")); + let names: Vec<&str> = env.iter().map(|(n, _)| n.as_str()).collect(); + for n in &names { + assert!( + [ + "PATH", + "SystemRoot", + "windir", + "USERPROFILE", + "HOME", + "TEMP", + "TMP", + "LOCALAPPDATA", + "APPDATA", + "ComSpec", + "PATHEXT", + "ProgramFiles" + ] + .contains(n), + "{n}" + ); + } + let home = env.iter().find(|(n, _)| n == "HOME").unwrap(); + assert_eq!(home.1, r"C:\Users\someone"); + let path = env.iter().find(|(n, _)| n == "PATH").unwrap(); + assert!(path.1.to_string_lossy().to_lowercase().contains("system32")); + } +} diff --git a/crates/ledger-host/src/tokens.rs b/crates/ledger-host/src/tokens.rs index 6bb90c4..b18cd51 100644 --- a/crates/ledger-host/src/tokens.rs +++ b/crates/ledger-host/src/tokens.rs @@ -125,6 +125,9 @@ impl Keeper { std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700)) .map_err(refused)?; } + // Windows: this account's alone, and what is made in it too. + #[cfg(windows)] + ledger_store::winsec::set_owner_only(&self.0).map_err(refused)?; let name = account.file_name(); let tmp = self.0.join(format!("{name}.tmp{}", std::process::id())); let _ = std::fs::remove_file(&tmp); diff --git a/crates/ledger-host/tests/windows_pipe.rs b/crates/ledger-host/tests/windows_pipe.rs new file mode 100644 index 0000000..31abf29 --- /dev/null +++ b/crates/ledger-host/tests/windows_pipe.rs @@ -0,0 +1,53 @@ +//! The books' named pipes on Windows (P11c, D-62, plan 3.9): what keeps a +//! second account out of them, in the two ways the design has (the pipe's +//! DACL, and the SID check of the process at the other end). Windows job +//! only. +#![cfg(windows)] + +use ledger_mcp::ipc::{Endpoint, Listener}; +use ledger_store::winsec; +use std::path::Path; + +/// The process id of `image` (a system program), if `tasklist` finds it. +fn pid_of(image: &str) -> Option { + let out = std::process::Command::new("tasklist") + .args(["/FI", &format!("IMAGENAME eq {image}"), "/FO", "CSV", "/NH"]) + .output() + .ok()?; + let text = String::from_utf8_lossy(&out.stdout); + // "services.exe","700","Services","0","12,345 K" + text.lines() + .find(|l| l.to_lowercase().contains(&image.to_lowercase()))? + .split(',') + .nth(1)? + .trim_matches('"') + .parse() + .ok() +} + +#[tokio::test] +async fn a_second_user_cannot_open_the_pipe() { + let dir = tempfile::tempdir().unwrap(); + let endpoint = Endpoint::new(dir.path(), "lease.sock").unwrap(); + let _listener = Listener::bind(&endpoint).unwrap(); + + // The pipe's own DACL names this account and no one else: Everyone, + // Users and Authenticated Users are not on it, so the system itself + // refuses another account's open. + let pipe = Path::new(endpoint.pipe_name()); + assert!( + winsec::is_private(pipe).unwrap(), + "the pipe is open to more than this account" + ); + + // And a process of another account is never taken for this one: this + // account's own is, and a system service (SYSTEM) is not. + assert!(winsec::pid_is_us(std::process::id())); + if let Some(pid) = pid_of("services.exe") { + assert!( + !winsec::pid_is_us(pid), + "a SYSTEM process ({pid}) was taken for this account" + ); + } + assert!(!winsec::pid_is_us(0)); +} diff --git a/crates/ledger-mcp/src/admin.rs b/crates/ledger-mcp/src/admin.rs index a5cbe12..696db43 100644 --- a/crates/ledger-mcp/src/admin.rs +++ b/crates/ledger-mcp/src/admin.rs @@ -521,14 +521,12 @@ impl std::fmt::Display for CallError { /// Ask the server on `socket` (the root's `run/admin.sock`), waiting as /// long as it takes (a create or an adoption). -#[cfg(unix)] pub fn call(socket: &std::path::Path, request: Request) -> Result { call_within(socket, request, None) } /// [`call`], waiting at most `wait` for the answer (a quick look, such /// as `doctor`'s). -#[cfg(unix)] pub fn call_within( socket: &std::path::Path, request: Request, @@ -574,15 +572,14 @@ pub fn call_within( .and_then(|n| n.to_str()) .unwrap_or_default(), ); - let short = ledger_store::root::SocketPath::new(dir, name) + let endpoint = crate::ipc::Endpoint::new(dir, name) .map_err(|e| not_sent(format!("can't reach the books server: {e}")))?; - let stream = - std::os::unix::net::UnixStream::connect(short.path()).map_err(|e| match e.kind() { - std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused => { - not_sent("the books server isn't running".to_string()) - } - _ => not_sent(format!("can't reach the books server: {}", e.kind())), - })?; + let stream = crate::ipc::connect_std(&endpoint).map_err(|e| match e.kind() { + std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused => { + not_sent("the books server isn't running".to_string()) + } + _ => not_sent(format!("can't reach the books server: {}", e.kind())), + })?; crate::local::check_peer_std(&stream).map_err(not_sent)?; stream .set_read_timeout(wait) @@ -896,26 +893,38 @@ mod tests { ); } - /// A fake books server on the socket `name` in `dir`: it takes one + /// A fake books server on the channel `name` in `dir`: it takes one /// request and answers `answer`, byte for byte. - #[cfg(unix)] fn answering( dir: &std::path::Path, name: &str, answer: Vec, ) -> (std::path::PathBuf, std::thread::JoinHandle<()>) { let socket = dir.join(name); - let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap(); + let endpoint = crate::ipc::Endpoint::new(dir, name).unwrap(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); let server = std::thread::spawn(move || { - use std::io::{BufRead, Write}; - let (stream, _) = listener.accept().unwrap(); - let mut request = String::new(); - std::io::BufReader::new(&stream) - .read_line(&mut request) + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() .unwrap(); - // A client that stops reading part way breaks the pipe: fine. - let _ = (&stream).write_all(&answer); + rt.block_on(async move { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; + let mut listener = crate::ipc::Listener::bind(&endpoint).unwrap(); + ready_tx.send(()).unwrap(); + let stream = listener.accept().await.unwrap(); + let (read, mut write) = crate::ipc::split_accepted(stream); + let mut request = String::new(); + tokio::io::BufReader::new(read) + .read_line(&mut request) + .await + .unwrap(); + // A client that stops reading part way breaks the pipe: fine. + let _ = write.write_all(&answer).await; + let _ = write.flush().await; + }); }); + ready_rx.recv().unwrap(); (socket, server) } @@ -931,7 +940,6 @@ mod tests { /// newline not counted, is taken whole (settings texts ride in them); /// a byte more is refused, as is an answer cut off before its /// newline, even when what came parses. - #[cfg(unix)] #[test] fn an_answer_is_one_line_of_at_most_four_mib() { let four_mib = 4 * 1024 * 1024; @@ -956,7 +964,6 @@ mod tests { /// No server on the socket: the request never left, and the command /// line says the server isn't running, whether there is no socket or /// one a stopped server left behind. - #[cfg(unix)] #[test] fn no_server_means_nothing_was_sent() { let dir = tempfile::tempdir().unwrap(); @@ -965,17 +972,21 @@ mod tests { call(&dir.path().join("admin.sock"), Request::List), not_running ); - let stale = dir.path().join("stale.sock"); - drop(std::os::unix::net::UnixListener::bind(&stale).unwrap()); - assert_eq!( - call( - &stale, - Request::Create { - id: "acme".into(), - currency: "USD".into(), - } - ), - not_running - ); + // (A pipe has no stale file to leave behind: only a socket does.) + #[cfg(unix)] + { + let stale = dir.path().join("stale.sock"); + drop(std::os::unix::net::UnixListener::bind(&stale).unwrap()); + assert_eq!( + call( + &stale, + Request::Create { + id: "acme".into(), + currency: "USD".into(), + } + ), + not_running + ); + } } } diff --git a/crates/ledger-mcp/src/ipc.rs b/crates/ledger-mcp/src/ipc.rs new file mode 100644 index 0000000..6e589d4 --- /dev/null +++ b/crates/ledger-mcp/src/ipc.rs @@ -0,0 +1,793 @@ +//! The local transport, one seam for both families of system (plan 3.9, +//! D-62): Unix sockets in the root's `run/` on Unix, named pipes on +//! Windows, each reachable by this account alone. +//! +//! ```text +//! GOAL: the books server, the host and everything that talks to them +//! meet on a channel that only this account can open, and each end +//! checks that the other end is this account. +//! SOURCE: plan/served.md 3.9 ("the local transport, the peer check": +//! 0600 sockets plus the peer's uid; on Windows a pipe with an +//! owner-only DACL plus the peer process's SID), D-32 (nothing +//! here confines a process), D-62. +//! INVARIANTS: +//! 1. A listener is made for this account alone: Unix sockets 0600; a +//! Windows pipe with a DACL of one allow entry (this account's SID) +//! and remote clients refused. +//! 2. [`Peer::peer_is_us`] is true only when the system says the other +//! end is this account (the effective uid; the user SID of the +//! process at the other end of the pipe); any doubt is false. +//! 3. A Windows pipe's name carries a hash of the root's `run/` folder, +//! so two roots on one machine never share a pipe, and the first +//! instance can be made only once (a second server is refused). +//! 4. Windows keeps idle pipe instances waiting, so a client never +//! finds every instance busy while the server makes the next one. +//! 5. A read with a time limit gives up at the limit with `TimedOut` +//! (Unix: `WouldBlock`); a pipe whose other end has gone reads as +//! the end of the stream. +//! ``` + +use std::io; + +pub use sys::*; + +/// Who is at the other end of a connection. +pub trait Peer { + /// Whether the other end is this account (INVARIANT 2). + fn peer_is_us(&self) -> bool; + /// The other end's process, where the system says it. + fn peer_pid(&self) -> Option; +} + +/// Whether the other end of `stream` is this account. +pub fn peer_is_us(stream: &impl Peer) -> bool { + stream.peer_is_us() +} + +/// The other end's process, where the system says. +pub fn peer_pid(stream: &impl Peer) -> Option { + stream.peer_pid() +} + +/// One of the signals that ask a program to stop: on Unix SIGTERM, +/// SIGINT and SIGHUP; on Windows Ctrl+C, Ctrl+Break, closing the console +/// and shutdown. +pub struct StopSignal(sys::RawStop); + +impl StopSignal { + /// The next time it arrives; `None` once it can't arrive again. + pub async fn recv(&mut self) -> Option<()> { + self.0.recv().await + } +} + +/// Watch for every signal that asks a program to stop. +pub fn stop_signals() -> io::Result> { + sys::raw_stops().map(|all| all.into_iter().map(StopSignal).collect()) +} + +#[cfg(unix)] +mod sys { + use super::Peer; + use std::io; + use std::os::fd::AsRawFd; + use std::path::{Path, PathBuf}; + + /// A connection a listener accepted. + pub type Accepted = tokio::net::UnixStream; + /// A connection this side made. + pub type Connected = tokio::net::UnixStream; + /// A connection this side made, for blocking code. + pub type StdStream = std::os::unix::net::UnixStream; + /// What [`Listener::addr_of`] says of a connection. + pub type Addr = tokio::net::unix::SocketAddr; + /// The reading half of an accepted connection. + pub type AcceptedRead = tokio::net::unix::OwnedReadHalf; + /// The writing half of an accepted connection. + pub type AcceptedWrite = tokio::net::unix::OwnedWriteHalf; + /// The reading half of a connection this side made. + pub type ConnectedRead = tokio::net::unix::OwnedReadHalf; + /// The writing half of a connection this side made. + pub type ConnectedWrite = tokio::net::unix::OwnedWriteHalf; + + /// Split an accepted connection into its halves. + pub fn split_accepted(stream: Accepted) -> (AcceptedRead, AcceptedWrite) { + stream.into_split() + } + + /// Split a connection this side made into its halves. + pub fn split_connected(stream: Connected) -> (ConnectedRead, ConnectedWrite) { + stream.into_split() + } + + /// Where a channel is: a socket `name` in the root's `run/`. + #[derive(Debug)] + pub struct Endpoint { + full: PathBuf, + short: ledger_store::root::SocketPath, + } + + impl Endpoint { + /// The channel `name` in the root's `run/` folder `dir`. + pub fn new(dir: &Path, name: &str) -> io::Result { + Ok(Endpoint { + full: dir.join(name), + short: ledger_store::root::SocketPath::new(dir, name)?, + }) + } + + /// The path to bind or connect to (good while this lives). + pub fn path(&self) -> &Path { + self.short.path() + } + } + + /// A listener that is this account's alone (INVARIANT 1). + pub struct Listener(tokio::net::UnixListener); + + impl Listener { + /// Bind `endpoint`, replacing a stale socket (only a server that + /// holds its lock gets here). + pub fn bind(endpoint: &Endpoint) -> io::Result { + match std::fs::remove_file(&endpoint.full) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + let listener = tokio::net::UnixListener::bind(endpoint.path())?; + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&endpoint.full, std::fs::Permissions::from_mode(0o600))?; + Ok(Listener(listener)) + } + + /// The next connection, whoever it is from. + pub async fn accept(&mut self) -> io::Result { + self.0.accept().await.map(|(stream, _)| stream) + } + + /// The address of an accepted connection. + pub fn addr_of(stream: &Accepted) -> io::Result { + stream.peer_addr() + } + + /// The address this listener is bound to. + pub fn local_addr(&self) -> io::Result { + self.0.local_addr() + } + } + + /// Connect to `endpoint`. + pub async fn connect(endpoint: &Endpoint) -> io::Result { + tokio::net::UnixStream::connect(endpoint.path()).await + } + + /// Connect to `endpoint`, for blocking code. + pub fn connect_std(endpoint: &Endpoint) -> io::Result { + std::os::unix::net::UnixStream::connect(endpoint.path()) + } + + /// The user ID at the other end of a Unix socket. + pub fn peer_uid(fd: &impl AsRawFd) -> io::Result { + let fd = fd.as_raw_fd(); + #[cfg(any(target_os = "linux", target_os = "android"))] + { + let mut cred = libc::ucred { + pid: 0, + uid: 0, + gid: 0, + }; + let mut len = std::mem::size_of::() as libc::socklen_t; + // SAFETY: `cred` and `len` are valid for writes of the sizes given. + let rc = unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_PEERCRED, + (&mut cred as *mut libc::ucred).cast(), + &mut len, + ) + }; + if rc != 0 { + return Err(io::Error::last_os_error()); + } + if len as usize != std::mem::size_of::() { + return Err(io::Error::other("the peer's credentials came back short")); + } + Ok(cred.uid) + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + let mut uid: libc::uid_t = 0; + let mut gid: libc::gid_t = 0; + // SAFETY: `uid` and `gid` are valid for writes. + if unsafe { libc::getpeereid(fd, &mut uid, &mut gid) } != 0 { + return Err(io::Error::last_os_error()); + } + Ok(uid) + } + } + + /// This process's user ID. + pub fn own_uid() -> u32 { + // SAFETY: geteuid has no preconditions and can't fail. + unsafe { libc::geteuid() } + } + + fn pid_of(fd: &impl AsRawFd) -> Option { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + let mut cred = libc::ucred { + pid: 0, + uid: 0, + gid: 0, + }; + let mut len = std::mem::size_of::() as libc::socklen_t; + // SAFETY: `cred` and `len` are valid for writes of the sizes given. + let rc = unsafe { + libc::getsockopt( + fd.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_PEERCRED, + (&mut cred as *mut libc::ucred).cast(), + &mut len, + ) + }; + (rc == 0 && len as usize == std::mem::size_of::() && cred.pid > 0) + .then_some(cred.pid as u32) + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + let _ = fd; + None + } + } + + impl Peer for T { + fn peer_is_us(&self) -> bool { + peer_uid(self).is_ok_and(|uid| uid == own_uid()) + } + + fn peer_pid(&self) -> Option { + pid_of(self) + } + } + + /// One stop signal. + pub struct RawStop(tokio::signal::unix::Signal); + + impl RawStop { + pub(super) async fn recv(&mut self) -> Option<()> { + self.0.recv().await + } + } + + pub(super) fn raw_stops() -> io::Result> { + use tokio::signal::unix::{SignalKind, signal}; + [ + SignalKind::terminate(), + SignalKind::interrupt(), + SignalKind::hangup(), + ] + .into_iter() + .map(|kind| signal(kind).map(RawStop)) + .collect() + } +} + +#[cfg(windows)] +mod sys { + use super::Peer; + use ledger_store::winsec; + use std::io; + use std::os::windows::fs::OpenOptionsExt; + use std::path::Path; + use std::sync::{Arc, Mutex}; + use std::time::{Duration, Instant}; + use tokio::net::windows::named_pipe::{ + ClientOptions, NamedPipeClient, NamedPipeServer, ServerOptions, + }; + use tokio::sync::mpsc; + + /// ERROR_PIPE_BUSY: every instance of the pipe is in use for now. + const PIPE_BUSY: i32 = 231; + /// ERROR_BROKEN_PIPE: the other end has gone. + const BROKEN_PIPE: i32 = 109; + /// SECURITY_IDENTIFICATION: the server may learn who the client is + /// and may not act as the client. + const IDENTIFICATION: u32 = 0x0001_0000; + /// How long a client waits for a busy pipe (INVARIANT 4). + const BUSY_WAIT: Duration = Duration::from_secs(3); + /// Instances kept waiting for a client (INVARIANT 4). + const IDLE_INSTANCES: usize = 16; + + /// A connection a listener accepted. + pub type Accepted = NamedPipeServer; + /// A connection this side made. + pub type Connected = NamedPipeClient; + /// What [`Listener::addr_of`] says of a connection: nothing. + pub type Addr = (); + /// The reading half of an accepted connection. + pub type AcceptedRead = tokio::io::ReadHalf; + /// The writing half of an accepted connection. + pub type AcceptedWrite = tokio::io::WriteHalf; + /// The reading half of a connection this side made. + pub type ConnectedRead = tokio::io::ReadHalf; + /// The writing half of a connection this side made. + pub type ConnectedWrite = tokio::io::WriteHalf; + + /// Split an accepted connection into its halves. + pub fn split_accepted(stream: Accepted) -> (AcceptedRead, AcceptedWrite) { + tokio::io::split(stream) + } + + /// Split a connection this side made into its halves. + pub fn split_connected(stream: Connected) -> (ConnectedRead, ConnectedWrite) { + tokio::io::split(stream) + } + + /// Where a channel is: the pipe `\\.\pipe\wanderbooks--` (INVARIANT 3). + #[derive(Debug, Clone)] + pub struct Endpoint { + name: String, + } + + impl Endpoint { + /// The channel `name` for the root's `run/` folder `dir`. + pub fn new(dir: &Path, name: &str) -> io::Result { + use sha2::{Digest, Sha256}; + // The folder as the system names it (a short 8.3 name and its + // long one are one folder; every process must hash the same + // text), without the `\\?\` prefix canonicalizing adds; a + // folder name is not case-sensitive here. + let dir = std::fs::canonicalize(dir).or_else(|_| std::path::absolute(dir))?; + let text = dir.to_string_lossy().to_lowercase(); + let key = text.strip_prefix(r"\\?\").unwrap_or(&text); + let digest = Sha256::digest(key.as_bytes()); + let hash: String = digest[..8].iter().map(|b| format!("{b:02x}")).collect(); + let name = name.strip_suffix(".sock").unwrap_or(name); + Ok(Endpoint { + name: format!(r"\\.\pipe\wanderbooks-{hash}-{name}"), + }) + } + + /// The pipe's name. + pub fn pipe_name(&self) -> &str { + &self.name + } + } + + fn make(name: &str, security: &winsec::OwnerOnly, first: bool) -> io::Result { + let mut options = ServerOptions::new(); + options + .first_pipe_instance(first) + .reject_remote_clients(true); + // SAFETY: the pointer is to security attributes that `security` + // keeps alive for the call, which copies them. + unsafe { options.create_with_security_attributes_raw(name, security.as_raw()) } + } + + /// A listener that is this account's alone (INVARIANT 1): a pool of + /// idle instances (INVARIANT 4), each handing its connection on when + /// a client comes. + pub struct Listener { + accepted: mpsc::UnboundedReceiver>, + tasks: Vec>, + } + + impl Drop for Listener { + fn drop(&mut self) { + for task in &self.tasks { + task.abort(); + } + } + } + + impl Listener { + /// Make the pipe. Refused when another server already has it + /// (INVARIANT 3). + pub fn bind(endpoint: &Endpoint) -> io::Result { + let security = Arc::new(winsec::OwnerOnly::new()?); + let first = make(&endpoint.name, &security, true)?; + let (tx, accepted) = mpsc::unbounded_channel(); + let mut tasks = Vec::new(); + let mut instance = Some(first); + for _ in 0..IDLE_INSTANCES { + let server = match instance.take() { + Some(server) => server, + None => make(&endpoint.name, &security, false)?, + }; + tasks.push(tokio::spawn(wait_for_clients( + server, + endpoint.name.clone(), + security.clone(), + tx.clone(), + ))); + } + Ok(Listener { accepted, tasks }) + } + + /// The next connection, whoever it is from. + pub async fn accept(&mut self) -> io::Result { + match self.accepted.recv().await { + Some(result) => result, + None => Err(io::Error::from(io::ErrorKind::BrokenPipe)), + } + } + + /// The address of an accepted connection: none. + pub fn addr_of(_stream: &Accepted) -> io::Result { + Ok(()) + } + + /// The address this listener is bound to: none. + pub fn local_addr(&self) -> io::Result { + Ok(()) + } + } + + /// One idle instance: wait for a client, put a fresh instance in its + /// place, and hand the connected one on. + async fn wait_for_clients( + mut server: NamedPipeServer, + name: String, + security: Arc, + tx: mpsc::UnboundedSender>, + ) { + loop { + if let Err(e) = server.connect().await { + // This instance is no good: say so, and put a fresh one in + // its place (nothing connected is handed on). + if tx.send(Err(e)).is_err() { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + if let Ok(fresh) = make(&name, &security, false) { + server = fresh; + } + continue; + } + let next = loop { + match make(&name, &security, false) { + Ok(next) => break next, + Err(e) => { + if tx.send(Err(e)).is_err() { + return; + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + }; + let connected = std::mem::replace(&mut server, next); + if tx.send(Ok(connected)).is_err() { + return; + } + } + } + + /// Connect to `endpoint`, waiting a little while every instance is + /// busy (INVARIANT 4). + pub async fn connect(endpoint: &Endpoint) -> io::Result { + let deadline = Instant::now() + BUSY_WAIT; + loop { + match ClientOptions::new() + .security_qos_flags(IDENTIFICATION) + .open(&endpoint.name) + { + Ok(client) => return Ok(client), + Err(e) if e.raw_os_error() == Some(PIPE_BUSY) && Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + Err(e) => return Err(e), + } + } + } + + /// A connection this side made, for blocking code. + #[derive(Debug)] + pub struct StdStream { + file: std::fs::File, + read_timeout: Mutex>, + } + + impl StdStream { + /// Give up reading after `timeout` (INVARIANT 5); `None` waits + /// for ever. + pub fn set_read_timeout(&self, timeout: Option) -> io::Result<()> { + *self.read_timeout.lock().unwrap_or_else(|e| e.into_inner()) = timeout; + Ok(()) + } + + fn read_some(&self, buf: &mut [u8]) -> io::Result { + let timeout = *self.read_timeout.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(timeout) = timeout { + // A blocking read can't be cut short, so it is made only + // when something is waiting. + let deadline = Instant::now() + timeout; + loop { + match winsec::pipe_bytes_waiting(&self.file) { + Ok(0) => {} + Ok(_) => break, + Err(e) if e.raw_os_error() == Some(BROKEN_PIPE) => return Ok(0), + Err(e) => return Err(e), + } + if Instant::now() >= deadline { + return Err(io::Error::from(io::ErrorKind::TimedOut)); + } + std::thread::sleep(Duration::from_millis(2)); + } + } + match (&self.file).read(buf) { + Err(e) if e.raw_os_error() == Some(BROKEN_PIPE) => Ok(0), + other => other, + } + } + } + + use std::io::{Read, Write}; + + impl Read for StdStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.read_some(buf) + } + } + + impl Read for &StdStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.read_some(buf) + } + } + + impl Write for StdStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + (&self.file).write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + (&self.file).flush() + } + } + + impl Write for &StdStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + (&self.file).write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + (&self.file).flush() + } + } + + /// Connect to `endpoint`, for blocking code, waiting a little while + /// every instance is busy (INVARIANT 4). + pub fn connect_std(endpoint: &Endpoint) -> io::Result { + let deadline = Instant::now() + BUSY_WAIT; + loop { + match std::fs::OpenOptions::new() + .read(true) + .write(true) + .security_qos_flags(IDENTIFICATION) + .open(&endpoint.name) + { + Ok(file) => { + return Ok(StdStream { + file, + read_timeout: Mutex::new(None), + }); + } + Err(e) if e.raw_os_error() == Some(PIPE_BUSY) && Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(10)); + } + Err(e) => return Err(e), + } + } + } + + impl Peer for NamedPipeServer { + fn peer_is_us(&self) -> bool { + self.peer_pid().is_some_and(winsec::pid_is_us) + } + + fn peer_pid(&self) -> Option { + winsec::pipe_client_pid(self).ok() + } + } + + impl Peer for NamedPipeClient { + fn peer_is_us(&self) -> bool { + self.peer_pid().is_some_and(winsec::pid_is_us) + } + + fn peer_pid(&self) -> Option { + winsec::pipe_server_pid(self).ok() + } + } + + impl Peer for StdStream { + fn peer_is_us(&self) -> bool { + self.peer_pid().is_some_and(winsec::pid_is_us) + } + + fn peer_pid(&self) -> Option { + winsec::pipe_server_pid(&self.file).ok() + } + } + + /// One stop signal. + pub enum RawStop { + /// Ctrl+C. + CtrlC(tokio::signal::windows::CtrlC), + /// Ctrl+Break. + CtrlBreak(tokio::signal::windows::CtrlBreak), + /// The console window is closed. + CtrlClose(tokio::signal::windows::CtrlClose), + /// The system is shutting down. + CtrlShutdown(tokio::signal::windows::CtrlShutdown), + } + + impl RawStop { + pub(super) async fn recv(&mut self) -> Option<()> { + match self { + RawStop::CtrlC(s) => s.recv().await, + RawStop::CtrlBreak(s) => s.recv().await, + RawStop::CtrlClose(s) => s.recv().await, + RawStop::CtrlShutdown(s) => s.recv().await, + } + } + } + + pub(super) fn raw_stops() -> io::Result> { + use tokio::signal::windows; + Ok(vec![ + RawStop::CtrlC(windows::ctrl_c()?), + RawStop::CtrlBreak(windows::ctrl_break()?), + RawStop::CtrlClose(windows::ctrl_close()?), + RawStop::CtrlShutdown(windows::ctrl_shutdown()?), + ]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; + + fn endpoint(dir: &tempfile::TempDir) -> Endpoint { + Endpoint::new(dir.path(), "test.sock").unwrap() + } + + /// Both ends of a connection are this account, and name this process. + #[tokio::test] + async fn a_listener_and_a_client_talk_and_each_knows_the_other_is_us() { + let dir = tempfile::tempdir().unwrap(); + let ep = endpoint(&dir); + let mut listener = Listener::bind(&ep).unwrap(); + let client = connect(&ep).await.unwrap(); + let server = listener.accept().await.unwrap(); + assert!(peer_is_us(&server)); + assert!(peer_is_us(&client)); + #[cfg(any(windows, target_os = "linux"))] + { + assert_eq!(peer_pid(&server), Some(std::process::id())); + assert_eq!(peer_pid(&client), Some(std::process::id())); + } + let (read, mut write) = split_accepted(server); + let (cread, mut cwrite) = split_connected(client); + cwrite.write_all(b"hello\n").await.unwrap(); + let mut line = String::new(); + tokio::io::BufReader::new(read) + .read_line(&mut line) + .await + .unwrap(); + assert_eq!(line, "hello\n"); + write.write_all(b"back\n").await.unwrap(); + let mut line = String::new(); + tokio::io::BufReader::new(cread) + .read_line(&mut line) + .await + .unwrap(); + assert_eq!(line, "back\n"); + } + + /// A blocking client gives up at its limit, and reads the end of the + /// stream once the server has gone (INVARIANT 5). + #[tokio::test] + async fn a_blocking_client_times_out_and_reads_the_end_when_the_server_goes() { + use std::io::Read; + let dir = tempfile::tempdir().unwrap(); + let ep = endpoint(&dir); + let mut listener = Listener::bind(&ep).unwrap(); + let ep2 = ep_for(&dir); + let client = tokio::task::spawn_blocking(move || connect_std(&ep2).unwrap()) + .await + .unwrap(); + let server = listener.accept().await.unwrap(); + assert!(peer_is_us(&client)); + client + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let (client, kind) = tokio::task::spawn_blocking(move || { + let mut byte = [0u8; 1]; + let kind = (&client).read(&mut byte).unwrap_err().kind(); + (client, kind) + }) + .await + .unwrap(); + assert!( + matches!( + kind, + std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock + ), + "{kind:?}" + ); + drop(server); + let n = tokio::task::spawn_blocking(move || { + let mut byte = [0u8; 1]; + (&client).read(&mut byte).unwrap() + }) + .await + .unwrap(); + assert_eq!(n, 0, "the end of the stream"); + } + + fn ep_for(dir: &tempfile::TempDir) -> Endpoint { + Endpoint::new(dir.path(), "test.sock").unwrap() + } + + /// Many clients at once all get in (INVARIANT 4). + #[tokio::test] + async fn many_clients_at_once_all_connect() { + let dir = tempfile::tempdir().unwrap(); + let ep = endpoint(&dir); + let mut listener = Listener::bind(&ep).unwrap(); + let ep = std::sync::Arc::new(ep); + let mut clients = Vec::new(); + for _ in 0..48 { + let ep = ep.clone(); + clients.push(tokio::spawn(async move { connect(&ep).await })); + } + let mut accepted = Vec::new(); + for _ in 0..48 { + accepted.push( + tokio::time::timeout(Duration::from_secs(10), listener.accept()) + .await + .expect("accepted in time") + .unwrap(), + ); + } + for c in clients { + c.await.unwrap().unwrap(); + } + } + + /// Nothing listening: not found, so callers say "not running". + #[tokio::test] + async fn nothing_listening_is_not_found() { + let dir = tempfile::tempdir().unwrap(); + let err = connect(&endpoint(&dir)).await.unwrap_err(); + assert!( + matches!( + err.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused + ), + "{err:?}" + ); + } + + /// Two roots on one machine never share a pipe (INVARIANT 3), and a + /// second server cannot take the first one's pipe. + #[cfg(windows)] + #[tokio::test] + async fn a_pipe_is_per_root_and_cannot_be_taken_twice() { + let (a, b) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()); + assert_ne!(endpoint(&a).pipe_name(), endpoint(&b).pipe_name()); + assert!( + endpoint(&a) + .pipe_name() + .starts_with(r"\\.\pipe\wanderbooks-") + ); + let _first = Listener::bind(&endpoint(&a)).unwrap(); + assert!(Listener::bind(&endpoint(&a)).is_err()); + } +} diff --git a/crates/ledger-mcp/src/lib.rs b/crates/ledger-mcp/src/lib.rs index 8bff8e7..42f9278 100644 --- a/crates/ledger-mcp/src/lib.rs +++ b/crates/ledger-mcp/src/lib.rs @@ -9,7 +9,7 @@ pub mod access; pub mod admin; pub mod backup; mod handbook; -#[cfg(unix)] +pub mod ipc; pub mod local; pub mod opening; mod policy; diff --git a/crates/ledger-mcp/src/local.rs b/crates/ledger-mcp/src/local.rs index 4511b2e..abf1016 100644 --- a/crates/ledger-mcp/src/local.rs +++ b/crates/ledger-mcp/src/local.rs @@ -60,6 +60,7 @@ //! ``` use crate::admin::{self, Request, Response}; +use crate::ipc::{self, Peer}; use crate::server::{BooksServer, Isolation, Served, SharedStore}; use ledger_store::Store; use ledger_store::access::{Access, AccessLog, Outcome, Principal}; @@ -69,7 +70,6 @@ use rmcp::transport::streamable_http_server::{ StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, }; use std::collections::BTreeMap; -use std::os::fd::AsRawFd; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, RwLock}; @@ -994,110 +994,25 @@ fn moved_in_at_adoption( } } -/// The user ID at the other end of a Unix socket. -pub fn peer_uid(fd: &impl AsRawFd) -> std::io::Result { - let fd = fd.as_raw_fd(); - #[cfg(any(target_os = "linux", target_os = "android"))] - { - let mut cred = libc::ucred { - pid: 0, - uid: 0, - gid: 0, - }; - let mut len = std::mem::size_of::() as libc::socklen_t; - // SAFETY: `cred` and `len` are valid for writes of the sizes given. - let rc = unsafe { - libc::getsockopt( - fd, - libc::SOL_SOCKET, - libc::SO_PEERCRED, - (&mut cred as *mut libc::ucred).cast(), - &mut len, - ) - }; - if rc != 0 { - return Err(std::io::Error::last_os_error()); - } - if len as usize != std::mem::size_of::() { - return Err(std::io::Error::other( - "the peer's credentials came back short", - )); - } - Ok(cred.uid) - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - { - let mut uid: libc::uid_t = 0; - let mut gid: libc::gid_t = 0; - // SAFETY: `uid` and `gid` are valid for writes. - if unsafe { libc::getpeereid(fd, &mut uid, &mut gid) } != 0 { - return Err(std::io::Error::last_os_error()); - } - Ok(uid) - } -} - -/// The process at the other end of a Unix socket, where the system says -/// (Linux); `None` elsewhere or if it can't be read. -pub fn peer_pid(fd: &impl AsRawFd) -> Option { - #[cfg(any(target_os = "linux", target_os = "android"))] - { - let mut cred = libc::ucred { - pid: 0, - uid: 0, - gid: 0, - }; - let mut len = std::mem::size_of::() as libc::socklen_t; - // SAFETY: `cred` and `len` are valid for writes of the sizes given. - let rc = unsafe { - libc::getsockopt( - fd.as_raw_fd(), - libc::SOL_SOCKET, - libc::SO_PEERCRED, - (&mut cred as *mut libc::ucred).cast(), - &mut len, - ) - }; - (rc == 0 && len as usize == std::mem::size_of::() && cred.pid > 0) - .then_some(cred.pid as u32) - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - { - let _ = fd; - None - } -} - -/// This process's user ID. -fn own_uid() -> u32 { - // SAFETY: geteuid has no preconditions and can't fail. - unsafe { libc::geteuid() } -} - -/// Whether the other end of a socket is this account (INVARIANT 2). -pub fn peer_is_us(fd: &impl AsRawFd) -> bool { - peer_uid(fd).is_ok_and(|uid| uid == own_uid()) -} - /// The client's check of the server (plan 3.9: both ends check). -pub(crate) fn check_peer_std(stream: &std::os::unix::net::UnixStream) -> Result<(), String> { - if peer_is_us(stream) { +pub(crate) fn check_peer_std(stream: &ipc::StdStream) -> Result<(), String> { + if stream.peer_is_us() { Ok(()) } else { Err("the books server's socket belongs to another account: not used".into()) } } -/// A Unix listener that hands on only this account's connections +/// A listener that hands on only this account's connections /// (INVARIANT 2). -struct OwnerOnly(tokio::net::UnixListener); +struct OwnerOnly(ipc::Listener); impl OwnerOnly { - async fn next(&mut self) -> tokio::net::UnixStream { + async fn next(&mut self) -> ipc::Accepted { loop { match self.0.accept().await { - Ok((stream, _)) if peer_is_us(&stream) => return stream, - Ok((stream, _)) => { + Ok(stream) if stream.peer_is_us() => return stream, + Ok(stream) => { eprintln!("wanderbooks server: refused a connection from another account"); drop(stream); } @@ -1120,13 +1035,13 @@ impl OwnerOnly { } impl axum::serve::Listener for OwnerOnly { - type Io = tokio::net::UnixStream; - type Addr = tokio::net::unix::SocketAddr; + type Io = ipc::Accepted; + type Addr = ipc::Addr; async fn accept(&mut self) -> (Self::Io, Self::Addr) { loop { let stream = self.next().await; - if let Ok(addr) = stream.peer_addr() { + if let Ok(addr) = ipc::Listener::addr_of(&stream) { return (stream, addr); } } @@ -1818,9 +1733,9 @@ async fn not_found() -> (axum::http::StatusCode, &'static str) { } /// One admin connection: one request, one answer (admin INVARIANT 1). -async fn serve_admin(stream: tokio::net::UnixStream, books: Arc) { +async fn serve_admin(stream: ipc::Accepted, books: Arc) { use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; - let (read, mut write) = stream.into_split(); + let (read, mut write) = ipc::split_accepted(stream); let mut reader = tokio::io::BufReader::new(read.take(admin::MAX_LINE as u64 + 1)); let mut line = String::new(); // A request still arriving when closing begins is turned away. @@ -1869,18 +1784,8 @@ async fn serve_admin(stream: tokio::net::UnixStream, books: Arc) { /// Bind a socket in `run/`, replacing a stale one (INVARIANT 1: only the /// server holding the lock gets here). -fn bind(run: &Path, name: &str) -> std::io::Result { - let path = run.join(name); - match std::fs::remove_file(&path) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(e), - } - let short = ledger_store::root::SocketPath::new(run, name)?; - let listener = tokio::net::UnixListener::bind(short.path())?; - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?; - Ok(listener) +fn bind(run: &Path, name: &str) -> std::io::Result { + ipc::Listener::bind(&ipc::Endpoint::new(run, name)?) } /// The address `--http` takes: loopback only (plan 3.9, P9a). @@ -2025,8 +1930,14 @@ pub fn run( } { // Only once it is ours: a refused server touches nothing. - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&lock_path, std::fs::Permissions::from_mode(0o600)) + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&lock_path, std::fs::Permissions::from_mode(0o600)) + .map_err(|e| format!("can't make the server lock private: {}", e.kind()))?; + } + #[cfg(windows)] + ledger_store::winsec::set_owner_only(&lock_path) .map_err(|e| format!("can't make the server lock private: {}", e.kind()))?; } let books = Arc::new(Books { @@ -2292,7 +2203,6 @@ async fn serve( /// SIGTERM, SIGINT or SIGHUP; hurry on the next signal. Runs inside the /// runtime, before the books are opened. fn watch_for_close(stops: &Stops) -> std::io::Result<()> { - use tokio::signal::unix::{SignalKind, signal}; let on_eof = stops.close.clone(); // A plain thread: a blocked read must never hold up the runtime's end. std::thread::Builder::new() @@ -2311,12 +2221,7 @@ fn watch_for_close(stops: &Stops) -> std::io::Result<()> { } on_eof.cancel(); })?; - for kind in [ - SignalKind::terminate(), - SignalKind::interrupt(), - SignalKind::hangup(), - ] { - let mut signals = signal(kind)?; + for mut signals in ipc::stop_signals()? { let (close, hurry) = (stops.close.clone(), stops.hurry.clone()); tokio::spawn(async move { while signals.recv().await.is_some() { @@ -2613,34 +2518,38 @@ mod tests { assert_eq!(backup_due(Some(&w), 5, at(9 + retry * 3), quiet), None); } + #[cfg(unix)] #[test] fn a_socket_pair_is_this_account_at_both_ends() { let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); - assert_eq!(peer_uid(&a).unwrap(), own_uid()); - assert!(peer_is_us(&b)); + assert_eq!(ipc::peer_uid(&a).unwrap(), ipc::own_uid()); + assert!(ipc::peer_is_us(&b)); assert!(check_peer_std(&a).is_ok()); } + #[cfg(unix)] #[test] fn a_socket_pair_names_this_process() { let (a, _b) = std::os::unix::net::UnixStream::pair().unwrap(); if cfg!(target_os = "linux") { - assert_eq!(peer_pid(&a), Some(std::process::id())); + assert_eq!(ipc::peer_pid(&a), Some(std::process::id())); } let file = std::fs::File::open("/dev/null").unwrap(); - assert_eq!(peer_pid(&file), None); + assert_eq!(ipc::peer_pid(&file), None); } + #[cfg(unix)] #[test] fn a_descriptor_that_isnt_a_socket_is_never_taken_for_us() { let file = std::fs::File::open("/dev/null").unwrap(); - assert!(peer_uid(&file).is_err()); - assert!(!peer_is_us(&file)); + assert!(ipc::peer_uid(&file).is_err()); + assert!(!ipc::peer_is_us(&file)); } /// A socket with no other end: the system answers for it (Linux: no /// process, no user), and that answer names neither this process nor /// this account; the client's check refuses it. + #[cfg(unix)] #[test] fn a_socket_with_no_other_end_names_no_process_and_isnt_us() { use std::os::fd::FromRawFd; @@ -2650,8 +2559,8 @@ mod tests { assert!(fd >= 0); // SAFETY: a fresh descriptor of a Unix stream socket, owned by none. let lone = unsafe { std::os::unix::net::UnixStream::from_raw_fd(fd) }; - assert_eq!(peer_pid(&lone), None); - assert!(!peer_is_us(&lone)); + assert_eq!(ipc::peer_pid(&lone), None); + assert!(!ipc::peer_is_us(&lone)); assert_eq!( check_peer_std(&lone).unwrap_err(), "the books server's socket belongs to another account: not used" @@ -3196,6 +3105,7 @@ mod tests { let id = BookId::parse(id).unwrap(); let dir = books.root.book_dir(&id); std::fs::create_dir(&dir).unwrap(); + #[cfg(unix)] std::fs::set_permissions(&dir, std::os::unix::fs::PermissionsExt::from_mode(0o700)) .unwrap(); let mut s = diff --git a/crates/ledger-store/src/backup.rs b/crates/ledger-store/src/backup.rs index c2d9b5b..3c3d02f 100644 --- a/crates/ledger-store/src/backup.rs +++ b/crates/ledger-store/src/backup.rs @@ -124,11 +124,20 @@ impl Git { .unwrap_or_default(); dirs.iter() .chain( - ["/usr/bin", "/bin", "/usr/local/bin", "/opt/homebrew/bin"] - .iter() - .map(PathBuf::from) - .collect::>() - .iter(), + [ + "/usr/bin", + "/bin", + "/usr/local/bin", + "/opt/homebrew/bin", + // Git for Windows' own folders (a host started clean + // has a short PATH). + r"C:\Program Files\Git\cmd", + r"C:\Program Files\Git\bin", + ] + .iter() + .map(PathBuf::from) + .collect::>() + .iter(), ) .filter(|d| d.is_absolute()) .map(|d| d.join(if cfg!(windows) { "git.exe" } else { "git" })) @@ -266,8 +275,24 @@ impl Git { let kill_group = |pid: u32| unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL); }; - #[cfg(not(unix))] - let kill_group = |_: u32| {}; + // Windows has no process group to signal: a git that has gone quiet + // is stopped with the tree of processes it started (`taskkill /T`). + // Once git has ended normally nothing is looked for (a process + // spawned per git call would cost more than the leftovers it could + // find). + #[cfg(windows)] + let kill_group = |pid: u32| { + let _ = std::process::Command::new("taskkill") + .args(["/T", "/F", "/PID", &pid.to_string()]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + }; + #[cfg(unix)] + let kill_leftovers = kill_group; + #[cfg(windows)] + let kill_leftovers = |_: u32| {}; let status = loop { if let Some(status) = child.try_wait()? { break status; @@ -284,7 +309,7 @@ impl Git { std::thread::sleep(Duration::from_millis(20)); }; // What it started may still hold its pipes: not waited for long. - kill_group(child.id()); + kill_leftovers(child.id()); let wait = Duration::from_secs(10); let stdout = stdout.recv_timeout(wait).unwrap_or_default(); let stderr = stderr.recv_timeout(wait).unwrap_or_default(); diff --git a/crates/ledger-store/src/winsec.rs b/crates/ledger-store/src/winsec.rs index bcf0638..9f7f629 100644 --- a/crates/ledger-store/src/winsec.rs +++ b/crates/ledger-store/src/winsec.rs @@ -415,11 +415,15 @@ pub fn pid_is_us(pid: u32) -> bool { } /// The process at the client end of the pipe `handle` (a server's). -pub fn pipe_client_pid(handle: std::os::windows::io::RawHandle) -> io::Result { +pub fn pipe_client_pid(pipe: &impl std::os::windows::io::AsRawHandle) -> io::Result { let mut pid = 0u32; // SAFETY: a pipe handle the caller owns and an out value. - if unsafe { windows_sys::Win32::System::Pipes::GetNamedPipeClientProcessId(handle, &mut pid) } - == 0 + if unsafe { + windows_sys::Win32::System::Pipes::GetNamedPipeClientProcessId( + pipe.as_raw_handle(), + &mut pid, + ) + } == 0 { return Err(io::Error::last_os_error()); } @@ -427,11 +431,15 @@ pub fn pipe_client_pid(handle: std::os::windows::io::RawHandle) -> io::Result io::Result { +pub fn pipe_server_pid(pipe: &impl std::os::windows::io::AsRawHandle) -> io::Result { let mut pid = 0u32; // SAFETY: a pipe handle the caller owns and an out value. - if unsafe { windows_sys::Win32::System::Pipes::GetNamedPipeServerProcessId(handle, &mut pid) } - == 0 + if unsafe { + windows_sys::Win32::System::Pipes::GetNamedPipeServerProcessId( + pipe.as_raw_handle(), + &mut pid, + ) + } == 0 { return Err(io::Error::last_os_error()); } @@ -440,13 +448,13 @@ pub fn pipe_server_pid(handle: std::os::windows::io::RawHandle) -> io::Result io::Result { +pub fn pipe_bytes_waiting(pipe: &impl std::os::windows::io::AsRawHandle) -> io::Result { let mut avail = 0u32; // SAFETY: a pipe handle the caller owns; only the available count is // asked for. let ok = unsafe { windows_sys::Win32::System::Pipes::PeekNamedPipe( - handle, + pipe.as_raw_handle(), std::ptr::null_mut(), 0, std::ptr::null_mut(),