diff --git a/.github/workflows/windows.yml b/.github/workflows/windows.yml index 8860481..a5b9de5 100644 --- a/.github/workflows/windows.yml +++ b/.github/workflows/windows.yml @@ -126,6 +126,10 @@ jobs: if: failure() && steps.tests.outcome == 'failure' timeout-minutes: 3 run: cargo test --locked --test runtime -- --test-threads=1 --nocapture + - name: Diagnose failed daemon tests + if: failure() && steps.tests.outcome == 'failure' + timeout-minutes: 3 + run: cargo test --locked --test windows-daemon -- --test-threads=1 --nocapture - name: Smoke test executable if: ${{ !cancelled() && steps.build.outcome == 'success' }} run: | @@ -145,9 +149,10 @@ jobs: Target: $env:CARGO_BUILD_TARGET - This job builds the executable and runs all enabled tests natively, including libghostty parser, protocol, and direct TerminalService ConPTY runtime tests. + This job builds the executable and runs all enabled tests natively, including libghostty parser, protocol, direct TerminalService ConPTY runtime, named-pipe transport, and daemon tests. The runtime suite spawns real Rust console children and checks input/output, Unicode, cwd/environment/argv/PATH, OS resize, snapshots, replay, terminal replies, and independent terminals. + Daemon tests cover private atomic registration, ownership/handoff, stale registration/locking, partial requests, blocked subscribers, and live ConPTY operations over real named pipes. The older service, ownership, playground, and rows integration suites remain Unix-only. - A passing run does not imply Windows named-pipe transport or complete ConPTY shutdown coverage is implemented. + Basic tests do not establish exhaustive lifecycle coverage or Windows interactive CLI parity. No packages or releases are published. "@ >> $env:GITHUB_STEP_SUMMARY diff --git a/Cargo.toml b/Cargo.toml index 2436e88..b4d8a9d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ windows-sys = { version = "0.61.2", features = [ "Win32_System_IO", "Win32_System_Pipes", "Win32_System_Threading", + "Win32_System_WindowsProgramming", ] } [target.'cfg(windows)'.dev-dependencies] diff --git a/README.md b/README.md index fed4ddc..f04df8a 100644 --- a/README.md +++ b/README.md @@ -10,18 +10,19 @@ cannot block the service or another terminal. Every daemon has one owner connection. The playground starts a daemon and holds that connection until exit; exiting the playground stops the daemon and all its terminals. Observer commands connect to an existing daemon without taking ownership. -The service uses a private authenticated Unix socket and atomic registration file. +The service uses a private authenticated local byte stream (Unix socket or Windows +named pipe) and an atomic registration file. Integrations launch `opencode-pty daemon` (protocol 7). The server must claim the daemon within 5 seconds by sending the authenticated framed envelope `{"token":"...","request":{"op":"own","instance_id":"..."}}`. The response is `{"type":"owned"}`; that connection stays open as the sole -owner. Ordinary requests and subscriptions use their existing separate sockets. +owner. Ordinary requests and subscriptions use separate connections. The instance ID and token come from the private registration file. Losing the owner connection stops the daemon and its terminals unless the owner first sends `{"token":"...","request":{"op":"prepare_handoff"}}` on that same -socket. The response is `{"type":"handoff","ticket":"...","expires_at":123}`, +connection. The response is `{"type":"handoff","ticket":"...","expires_at":123}`, where `expires_at` is Unix milliseconds, 120 seconds from preparation. Repeated preparation during that window returns the same ticket and deadline. After the old owner disconnects, a successor claims the same instance with the ticket in @@ -60,8 +61,22 @@ response or the final subscription event. Completion retains queued bytes while waiting for that close, with a two-second grace period and daemon cancellation; it never calls the potentially unbounded `FlushFileBuffers`. Stopping acceptance retains a pipe instance until registration is removed, preventing namespace -squatting during cleanup. The backend has native Windows tests, but the Windows -daemon entrypoint/registration lifecycle is not enabled yet. +squatting during cleanup. + +On Windows, `opencode-pty daemon` stores `service.json` and `service.lock` in +`%LOCALAPPDATA%\opencode-pty`, or an absolute `OPENCODE_PTY_RUNTIME_DIR` override. +Missing `LOCALAPPDATA` without an override is an error. Storage entries must not +be reparse points or owned by another user; the directory, lock, and registration +use protected current-user-only ACLs. The held directory/lock handles deny delete +sharing, and registration is atomically replaced under the exclusive lock. +Stale registration is replaced with a fresh instance ID, token, and pipe name; +cleanup removes registration only if its instance ID still matches. + +The Windows Rust `TerminalClient` and interactive CLI are not ported. Integrations +can use protocol 7 directly and the minimal `daemon::PipeConnection` byte-stream +helper (connect plus optional read/write timeouts). On Windows, `service_dir()` +and `registration_path()` are fallible because runtime storage must be resolved +without an insecure fallback. ## Architecture @@ -263,11 +278,15 @@ observes actual console stdin without echo; `Size` and `Context` inspect the child's OS console and process context. The ignored `child` test is only its subprocess entry point, not skipped runtime coverage. -The older service, ownership, playground, and rows integration suites remain -Unix-only. Windows library tests also exercise real named-pipe roundtrips, -multiple connections, namespace ownership, cancellation, and final-frame -completion. These checks do not yet establish Windows daemon support or -complete ConPTY shutdown behavior. +The original service, ownership, playground, and rows integration suites remain +Unix-only. Windows library tests exercise real named-pipe roundtrips, multiple +connections, namespace ownership, cancellation, final-frame completion, and +private atomic registration. `tests/windows-daemon.rs` exercises authenticated +ownership/handoff, locking/stale registration, partial requests, blocked +subscribers, and live ConPTY create/input/output/resize/shutdown through the real +daemon. Natural child-exit/ConPTY EOF and runtime cleanup are checked by the +direct runtime suite; the basic daemon tests do not establish complete lifecycle +coverage on their own. On Windows, normal root-child exit hands the ConPTY master to the existing child-wait worker for closing. The actor and reader continue draining until the @@ -322,8 +341,8 @@ uses named pipes. Platform signing will be added later. ## Current Limits -- The Windows named-pipe backend is tested independently; persistent daemon - startup and private registration storage are still Unix-only. +- Windows supports the daemon/protocol transport, not the Rust interactive + TerminalClient/play/watch CLI. - Ordinary API operations use one framed JSON request per connection; subscriptions keep the authenticated connection open for ordered live events. - The OpenCode backend proxy and ordered group APIs are implemented, but the diff --git a/src/daemon.rs b/src/daemon.rs index b7aed43..fb2e678 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -18,25 +18,33 @@ pub struct Registration { #[cfg(unix)] #[path = "daemon/unix.rs"] mod platform; -#[cfg(unix)] +#[cfg(windows)] +#[path = "daemon/windows.rs"] +mod platform; +#[cfg(any(unix, windows))] mod server; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub use platform::{read_registration, registration_path, service_dir}; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub use server::run; -#[cfg(not(unix))] +/// Minimal Windows byte-stream client for integrations using protocol framing. +/// This does not start a daemon or implement the interactive TerminalClient CLI. +#[cfg(windows)] +pub use crate::transport::windows::Connection as PipeConnection; + +#[cfg(not(any(unix, windows)))] pub fn run() -> anyhow::Result<()> { anyhow::bail!("persistent opencode-pty transport is not implemented on this platform") } -#[cfg(not(unix))] +#[cfg(not(any(unix, windows)))] pub fn read_registration() -> anyhow::Result { anyhow::bail!("persistent opencode-pty transport is not implemented on this platform") } -#[cfg(not(unix))] +#[cfg(not(any(unix, windows)))] pub fn registration_path() -> PathBuf { PathBuf::from("opencode-pty-service.json") } diff --git a/src/daemon/server.rs b/src/daemon/server.rs index 6ccb043..452e352 100644 --- a/src/daemon/server.rs +++ b/src/daemon/server.rs @@ -20,41 +20,49 @@ pub fn run() -> Result<()> { let service = Arc::new(TerminalService::default()); let shutdown = Arc::new(AtomicBool::new(false)); let mut handlers = Vec::<(Cancellation, thread::JoinHandle<()>)>::new(); - while !shutdown.load(Ordering::Acquire) { - if ownership - .lock() - .map_err(|_| anyhow!("ownership lock poisoned"))? - .tick(Instant::now()) - { - shutdown.store(true, Ordering::Release); - break; - } - for (_, handler) in handlers.extract_if(.., |(_, handler)| handler.is_finished()) { - let _ = handler.join(); - } - match listener.accept() { - Ok(Some(stream)) => { - let control = stream.cancellation()?; - let service = Arc::clone(&service); - let shutdown = Arc::clone(&shutdown); - let ownership = Arc::clone(&ownership); - let registration = registration.clone(); - let handle = thread::spawn(move || { - if let Err(error) = - handle_connection(stream, &service, ®istration, &shutdown, &ownership) - { - eprintln!("opencode-pty request failed: {error:#}"); - } - }); - handlers.push((control, handle)); + let result = (|| -> Result<()> { + while !shutdown.load(Ordering::Acquire) { + if ownership + .lock() + .map_err(|_| anyhow!("ownership lock poisoned"))? + .tick(Instant::now()) + { + shutdown.store(true, Ordering::Release); + break; + } + for (_, handler) in handlers.extract_if(.., |(_, handler)| handler.is_finished()) { + let _ = handler.join(); } - Ok(None) => { - thread::sleep(Duration::from_millis(10)); + match listener.accept() { + Ok(Some(stream)) => { + let control = stream.cancellation()?; + let service = Arc::clone(&service); + let shutdown = Arc::clone(&shutdown); + let ownership = Arc::clone(&ownership); + let registration = registration.clone(); + let handle = thread::spawn(move || { + if let Err(error) = handle_connection( + stream, + &service, + ®istration, + &shutdown, + &ownership, + ) { + eprintln!("opencode-pty request failed: {error:#}"); + } + }); + handlers.push((control, handle)); + } + Ok(None) => { + thread::sleep(Duration::from_millis(10)); + } + Err(error) => return Err(error.into()), } - Err(error) => return Err(error.into()), } - } + Ok(()) + })(); + shutdown.store(true, Ordering::Release); listener.stop(); // Unblock partial requests, owner reads, and backpressured subscriptions // before joining. PTY workers still use their existing termination path. @@ -75,12 +83,12 @@ pub fn run() -> Result<()> { let _ = handler.join(); } drop(service); - platform::cleanup(registration)?; + let cleanup = platform::cleanup(registration); drop(listener); let _ = cleanup_tx.send(()); let _ = watchdog.join(); drop(runtime); - Ok(()) + result.and(cleanup) } fn handle_connection( diff --git a/src/daemon/windows.rs b/src/daemon/windows.rs new file mode 100644 index 0000000..e55262f --- /dev/null +++ b/src/daemon/windows.rs @@ -0,0 +1,352 @@ +use std::ffi::OsString; +use std::fs::{self, File}; +use std::io::{Read, Write}; +use std::os::windows::ffi::{OsStrExt, OsStringExt}; +use std::os::windows::io::AsRawHandle; +use std::path::{Path, PathBuf}; +use std::ptr::null_mut; + +use anyhow::{Context, Result, bail}; +use fs2::FileExt; +use windows_sys::Win32::Foundation::{ERROR_ALREADY_EXISTS, GENERIC_READ, GENERIC_WRITE}; +use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, CREATE_NEW, CreateDirectoryW, CreateFileW, DELETE, + FILE_ATTRIBUTE_DIRECTORY, FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS, + FILE_FLAG_OPEN_REPARSE_POINT, FILE_READ_ATTRIBUTES, FILE_RENAME_INFO, FILE_RENAME_INFO_0, + FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_TRAVERSE, FileRenameInfoEx, + GetFileInformationByHandle, GetFinalPathNameByHandleW, OPEN_ALWAYS, OPEN_EXISTING, + READ_CONTROL, SetFileInformationByHandle, WRITE_DAC, +}; +use windows_sys::Win32::System::WindowsProgramming::{ + FILE_RENAME_FLAG_POSIX_SEMANTICS, FILE_RENAME_FLAG_REPLACE_IF_EXISTS, +}; + +use super::{LOCK_FILE, REGISTRATION_FILE, Registration}; +use crate::protocol::PROTOCOL_VERSION; +use crate::transport::Listener; +use crate::transport::windows::{owned_handle, security::PrivateSecurity}; + +pub fn service_dir() -> Result { + let directory = if let Some(path) = std::env::var_os("OPENCODE_PTY_RUNTIME_DIR") { + PathBuf::from(path) + } else { + PathBuf::from( + std::env::var_os("LOCALAPPDATA") + .context("LOCALAPPDATA is unavailable; set OPENCODE_PTY_RUNTIME_DIR")?, + ) + .join("opencode-pty") + }; + if !directory.is_absolute() { + bail!("PTY runtime directory must be an absolute path"); + } + Ok(directory) +} + +pub fn registration_path() -> Result { + Ok(service_dir()?.join(REGISTRATION_FILE)) +} + +pub fn read_registration() -> Result { + read_registration_at(®istration_path()?) +} + +fn read_registration_at(path: &Path) -> Result { + let security = PrivateSecurity::new()?; + let mut file = open( + path, + GENERIC_READ | READ_CONTROL, + OPEN_EXISTING, + FILE_SHARE_READ | FILE_SHARE_DELETE, + &security, + )?; + check_kind(&file, false)?; + security.check_private(file.as_raw_handle())?; + let mut data = Vec::new(); + // Registration is tiny; do not allocate an unbounded stale/corrupt file. + Read::by_ref(&mut file) + .take(16 * 1024 + 1) + .read_to_end(&mut data)?; + if data.len() > 16 * 1024 { + bail!("PTY registration is too large"); + } + serde_json::from_slice(&data).context("invalid opencode-pty registration") +} + +pub(super) struct Runtime { + pub registration: Registration, + _lock: File, + _directory: File, +} + +impl Runtime { + pub fn bind() -> Result<(Self, Listener)> { + let security = PrivateSecurity::new()?; + let directory = service_dir()?; + if let Some(parent) = directory.parent() { + fs::create_dir_all(parent)?; + } + let wide = wide_path(&directory)?; + // SAFETY: wide path and private descriptor are valid through creation. + if unsafe { CreateDirectoryW(wide.as_ptr(), &security.attributes()) } == 0 { + let error = std::io::Error::last_os_error(); + if error.raw_os_error() != Some(ERROR_ALREADY_EXISTS as i32) { + return Err(error.into()); + } + } + // Deny delete sharing so the private directory and lock cannot be + // replaced out from under the running daemon. + let directory_handle = open( + &directory, + FILE_READ_ATTRIBUTES | FILE_TRAVERSE | READ_CONTROL | WRITE_DAC, + OPEN_EXISTING, + FILE_SHARE_READ | FILE_SHARE_WRITE, + &security, + )?; + check_kind(&directory_handle, true)?; + security.make_private(directory_handle.as_raw_handle())?; + let directory = held_directory_path(&directory_handle)?; + let lock = open( + &directory.join(LOCK_FILE), + GENERIC_READ | GENERIC_WRITE | READ_CONTROL | WRITE_DAC, + OPEN_ALWAYS, + FILE_SHARE_READ | FILE_SHARE_WRITE, + &security, + )?; + check_kind(&lock, false)?; + security.make_private(lock.as_raw_handle())?; + lock.try_lock_exclusive() + .context("another opencode-pty process already owns the service lock")?; + + let instance_id = format!("{:032x}", rand::random::()); + let registration = Registration { + socket: PathBuf::from(format!(r"\\.\pipe\opencode-pty-{instance_id}")), + instance_id, + pid: std::process::id(), + protocol: PROTOCOL_VERSION, + token: format!("{:032x}", rand::random::()), + }; + let listener = Listener::bind(®istration.socket)?; + write_registration(&directory, &directory_handle, ®istration, &security)?; + Ok(( + Self { + registration, + _lock: lock, + _directory: directory_handle, + }, + listener, + )) + } +} + +fn write_registration( + directory: &Path, + directory_handle: &File, + registration: &Registration, + security: &PrivateSecurity, +) -> Result<()> { + let temporary = directory.join(format!("service.{}.tmp", registration.instance_id)); + let result = (|| -> Result<()> { + let mut file = open( + &temporary, + GENERIC_WRITE | READ_CONTROL | DELETE, + CREATE_NEW, + 0, + security, + )?; + file.write_all(&serde_json::to_vec_pretty(registration)?)?; + file.sync_all()?; + publish(&file, directory_handle) + })(); + if result.is_err() { + let _ = fs::remove_file(temporary); + } + result +} + +fn publish(file: &File, directory: &File) -> Result<()> { + // The Win32 wrapper rejects a non-null RootDirectory, unlike the native + // NtSetInformationFile interface. Resolve the full destination from the held + // directory handle, never from the unvalidated environment path. + let name = held_directory_path(directory)? + .join(REGISTRATION_FILE) + .as_os_str() + .encode_wide() + .collect::>(); + let name_bytes = u32::try_from(size_of_val(name.as_slice()))?; + let bytes = size_of::() + .checked_add(name_bytes as usize) + .context("registration rename buffer overflow")?; + let buffer_bytes = u32::try_from(bytes)?; + // FILE_RENAME_INFO has a variable-length trailing UTF-16 name. usize gives + // this buffer the struct's alignment on both supported 64-bit architectures. + const { + assert!(align_of::() <= align_of::()); + } + let mut buffer = vec![0_usize; bytes.div_ceil(size_of::())]; + let allocation = buffer.as_mut_ptr().cast::(); + let info = allocation.cast::(); + // SAFETY: the zeroed, aligned allocation covers the header and complete name. + // Use raw pointers into the whole allocation, not a borrow of FileName[1] + // whose extent is smaller than the variable-sized trailing data. + unsafe { + (&raw mut (*info).Anonymous).write(FILE_RENAME_INFO_0 { + Flags: FILE_RENAME_FLAG_REPLACE_IF_EXISTS | FILE_RENAME_FLAG_POSIX_SEMANTICS, + }); + (&raw mut (*info).RootDirectory).write(null_mut()); + (&raw mut (*info).FileNameLength).write(name_bytes); + let destination = allocation + .add(std::mem::offset_of!(FILE_RENAME_INFO, FileName)) + .cast::(); + std::ptr::copy_nonoverlapping(name.as_ptr(), destination, name.len()); + // Unlike ordinary MoveFileEx replacement, POSIX semantics preserve old + // reader handles while new opens see the new file, with no missing-name + // interval. Source and destination directory stay held throughout; the + // source is never reopened and no inherited ACL/metadata merge occurs. + if SetFileInformationByHandle( + file.as_raw_handle(), + FileRenameInfoEx, + info.cast(), + buffer_bytes, + ) == 0 + { + return Err(std::io::Error::last_os_error()) + .context("failed to atomically publish PTY registration"); + } + } + Ok(()) +} + +fn held_directory_path(directory: &File) -> Result { + // SAFETY: query the normalized DOS path length for a live directory handle. + let length = unsafe { GetFinalPathNameByHandleW(directory.as_raw_handle(), null_mut(), 0, 0) }; + if length == 0 { + return Err(std::io::Error::last_os_error().into()); + } + let mut path = vec![0_u16; length as usize]; + // SAFETY: the buffer has the requested length and the directory stays held. + let written = unsafe { + GetFinalPathNameByHandleW(directory.as_raw_handle(), path.as_mut_ptr(), length, 0) + }; + if written == 0 { + return Err(std::io::Error::last_os_error().into()); + } + if written >= length { + bail!("PTY runtime directory path changed during publication"); + } + Ok(PathBuf::from(OsString::from_wide( + &path[..written as usize], + ))) +} + +pub(super) fn cleanup(registration: &Registration) -> Result<()> { + let path = registration_path()?; + if read_registration_at(&path) + .is_ok_and(|current| current.instance_id == registration.instance_id) + { + fs::remove_file(path)?; + } + // A named pipe is not a filesystem entry. Its listener remains held until + // this registration cleanup has completed, then all handles are closed. + Ok(()) +} + +fn open( + path: &Path, + access: u32, + creation: u32, + sharing: u32, + security: &PrivateSecurity, +) -> Result { + let path = wide_path(path)?; + // SAFETY: all pointers are valid; OPEN_REPARSE_POINT lets check_kind reject + // links rather than inspecting a different target after following them. + let handle = unsafe { + CreateFileW( + path.as_ptr(), + access, + sharing, + &security.attributes(), + creation, + FILE_FLAG_OPEN_REPARSE_POINT | FILE_FLAG_BACKUP_SEMANTICS, + null_mut(), + ) + }; + Ok(File::from(owned_handle(handle)?)) +} + +fn check_kind(file: &File, directory: bool) -> Result<()> { + let mut info = BY_HANDLE_FILE_INFORMATION::default(); + // SAFETY: file owns a live handle; info is valid writable output. + if unsafe { GetFileInformationByHandle(file.as_raw_handle(), &mut info) } == 0 { + return Err(std::io::Error::last_os_error().into()); + } + if info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + bail!("PTY storage must not be a reparse point"); + } + if (info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY != 0) != directory { + bail!("unexpected PTY storage file type"); + } + Ok(()) +} + +fn wide_path(path: &Path) -> Result> { + let mut wide = path.as_os_str().encode_wide().collect::>(); + if wide.contains(&0) { + bail!("PTY storage path contains NUL"); + } + wide.push(0); + Ok(wide) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registration_is_private_and_atomically_replaceable_with_a_reader_open() { + let directory = + std::env::temp_dir().join(format!("pty-registration-{:032x}", rand::random::())); + fs::create_dir(&directory).unwrap(); + let security = PrivateSecurity::new().unwrap(); + let directory_handle = open( + &directory, + FILE_READ_ATTRIBUTES | FILE_TRAVERSE, + OPEN_EXISTING, + FILE_SHARE_READ | FILE_SHARE_WRITE, + &security, + ) + .unwrap(); + let mut registration = Registration { + instance_id: "first".into(), + pid: 1, + protocol: PROTOCOL_VERSION, + socket: PathBuf::from(r"\\.\pipe\opencode-pty-first"), + token: "secret".into(), + }; + write_registration(&directory, &directory_handle, ®istration, &security).unwrap(); + let path = directory.join(REGISTRATION_FILE); + let mut old_reader = open( + &path, + GENERIC_READ | READ_CONTROL, + OPEN_EXISTING, + FILE_SHARE_READ | FILE_SHARE_DELETE, + &security, + ) + .unwrap(); + security.check_private(old_reader.as_raw_handle()).unwrap(); + registration.instance_id = "second".into(); + write_registration(&directory, &directory_handle, ®istration, &security).unwrap(); + assert_eq!(read_registration_at(&path).unwrap().instance_id, "second"); + let mut old = String::new(); + old_reader.read_to_string(&mut old).unwrap(); + assert_eq!( + serde_json::from_str::(&old) + .unwrap() + .instance_id, + "first" + ); + drop(old_reader); + drop(directory_handle); + fs::remove_dir_all(directory).unwrap(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 6b0cb8e..95c8c6f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,7 +1,7 @@ pub mod client; pub mod daemon; mod ghostty; -#[cfg(unix)] +#[cfg(any(unix, windows))] mod ownership; pub mod protocol; pub mod service; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 81d0809..e7a7873 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -18,7 +18,7 @@ pub(crate) use unix::{Cancellation, Connection, Listener}; #[cfg(windows)] mod retained; -// The backend is exercised natively before the daemon entrypoint is enabled. #[cfg(windows)] -#[cfg_attr(not(test), allow(dead_code))] pub(crate) mod windows; +#[cfg(windows)] +pub(crate) use windows::{Cancellation, Connection, Listener}; diff --git a/src/transport/windows.rs b/src/transport/windows.rs index 1b177b2..f2762d7 100644 --- a/src/transport/windows.rs +++ b/src/transport/windows.rs @@ -70,6 +70,8 @@ impl Listener { let previous = std::mem::replace(&mut self.pending, next); Ok(Some(Connection { pipe: Arc::clone(&previous.operation.pipe), + read_timeout: None, + write_timeout: None, })) } @@ -159,8 +161,10 @@ impl Pipe { } } -pub(crate) struct Connection { +pub struct Connection { pipe: Arc, + read_timeout: Option, + write_timeout: Option, } impl Connection { @@ -186,6 +190,8 @@ impl Connection { Ok(handle) => { return Ok(Self { pipe: Arc::new(Pipe::new(handle)?), + read_timeout: None, + write_timeout: None, }); } Err(error) if error.raw_os_error() == Some(ERROR_PIPE_BUSY as i32) => { @@ -205,14 +211,22 @@ impl Connection { } } - pub fn cancellation(&self) -> io::Result { + pub fn set_read_timeout(&mut self, timeout: Option) { + self.read_timeout = timeout; + } + + pub fn set_write_timeout(&mut self, timeout: Option) { + self.write_timeout = timeout; + } + + pub(crate) fn cancellation(&self) -> io::Result { Ok(Cancellation(Arc::clone(&self.pipe))) } /// Protocol 7 has an explicit response/final event, not a pipe half-close. /// Retain the pipe until the client reads that frame and closes its end. /// An uncooperative peer is cancelled after a bounded grace period. - pub fn finish_response(&mut self) -> io::Result<()> { + pub(crate) fn finish_response(&mut self) -> io::Result<()> { match self.read_with_timeout(&mut [0], Some(COMPLETION_TIMEOUT)) { Ok(0) => Ok(()), Ok(_) => Err(io::Error::new( @@ -226,9 +240,11 @@ impl Connection { } } - pub fn monitor_disconnect(&self) -> io::Result { + pub(crate) fn monitor_disconnect(&self) -> io::Result { let mut reader = Self { pipe: Arc::clone(&self.pipe), + read_timeout: None, + write_timeout: None, }; let cancellation = self.cancellation()?; let (sender, disconnected) = bounded(1); @@ -282,7 +298,7 @@ impl Connection { impl Read for Connection { fn read(&mut self, bytes: &mut [u8]) -> io::Result { - self.read_with_timeout(bytes, None) + self.read_with_timeout(bytes, self.read_timeout) } } @@ -307,7 +323,7 @@ impl Write for Connection { })?; Ok(match ready { Some(count) => count, - None => operation.wait(None)?, + None => operation.wait(self.write_timeout)?, } as usize) } diff --git a/src/transport/windows/security.rs b/src/transport/windows/security.rs index 369c76f..57f8f37 100644 --- a/src/transport/windows/security.rs +++ b/src/transport/windows/security.rs @@ -2,12 +2,16 @@ use std::io; use std::os::windows::io::AsRawHandle; use std::ptr::null_mut; -use windows_sys::Win32::Foundation::LocalFree; +use windows_sys::Win32::Foundation::{HANDLE, LocalFree}; use windows_sys::Win32::Security::Authorization::{ - ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1, + ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, GetSecurityInfo, + SDDL_REVISION_1, SE_FILE_OBJECT, SetSecurityInfo, }; use windows_sys::Win32::Security::{ - GetTokenInformation, SECURITY_ATTRIBUTES, TOKEN_QUERY, TOKEN_USER, TokenUser, + ACCESS_ALLOWED_ACE, ACE_HEADER, DACL_SECURITY_INFORMATION, EqualSid, GetAce, + GetSecurityDescriptorControl, GetSecurityDescriptorDacl, GetSecurityDescriptorOwner, + GetTokenInformation, OWNER_SECURITY_INFORMATION, PROTECTED_DACL_SECURITY_INFORMATION, + SE_DACL_PROTECTED, SECURITY_ATTRIBUTES, TOKEN_QUERY, TOKEN_USER, TokenUser, }; use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; @@ -46,6 +50,123 @@ impl PrivateSecurity { bInheritHandle: 0, } } + + /// Existing files/directories are never adopted from a different owner. + /// Apply the protected current-user DACL before any secrets are published. + pub fn make_private(&self, handle: HANDLE) -> io::Result<()> { + self.check_owner(handle)?; + let mut present = 0; + let mut defaulted = 0; + let mut dacl = null_mut(); + // SAFETY: self owns a valid descriptor and returned DACL remains inside it. + if unsafe { GetSecurityDescriptorDacl(self.0, &mut present, &mut dacl, &mut defaulted) } + == 0 + { + return Err(io::Error::last_os_error()); + } + // SAFETY: the live handle is opened with WRITE_DAC; dacl is valid above. + let error = unsafe { + SetSecurityInfo( + handle, + SE_FILE_OBJECT, + DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION, + null_mut(), + null_mut(), + dacl, + null_mut(), + ) + }; + if error != 0 { + return Err(io::Error::from_raw_os_error(error as i32)); + } + Ok(()) + } + + pub fn check_private(&self, handle: HANDLE) -> io::Result<()> { + let actual = self.check_owner(handle)?; + let mut present = 0; + let mut defaulted = 0; + let mut dacl = null_mut(); + let mut control = 0; + let mut revision = 0; + // SAFETY: actual owns a valid Windows descriptor throughout inspection. + if unsafe { GetSecurityDescriptorDacl(actual.0, &mut present, &mut dacl, &mut defaulted) } + == 0 + || unsafe { GetSecurityDescriptorControl(actual.0, &mut control, &mut revision) } == 0 + { + return Err(io::Error::last_os_error()); + } + if present == 0 || dacl.is_null() || control & SE_DACL_PROTECTED == 0 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "unprotected PTY storage DACL", + )); + } + let mut ace = null_mut(); + // SAFETY: dacl is non-null and belongs to the live actual descriptor. + if unsafe { (*dacl).AceCount } != 1 || unsafe { GetAce(dacl, 0, &mut ace) } == 0 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "PTY storage must grant only the current user", + )); + } + // ACCESS_ALLOWED_ACE_TYPE is zero. Other ACE layouts must not be cast. + // SAFETY: Windows returned an ACE from a validated security descriptor. + if unsafe { (*ace.cast::()).AceType } != 0 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "unexpected PTY storage ACE", + )); + } + // SAFETY: an ACCESS_ALLOWED_ACE has an inline SID beginning at SidStart. + let sid = unsafe { (&raw mut (*ace.cast::()).SidStart).cast() }; + if !self.is_current_user(sid)? { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "PTY storage grants another principal", + )); + } + Ok(()) + } + + fn check_owner(&self, handle: HANDLE) -> io::Result { + let mut owner = null_mut(); + let mut descriptor = null_mut(); + // SAFETY: query a live file handle. Descriptor is a unique LocalFree allocation. + let error = unsafe { + GetSecurityInfo( + handle, + SE_FILE_OBJECT, + OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION, + &mut owner, + null_mut(), + null_mut(), + null_mut(), + &mut descriptor, + ) + }; + if error != 0 { + return Err(io::Error::from_raw_os_error(error as i32)); + } + let actual = Self(descriptor); + if owner.is_null() || !self.is_current_user(owner)? { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "PTY storage is not owned by the current user", + )); + } + Ok(actual) + } + + fn is_current_user(&self, sid: *mut core::ffi::c_void) -> io::Result { + let mut owner = null_mut(); + let mut defaulted = 0; + // SAFETY: self owns the descriptor; supplied SID belongs to another live descriptor. + if unsafe { GetSecurityDescriptorOwner(self.0, &mut owner, &mut defaulted) } == 0 { + return Err(io::Error::last_os_error()); + } + Ok(unsafe { EqualSid(owner, sid) } != 0) + } } impl Drop for PrivateSecurity { @@ -106,3 +227,61 @@ fn current_user_sid() -> io::Result { }; Ok(sid) } + +#[cfg(test)] +mod tests { + use super::*; + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Foundation::{GENERIC_READ, GENERIC_WRITE}; + use windows_sys::Win32::Storage::FileSystem::{ + CREATE_NEW, CreateFileW, READ_CONTROL, WRITE_DAC, + }; + + #[test] + fn rejects_public_storage_acl_and_privatises_only_our_owned_file() { + let path = std::env::temp_dir().join(format!("pty-acl-{:032x}", rand::random::())); + let wide = path + .as_os_str() + .encode_wide() + .chain([0]) + .collect::>(); + let sid = current_user_sid().unwrap(); + let public = format!("O:{sid}D:P(A;;GA;;;WD)\0") + .encode_utf16() + .collect::>(); + let mut descriptor = null_mut(); + // SAFETY: valid SDDL; allocation is immediately adopted by the RAII owner. + assert_ne!( + unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + public.as_ptr(), + SDDL_REVISION_1, + &mut descriptor, + null_mut(), + ) + }, + 0 + ); + let public = PrivateSecurity(descriptor); + // SAFETY: a unique temporary file, live private owner + deliberately public + // test DACL. No secrets are written while the DACL is permissive. + let file = owned_handle(unsafe { + CreateFileW( + wide.as_ptr(), + GENERIC_READ | GENERIC_WRITE | READ_CONTROL | WRITE_DAC, + 0, + &public.attributes(), + CREATE_NEW, + 0, + null_mut(), + ) + }) + .unwrap(); + let private = PrivateSecurity::new().unwrap(); + assert!(private.check_private(file.as_raw_handle()).is_err()); + private.make_private(file.as_raw_handle()).unwrap(); + private.check_private(file.as_raw_handle()).unwrap(); + drop(file); + std::fs::remove_file(path).unwrap(); + } +} diff --git a/tests/windows-daemon.rs b/tests/windows-daemon.rs new file mode 100644 index 0000000..0cf92f8 --- /dev/null +++ b/tests/windows-daemon.rs @@ -0,0 +1,385 @@ +#![cfg(windows)] + +// Direct-service and daemon tests share the runtime lane's real child fixture, +// rather than maintaining separate console setup. +#[path = "support/terminal_fixture.rs"] +mod terminal_fixture; + +use std::io::Write; +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::{Duration, Instant}; + +use base64::Engine; +use opencode_pty::daemon::{PipeConnection, Registration}; +use opencode_pty::protocol::{ + AttachmentRole, Envelope, Request, Response, SubscriptionEvent, read_frame, + read_subscription_event, write_frame, +}; +use opencode_pty::service::{CreateTerminal, TerminalInfo}; +use terminal_fixture::{Command as FixtureCommand, Deadline, Fixture, TempDir}; +use windows_sys::Win32::Foundation::WAIT_OBJECT_0; +use windows_sys::Win32::System::Threading::{ + OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject, +}; + +struct Daemon { + child: Child, + registration: Registration, + directory: PathBuf, + root: TempDir, +} + +impl Daemon { + fn start() -> Self { + let root = TempDir::new(); + let directory = root.0.join("runtime"); + let mut child = spawn(&directory); + let registration = registration(&mut child, &directory); + Self { + child, + registration, + directory, + root, + } + } + + fn connect(&self) -> PipeConnection { + let mut stream = PipeConnection::connect(&self.registration.socket).unwrap(); + stream.set_read_timeout(Some(Duration::from_secs(5))); + stream.set_write_timeout(Some(Duration::from_secs(5))); + stream + } + + fn send(&self, stream: &mut PipeConnection, request: Request) -> Response { + write_frame( + &mut *stream, + &Envelope { + token: self.registration.token.clone(), + request, + }, + ) + .unwrap(); + read_frame(stream).unwrap() + } + + fn request(&self, request: Request) -> Response { + self.send(&mut self.connect(), request) + } + + fn own(&self, ticket: Option) -> (PipeConnection, Response) { + let mut stream = self.connect(); + let response = self.send( + &mut stream, + Request::Own { + instance_id: self.registration.instance_id.clone(), + ticket, + }, + ); + (stream, response) + } + + fn create(&self, fixture: &Fixture) -> TerminalInfo { + let CreateTerminal { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + } = fixture.request(); + match self.request(Request::Create { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + }) { + Response::Created { terminal } => terminal, + response => panic!("create: {response:?}"), + } + } + + fn subscribe(&self, id: u64, role: AttachmentRole) -> PipeConnection { + let mut stream = self.connect(); + assert!(matches!( + self.send( + &mut stream, + Request::Subscribe { + id, + offset: 0, + attachment_id: "daemon-test".into(), + role, + takeover: false, + } + ), + Response::Attached { .. } + )); + stream + } + + fn wait(&mut self) { + let status = wait(&mut self.child); + assert!(status.success(), "daemon failed: {status}"); + assert!(!self.directory.join("service.json").exists()); + assert!(PipeConnection::connect(&self.registration.socket).is_err()); + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + if self.child.try_wait().ok().flatten().is_none() { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } +} + +fn spawn(directory: &Path) -> Child { + Command::new(env!("CARGO_BIN_EXE_opencode-pty")) + .arg("daemon") + .env("OPENCODE_PTY_RUNTIME_DIR", directory) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .spawn() + .unwrap() +} + +fn registration(child: &mut Child, directory: &Path) -> Registration { + let deadline = Instant::now() + Duration::from_secs(4); + loop { + if let Ok(data) = std::fs::read(directory.join("service.json")) + && let Ok(registration) = serde_json::from_slice::(&data) + && registration.pid == child.id() + { + assert_eq!(registration.protocol, 7); + assert_eq!( + registration.socket, + PathBuf::from(format!( + r"\\.\pipe\opencode-pty-{}", + registration.instance_id + )) + ); + return registration; + } + assert!( + child.try_wait().unwrap().is_none(), + "daemon exited before registering" + ); + assert!(Instant::now() < deadline, "daemon did not register"); + thread::sleep(Duration::from_millis(10)); + } +} + +fn wait(child: &mut Child) -> std::process::ExitStatus { + let deadline = Instant::now() + Duration::from_secs(8); + loop { + if let Some(status) = child.try_wait().unwrap() { + return status; + } + assert!(Instant::now() < deadline, "daemon did not exit"); + thread::sleep(Duration::from_millis(10)); + } +} + +#[test] +fn authenticated_ownership_handoff_and_shutdown_use_real_pipes() { + let _deadline = Deadline::new(); + let mut daemon = Daemon::start(); + let mut invalid = daemon.connect(); + write_frame( + &mut invalid, + &Envelope { + token: "wrong".into(), + request: Request::Ping, + }, + ) + .unwrap(); + assert!( + matches!(read_frame(&mut invalid).unwrap(), Response::Error { message } if message == "authentication failed") + ); + drop(invalid); + assert!(matches!( + daemon.request(Request::Own { + instance_id: "wrong".into(), + ticket: None + }), + Response::Error { .. } + )); + let (mut owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + assert!(matches!(daemon.own(None).1, Response::Error { .. })); + let Response::Handoff { ticket, expires_at } = daemon.send(&mut owner, Request::PrepareHandoff) + else { + panic!("handoff response"); + }; + assert!( + matches!(daemon.send(&mut owner, Request::PrepareHandoff), Response::Handoff { ticket: repeated, expires_at: deadline } if repeated == ticket && deadline == expires_at) + ); + assert!(matches!( + daemon.own(Some(ticket.clone())).1, + Response::Error { .. } + )); + drop(owner); + let deadline = Instant::now() + Duration::from_secs(3); + let mut successor = loop { + let (stream, response) = daemon.own(Some(ticket.clone())); + if matches!(response, Response::Owned) { + break stream; + } + assert!(Instant::now() < deadline, "handoff claim: {response:?}"); + thread::sleep(Duration::from_millis(10)); + }; + assert!(matches!( + daemon.send(&mut successor, Request::Shutdown), + Response::Ok + )); + drop(successor); // Protocol completion, not Unix half-close. + daemon.wait(); +} + +#[test] +fn service_lock_and_stale_registration_are_instance_scoped() { + let _deadline = Deadline::new(); + let mut daemon = Daemon::start(); + let mut duplicate = spawn(&daemon.directory); + assert!(!wait(&mut duplicate).success()); + let old = daemon.registration.clone(); + daemon.child.kill().unwrap(); + daemon.child.wait().unwrap(); + assert!(daemon.directory.join("service.json").exists()); + daemon.child = spawn(&daemon.directory); + daemon.registration = registration(&mut daemon.child, &daemon.directory); + assert_ne!(old.instance_id, daemon.registration.instance_id); + assert_ne!(old.token, daemon.registration.token); + assert_ne!(old.socket, daemon.registration.socket); + let (owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + assert!(matches!(daemon.request(Request::Shutdown), Response::Ok)); + drop(owner); + daemon.wait(); +} + +#[test] +fn unclaimed_daemon_cancels_a_partial_request() { + let _deadline = Deadline::new(); + let mut daemon = Daemon::start(); + let mut partial = daemon.connect(); + partial.write_all(&[0, 0]).unwrap(); + daemon.wait(); +} + +#[test] +fn named_pipe_daemon_create_input_output_resize_and_shutdown() { + let _deadline = Deadline::new(); + let mut daemon = Daemon::start(); + let (owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + let fixture = Fixture::new(); + // Exercise a path containing spaces/Unicode using the shared child fixture. + let executable = daemon.root.executable("daemon fixture.exe"); + let mut request = fixture.request(); + request.program = executable.to_str().unwrap().into(); + let CreateTerminal { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + } = request; + let Response::Created { terminal } = daemon.request(Request::Create { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + }) else { + panic!("create response"); + }; + let mut child = fixture.connect(); + let mut subscription = daemon.subscribe(terminal.id, AttachmentRole::Observer); + child.command(FixtureCommand::Output("\x1b[2J\x1b[Hdaemon-output".into())); + let mut bytes = Vec::new(); + while !String::from_utf8_lossy(&bytes).contains("daemon-output") { + if let SubscriptionEvent::Output { bytes: output, .. } = + read_subscription_event(&mut subscription).unwrap() + { + bytes.extend(output); + } + } + let input = b"real-input"; + assert!(matches!( + daemon.request(Request::Write { + id: terminal.id, + attachment_id: None, + data_base64: base64::engine::general_purpose::STANDARD.encode(input) + }), + Response::Ok + )); + assert_eq!( + child.command(FixtureCommand::Read(input.len())), + serde_json::json!(input) + ); + assert!(matches!( + daemon.request(Request::Resize { + id: terminal.id, + attachment_id: None, + cols: 93, + rows: 31 + }), + Response::Ok + )); + assert_eq!( + child.command(FixtureCommand::Size), + serde_json::json!([93, 31]) + ); + assert!( + matches!(daemon.request(Request::Snapshot { id: terminal.id }), Response::Snapshot { text, .. } if text.contains("daemon-output")) + ); + // Natural-exit/EOF ordering belongs to the runtime cleanup milestone. This + // basic daemon test deliberately exercises live-child operations + shutdown. + drop(subscription); + assert!(matches!(daemon.request(Request::Shutdown), Response::Ok)); + drop(owner); + daemon.wait(); +} + +#[test] +fn owner_loss_cancels_blocked_subscriber_and_partial_request() { + let _deadline = Deadline::new(); + let mut daemon = Daemon::start(); + let (owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + let fixture = Fixture::new(); + let terminal = daemon.create(&fixture); + let mut child = fixture.connect(); + // SAFETY: open the live child for waiting; PID is only used to obtain a + // stable process handle for this assertion, not as terminal identity. + let process = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, terminal.pid.unwrap()) }; + assert!(!process.is_null()); + let process = unsafe { OwnedHandle::from_raw_handle(process) }; + let _blocked = daemon.subscribe(terminal.id, AttachmentRole::Controller); + let mut partial = daemon.connect(); + partial.write_all(&[0, 0]).unwrap(); + child.command(FixtureCommand::Output("x".repeat(256 * 1024))); + thread::sleep(Duration::from_millis(100)); + drop(owner); + daemon.wait(); + assert_eq!( + unsafe { WaitForSingleObject(process.as_raw_handle(), 3000) }, + WAIT_OBJECT_0 + ); +}