From 60399554db753cb010a6770e24ec1b8fa2dcae93 Mon Sep 17 00:00:00 2001 From: RomainROCH Date: Sun, 4 Oct 2026 18:50:52 +0200 Subject: [PATCH] fix: bound control pipe request reads --- docs/SECURITY_AUDIT.md | 19 +- src/ipc.rs | 117 ++++++++++-- tests/ipc_request_limits.rs | 358 ++++++++++++++++++++++++++++++++++++ 3 files changed, 466 insertions(+), 28 deletions(-) create mode 100644 tests/ipc_request_limits.rs diff --git a/docs/SECURITY_AUDIT.md b/docs/SECURITY_AUDIT.md index f6656b9..0048553 100644 --- a/docs/SECURITY_AUDIT.md +++ b/docs/SECURITY_AUDIT.md @@ -7,7 +7,7 @@ ## Summary -Wardoff is materially safer than it was at the previous audit point and is acceptable for distribution to source-first technical early adopters on self-administered machines, but it is still **not yet hardened** for a hostile local-attacker model or multi-user deployment scenarios. The five original pre-release blockers around writable-path autostart, log-directory symlink abuse, named-pipe squatting, cross-session mutex scope, and spoofed `WM_ENDSESSION` teardown are resolved in the current `dev` state. The remaining concerns are same-session/local hardening debt: residual log file targeting, same-session mutex denial of service, the status pipe's default DACL, and the control pipe's unbounded single-client request read path. +Wardoff is materially safer than it was at the previous audit point and is acceptable for distribution to source-first technical early adopters on self-administered machines, but it is still **not yet hardened** for a hostile local-attacker model or multi-user deployment scenarios. The five original pre-release blockers around writable-path autostart, log-directory symlink abuse, named-pipe squatting, cross-session mutex scope, and spoofed `WM_ENDSESSION` teardown are resolved in the current `dev` state. The remaining concerns are same-session/local hardening debt: residual log file targeting, same-session mutex denial of service, and the status pipe's default DACL. The follow-up below records the bounded control-request reader. ## Findings @@ -91,15 +91,16 @@ Wardoff is materially safer than it was at the previous audit point and is accep - **Fix status:** **still open** - **Recommended fix:** Apply an explicit reviewed security descriptor to the status pipe as well. -### [MEDIUM] Control pipe request handling still allows single-client read denial of service +### [INFORMATIONAL] Fixed: control pipe request reads are bounded -- **Location:** `src/ipc.rs:430-537`, `src/ipc.rs:899-914` +- **Location:** `src/ipc.rs` (`read_request`, `read_request_until_deadline`), `tests/ipc_request_limits.rs` - **Attack surface:** Named pipe security -- **Description:** The control server still accepts one client at a time and blocks on `BufRead::read_line` with no explicit size cap or read timeout. A client that connects and never finishes a line can still stall command processing. -- **Exploitability:** Any local client that can connect to the control pipe can deny service to other control requests for the duration of that stalled read. -- **Risk for early-adopter release:** **fix before broader distribution** -- **Fix status:** **still open** -- **Recommended fix:** Add request-size caps plus read timeouts or a more defensive multi-client/overlapped handling model. +- **Description:** The original single-client server blocked on an unbounded `BufRead::read_line`. The 2026-10-04 follow-up limits each control frame to 4096 wire bytes including its required newline and applies a one-second total read deadline. Nonblocking Win32 byte reads distinguish temporary lack of data from EOF; incoming fragments do not renew the deadline. Rejected connections retain nonblocking writes so a client refusing its error reply cannot stall the next accept. +- **Exploitability:** One incomplete client can occupy the serial reader only until its deadline, rather than indefinitely. Repeated connections and existing UI/valid-response waits remain outside this targeted fix; this is not a general hostile-client or multi-user hardening claim. +- **Risk for early-adopter release:** **informational for the original unbounded-read finding** +- **Fix status:** **resolved in source** +- **Validation:** An isolated test runs the production control/status loops on process-unique endpoints with an inert UI consumer. It covers valid and malformed JSON, empty input, UTF-8 errors, size boundaries and fragmented oversized input, missing newline, silent and trickling clients, disconnect/recovery, independent status response before the control deadline, and orderly shutdown while a client is incomplete. No power protections or autostart changes are required. Run `cargo test --test ipc_request_limits -- --nocapture`. +- **Recommended follow-up:** Retain this regression test and validate supported Windows environments before broader distribution. The same-user/elevation policy and status-pipe ACL are unchanged. ### [MEDIUM] Scheduled-task ACL hardening is still not explicit @@ -154,4 +155,4 @@ Additional notes: ## Conclusion -Wardoff is acceptable for distribution to source-first technical early adopters on self-administered machines. The five original pre-release blockers (writable-path autostart, log-directory symlink abuse, named-pipe squatting, cross-session mutex, WM_ENDSESSION spoofing) are resolved. Known remaining items (same-session mutex DoS, status pipe default DACL, control pipe unbounded read, residual log file targeting) are local hardening debt against same-session adversaries, not exploitable privilege escalation. Wardoff is not yet hardened against a hostile local-attacker model or multi-user deployment scenarios. +Wardoff is acceptable for distribution to source-first technical early adopters on self-administered machines. The five original pre-release blockers (writable-path autostart, log-directory symlink abuse, named-pipe squatting, cross-session mutex, WM_ENDSESSION spoofing) are resolved, and the follow-up bounds the control-request read. Known remaining items (same-session mutex DoS, status pipe default DACL, residual log file targeting) are local hardening debt against same-session adversaries, not exploitable privilege escalation. Wardoff is not yet hardened against a hostile local-attacker model or multi-user deployment scenarios. diff --git a/src/ipc.rs b/src/ipc.rs index d335da8..a71a7b5 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -7,19 +7,19 @@ use serde::{Deserialize, Serialize}; use std::fs::{File, OpenOptions}; use std::io::{BufRead, BufReader, Write}; use std::mem::size_of; -use std::os::windows::io::{FromRawHandle, RawHandle}; +use std::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle}; use std::sync::{ atomic::{AtomicBool, Ordering}, mpsc::{self, Sender}, Arc, }; use std::thread::{self, JoinHandle}; -use std::time::Duration; +use std::time::{Duration, Instant}; use windows::core::{Error as WindowsError, PCWSTR, PWSTR}; use windows::Win32::Foundation::{ - CloseHandle, GetLastError, LocalFree, ERROR_ACCESS_DENIED, ERROR_FILE_NOT_FOUND, - ERROR_INSUFFICIENT_BUFFER, ERROR_PATH_NOT_FOUND, ERROR_PIPE_BUSY, ERROR_PIPE_CONNECTED, HANDLE, - HLOCAL, LPARAM, WPARAM, + CloseHandle, GetLastError, LocalFree, ERROR_ACCESS_DENIED, ERROR_BROKEN_PIPE, + ERROR_FILE_NOT_FOUND, ERROR_INSUFFICIENT_BUFFER, ERROR_NO_DATA, ERROR_PATH_NOT_FOUND, + ERROR_PIPE_BUSY, ERROR_PIPE_CONNECTED, HANDLE, HLOCAL, LPARAM, WPARAM, }; use windows::Win32::Security::Authorization::{ ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1, @@ -29,12 +29,12 @@ use windows::Win32::Security::{ TOKEN_USER, }; use windows::Win32::Storage::FileSystem::{ - FILE_FLAGS_AND_ATTRIBUTES, FILE_FLAG_FIRST_PIPE_INSTANCE, PIPE_ACCESS_DUPLEX, + ReadFile, FILE_FLAGS_AND_ATTRIBUTES, FILE_FLAG_FIRST_PIPE_INSTANCE, PIPE_ACCESS_DUPLEX, PIPE_ACCESS_OUTBOUND, }; use windows::Win32::System::Pipes::{ - ConnectNamedPipe, CreateNamedPipeW, PIPE_READMODE_MESSAGE, PIPE_REJECT_REMOTE_CLIENTS, - PIPE_TYPE_MESSAGE, PIPE_WAIT, + ConnectNamedPipe, CreateNamedPipeW, SetNamedPipeHandleState, PIPE_NOWAIT, PIPE_READMODE_BYTE, + PIPE_READMODE_MESSAGE, PIPE_REJECT_REMOTE_CLIENTS, PIPE_TYPE_MESSAGE, PIPE_WAIT, }; use windows::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; use windows::Win32::UI::WindowsAndMessaging::{PostThreadMessageW, WM_APP}; @@ -42,6 +42,10 @@ use windows::Win32::UI::WindowsAndMessaging::{PostThreadMessageW, WM_APP}; const IPC_SERVER_THREAD_NAME: &str = "wardoff-ipc-server"; const STATUS_SERVER_THREAD_NAME: &str = "wardoff-status-server"; const PIPE_BUFFER_SIZE: u32 = 4096; +// Includes the terminating newline; all current control commands are much smaller. +const MAX_REQUEST_BYTES: usize = 4096; +const REQUEST_READ_TIMEOUT: Duration = Duration::from_secs(1); +const REQUEST_READ_POLL_DELAY: Duration = Duration::from_millis(10); const PIPE_CONNECT_ATTEMPTS: usize = 20; const PIPE_CONNECT_DELAY: Duration = Duration::from_millis(100); const INITIAL_PIPE_CLAIM_ATTEMPTS: usize = 10; @@ -897,20 +901,95 @@ fn is_retryable_initial_pipe_claim_error(error_code: u32) -> bool { } fn read_request(pipe: &mut File) -> Result { - let mut request_line = String::new(); - let bytes_read = { - let mut reader = BufReader::new(pipe); - reader - .read_line(&mut request_line) - .map_err(|error| format!("Wardoff could not read an IPC request: {error}"))? - }; + // Nonblocking byte reads also handle a JSON line split across pipe messages. + // The pipe is owned by this server thread; no helper thread or pending I/O survives a timeout. + let handle = HANDLE(pipe.as_raw_handle()); + let read_mode = PIPE_READMODE_BYTE | PIPE_NOWAIT; + unsafe { SetNamedPipeHandleState(handle, Some(&read_mode), None, None) } + .map_err(|error| format!("Wardoff could not bound its IPC request read: {error}"))?; + + let result = read_request_until_deadline(pipe); + // Valid commands keep the existing reply behavior. Rejections remain nonblocking: + // a client that never reads its error reply must not stall the next accept either. + if result.is_ok() { + let reply_mode = PIPE_READMODE_MESSAGE | PIPE_WAIT; + unsafe { SetNamedPipeHandleState(handle, Some(&reply_mode), None, None) } + .map_err(|error| format!("Wardoff could not restore its IPC reply mode: {error}"))?; + } + result +} - if bytes_read == 0 { - return Err("Wardoff received an empty IPC request.".to_string()); +fn read_request_until_deadline(pipe: &mut File) -> Result { + let deadline = Instant::now() + REQUEST_READ_TIMEOUT; + let mut request_line = Vec::with_capacity(MAX_REQUEST_BYTES); + loop { + if Instant::now() >= deadline { + return Err("Wardoff timed out reading an IPC request.".to_string()); + } + + let mut buffer = [0; 512]; + let mut bytes_read = 0; + // std::fs::File::read maps ERROR_NO_DATA to EOF on Windows. Preserve that + // Win32 result here so an idle connected client really waits for the deadline. + let read = unsafe { + ReadFile( + HANDLE(pipe.as_raw_handle()), + Some(&mut buffer), + Some(&mut bytes_read), + None, + ) + }; + let error_code = unsafe { GetLastError() }; + match read { + Ok(()) if bytes_read == 0 => { + return Err(incomplete_request_error(request_line.is_empty())) + } + Ok(()) => { + let bytes_read = bytes_read as usize; + let newline = buffer[..bytes_read].iter().position(|byte| *byte == b'\n'); + let line_bytes = newline.map_or(bytes_read, |position| position + 1); + if request_line.len() + line_bytes > MAX_REQUEST_BYTES { + return Err(format!( + "Wardoff IPC request exceeds {MAX_REQUEST_BYTES} bytes." + )); + } + request_line.extend_from_slice(&buffer[..line_bytes]); + if newline.is_some() { + let line = std::str::from_utf8(&request_line) + .map_err(|error| format!("Wardoff received malformed IPC JSON: {error}"))?; + if line.trim().is_empty() { + return Err("Wardoff received an empty IPC request.".to_string()); + } + return serde_json::from_str(line.trim_end()) + .map_err(|error| format!("Wardoff received malformed IPC JSON: {error}")); + } + // A full buffer without a newline cannot complete within the frame limit. + if request_line.len() == MAX_REQUEST_BYTES { + return Err(format!( + "Wardoff IPC request exceeds {MAX_REQUEST_BYTES} bytes." + )); + } + } + Err(_) if error_code == ERROR_NO_DATA => { + // One absolute deadline, not a fresh timeout for every arriving byte. + thread::sleep( + REQUEST_READ_POLL_DELAY.min(deadline.saturating_duration_since(Instant::now())), + ); + } + Err(_) if error_code == ERROR_BROKEN_PIPE => { + return Err(incomplete_request_error(request_line.is_empty())); + } + Err(error) => return Err(format!("Wardoff could not read an IPC request: {error}")), + } } +} - serde_json::from_str(request_line.trim_end()) - .map_err(|error| format!("Wardoff received malformed IPC JSON: {error}")) +fn incomplete_request_error(empty: bool) -> String { + if empty { + "Wardoff received an empty IPC request.".to_string() + } else { + "Wardoff received an IPC request without a terminating newline.".to_string() + } } fn write_response(pipe: &mut File, response: &IpcResponse) -> Result<(), String> { diff --git a/tests/ipc_request_limits.rs b/tests/ipc_request_limits.rs new file mode 100644 index 0000000..9673166 --- /dev/null +++ b/tests/ipc_request_limits.rs @@ -0,0 +1,358 @@ +//! Runs the production IPC server on process-unique pipes with an inert UI consumer. +//! No blocker, autostart, singleton, logger, or production endpoint is started. +#![allow(dead_code)] + +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +use std::os::windows::io::AsRawHandle; +use std::sync::mpsc; +use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, +}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; +use windows::Win32::Foundation::HANDLE; +use windows::Win32::System::Pipes::PeekNamedPipe; +use windows::Win32::System::Threading::GetCurrentThreadId; +use windows::Win32::UI::WindowsAndMessaging::{PeekMessageW, MSG, PM_NOREMOVE}; + +mod blocker { + pub(crate) enum BlockerMode { + Block, + Allow, + } +} + +mod cli { + #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] + pub(crate) struct StatusOutput { + pub(crate) state: String, + } + + impl StatusOutput { + pub(crate) fn to_json(&self) -> Result { + serde_json::to_string(self) + } + } +} + +mod logger { + pub(crate) enum EventSource { + Ipc, + } + + pub(crate) fn log_event(_: &str, _: EventSource, _: impl Into, _: bool) {} +} + +mod session_scope { + pub(crate) fn current_control_pipe_path() -> Result { + Ok(format!( + r"\\.\pipe\WardoffIpcTest-Control-{}", + std::process::id() + )) + } + + pub(crate) fn current_status_pipe_path() -> Result { + Ok(format!( + r"\\.\pipe\WardoffIpcTest-Status-{}", + std::process::id() + )) + } +} + +#[path = "../src/ipc.rs"] +mod ipc; + +struct TestRuntime { + server: ipc::IpcServer, + ui_thread: Option>, + dispatched: Arc, +} + +impl TestRuntime { + fn start() -> Self { + let (request_tx, request_rx) = mpsc::channel::(); + let (ready_tx, ready_rx) = mpsc::channel(); + let dispatched = Arc::new(AtomicUsize::new(0)); + let ui_dispatched = Arc::clone(&dispatched); + let ui_thread = thread::spawn(move || { + // PostThreadMessageW requires the target thread to own a message queue. + let mut message = MSG::default(); + unsafe { + let _ = PeekMessageW(&mut message, None, 0, 0, PM_NOREMOVE); + } + ready_tx + .send(unsafe { GetCurrentThreadId() }) + .expect("UI ready"); + for pending in request_rx { + ui_dispatched.fetch_add(1, Ordering::SeqCst); + let response = match pending.request { + ipc::IpcRequest::Status => ipc::IpcResponse::Status { + status: cli::StatusOutput { + state: "inert".into(), + }, + }, + _ => ipc::IpcResponse::Ok, + }; + let _ = pending.response_tx.send(response); + } + }); + let server = ipc::IpcServer::start(request_tx, ready_rx.recv().expect("UI thread ID")) + .expect("isolated IPC startup"); + Self { + server, + ui_thread: Some(ui_thread), + dispatched, + } + } + + fn stop(&mut self) { + self.server.shutdown().expect("clean IPC shutdown"); + if let Some(ui_thread) = self.ui_thread.take() { + ui_thread.join().expect("inert UI joined"); + } + } +} + +impl Drop for TestRuntime { + fn drop(&mut self) { + self.stop(); + } +} + +fn open_pipe(status: bool) -> File { + let path = if status { + session_scope::current_status_pipe_path() + } else { + session_scope::current_control_pipe_path() + } + .expect("isolated path"); + let deadline = Instant::now() + Duration::from_secs(3); + loop { + match OpenOptions::new().read(true).write(!status).open(&path) { + Ok(pipe) => return pipe, + Err(error) if Instant::now() >= deadline => panic!("connect {path}: {error}"), + Err(_) => thread::sleep(Duration::from_millis(10)), + } + } +} + +fn read_line_with_deadline(pipe: &mut File) -> Result { + let deadline = Instant::now() + Duration::from_secs(2); + let mut reply = Vec::new(); + loop { + let mut buffer = [0; 512]; + let mut available = 0; + // This test client is the sole reader, so observed bytes cannot be consumed elsewhere. + unsafe { + PeekNamedPipe( + HANDLE(pipe.as_raw_handle()), + None, + 0, + None, + Some(&mut available), + None, + ) + } + .map_err(|error| error.to_string())?; + let to_read = buffer.len().min(available as usize); + if to_read != 0 { + match pipe.read(&mut buffer[..to_read]) { + Ok(0) => return Err("server disconnected before replying".into()), + Ok(size) => { + reply.extend_from_slice(&buffer[..size]); + if reply.contains(&b'\n') { + return String::from_utf8(reply).map_err(|error| error.to_string()); + } + } + Err(error) => return Err(error.to_string()), + } + } + if Instant::now() >= deadline { + return Err("test client received no reply within 2 seconds".into()); + } + thread::sleep(Duration::from_millis(10)); + } +} + +fn exchange(payload: &[u8]) -> Result { + let mut pipe = open_pipe(false); + // Match send_request's separate JSON and newline pipe messages. + if let Some(json) = payload.strip_suffix(b"\n") { + pipe.write_all(json).map_err(|error| error.to_string())?; + pipe.write_all(b"\n").map_err(|error| error.to_string())?; + } else { + pipe.write_all(payload).map_err(|error| error.to_string())?; + } + serde_json::from_str(&read_line_with_deadline(&mut pipe)?).map_err(|error| error.to_string()) +} + +fn assert_normal_command() { + assert!(matches!( + exchange(b"{\"command\":\"set_mode\",\"mode\":\"allow\"}\n"), + Ok(ipc::IpcResponse::Ok) + )); +} + +fn assert_error(payload: &[u8], expected: &str) { + let reply = exchange(payload).expect("error response"); + assert!(matches!(reply, ipc::IpcResponse::Error { message } if message.contains(expected))); + assert_normal_command(); +} + +#[test] +fn isolated_runtime_rejects_bad_clients_and_recovers() { + let mut runtime = TestRuntime::start(); + assert_normal_command(); + assert!(matches!( + exchange(b"{\"command\":\"set_mode\",\"mode\":\"block\"}\r\n"), + Ok(ipc::IpcResponse::Ok) + )); + + // A rejected client may keep its handle open and refuse to read the error. + // The unknown variant diagnostic can be larger than the outbound pipe buffer. + let mut unread = open_pipe(false); + let command = "x".repeat(4070); + let payload = format!("{{\"command\":\"{command}\"}}\n"); + unread + .write_all(payload.as_bytes()) + .expect("large malformed write"); + assert_normal_command(); + drop(unread); + for (payload, expected) in [ + (&b"not JSON\n"[..], "malformed"), + (&b"\n"[..], "empty"), + (&b"{\"command\":\"unknown\"}\n"[..], "malformed"), + (&b"\xff\n"[..], "malformed"), + ] { + assert_error(payload, expected); + } + + // The cap counts wire bytes including LF, regardless of JSON whitespace or message size. + let mut boundary = b"{\"command\":\"status\"}".to_vec(); + boundary.resize(4095, b' '); + boundary.push(b'\n'); + assert!(matches!( + exchange(&boundary), + Ok(ipc::IpcResponse::Status { .. }) + )); + assert_normal_command(); + boundary.insert(boundary.len() - 1, b' '); + assert_error(&boundary, "exceeds 4096 bytes"); + assert_error(&vec![b' '; 4097], "exceeds 4096 bytes"); + + // Small individual messages must not bypass the cumulative frame cap. + let mut oversized = open_pipe(false); + for _ in 0..9 { + let _ = oversized.write_all(&[b' '; 512]); + } + let reply = read_line_with_deadline(&mut oversized).expect("fragmented oversized reply"); + drop(oversized); + assert!(reply.contains("exceeds 4096 bytes")); + assert_normal_command(); + + // EOF never turns an unterminated control command into an accepted request. + for payload in [ + &b""[..], + &b"{\"command\":\"set_mode\",\"mode\":\"allow\"}"[..], + ] { + let before = runtime.dispatched.load(Ordering::SeqCst); + let mut disconnected = open_pipe(false); + disconnected + .write_all(payload) + .expect("write before disconnect"); + drop(disconnected); + assert_normal_command(); + assert_eq!(runtime.dispatched.load(Ordering::SeqCst), before + 1); + } + + // A syntactically valid request without its newline must expire while open. + let started = Instant::now(); + let mut stalled = open_pipe(false); + stalled + .write_all(b"{\"command\":\"status\"}") + .expect("partial write"); + let status_started = Instant::now(); + let mut status_pipe = open_pipe(true); + let status = read_line_with_deadline(&mut status_pipe).expect("independent status response"); + let status_elapsed = status_started.elapsed(); + assert!( + status_elapsed < Duration::from_millis(500), + "status waited {status_elapsed:?}" + ); + eprintln!("status during incomplete request: {status_elapsed:?}"); + assert_eq!( + serde_json::from_str::(&status).expect("status JSON")["state"], + "inert" + ); + drop(status_pipe); + let reply = read_line_with_deadline(&mut stalled); + drop(stalled); // Unblocks the original reader too, even if this assertion fails. + assert!( + matches!(reply.as_ref().ok().and_then(|line| serde_json::from_str::(line).ok()), Some(ipc::IpcResponse::Error { message }) if message.contains("timed out")), + "{reply:?}" + ); + let partial_elapsed = started.elapsed(); + assert!(partial_elapsed >= Duration::from_millis(800)); + assert!(partial_elapsed < Duration::from_millis(1500)); + eprintln!("unterminated request rejected: {partial_elapsed:?}"); + assert_normal_command(); + + // A client sending nothing also expires; another queued client then succeeds. + let mut silent = open_pipe(false); + let normal = thread::spawn(assert_normal_command); + let reply = read_line_with_deadline(&mut silent).expect("silent timeout reply"); + drop(silent); + assert!(reply.contains("timed out")); + normal + .join() + .expect("normal client recovered after timeout"); + + // Trickle traffic must not reset the absolute deadline. + let mut slow = open_pipe(false); + let mut slow_writer = slow.try_clone().expect("slow writer handle"); + let started = Instant::now(); + let writer = thread::spawn(move || { + for _ in 0..20 { + if slow_writer.write_all(b" ").is_err() { + break; + } + thread::sleep(Duration::from_millis(80)); + } + }); + let reply = read_line_with_deadline(&mut slow); + drop(slow); + let elapsed = started.elapsed(); + writer.join().expect("slow writer joined"); + assert!(reply.expect("trickle timeout reply").contains("timed out")); + assert!(elapsed < Duration::from_millis(1500)); + assert!(elapsed >= Duration::from_millis(800)); + eprintln!("trickle request rejected: {elapsed:?}"); + assert_normal_command(); + + // Clean shutdown must still finish when the current control client is incomplete. + let mut shutdown_stall = open_pipe(false); + shutdown_stall + .write_all(b"{") + .expect("shutdown partial write"); + let (stopped_tx, stopped_rx) = mpsc::channel(); + let shutdown = thread::spawn(move || { + runtime.stop(); + stopped_tx.send(()).expect("shutdown result"); + }); + let stopped = stopped_rx.recv_timeout(Duration::from_secs(2)); + drop(shutdown_stall); + shutdown.join().expect("shutdown joined"); + stopped.expect("shutdown completed despite incomplete client"); + // Both owned endpoints are gone after orderly shutdown. + for path in [ + session_scope::current_control_pipe_path(), + session_scope::current_status_pipe_path(), + ] { + assert!(OpenOptions::new() + .read(true) + .open(path.expect("test path")) + .is_err()); + } +}