diff --git a/.github/workflows/core.yml b/.github/workflows/core.yml index 4bcf70e..2fe071d 100644 --- a/.github/workflows/core.yml +++ b/.github/workflows/core.yml @@ -26,11 +26,15 @@ jobs: run: rustup toolchain install nightly-2026-09-02 --profile minimal --component miri --component rust-src - run: cargo +nightly-2026-09-02 miri setup - name: Check Stacked Borrows - run: cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/ghostty-effects/Cargo.toml + run: | + cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/ghostty-effects/Cargo.toml + cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/transport-retained/Cargo.toml - name: Check Tree Borrows env: MIRIFLAGS: -Zmiri-tree-borrows - run: cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/ghostty-effects/Cargo.toml + run: | + cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/ghostty-effects/Cargo.toml + cargo +nightly-2026-09-02 miri test --locked --manifest-path tests/transport-retained/Cargo.toml test: name: Core (${{ matrix.os }}) diff --git a/Cargo.lock b/Cargo.lock index 9673431..d517fb5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -198,6 +198,7 @@ dependencies = [ "serde_json", "sha2", "shell-words", + "windows-sys", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 638cf20..0b87bd2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,17 @@ libc = "0.2" nix = { version = "0.28", features = ["process", "term", "user"] } sha2 = "0.10" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61.2", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_Security_Authorization", + "Win32_Storage_FileSystem", + "Win32_System_IO", + "Win32_System_Pipes", + "Win32_System_Threading", +] } + [profile.release] codegen-units = 1 lto = "thin" diff --git a/README.md b/README.md index 4e7c1c0..8b2fdd9 100644 --- a/README.md +++ b/README.md @@ -40,15 +40,28 @@ backend preserves half-close when completing subscriptions so queued final frames are not discarded on macOS. Shutdown cancellation is a separate operation that wakes partial requests and blocked subscription writes before joining. -The planned Windows backend keeps protocol 7 and the registration fields +The Windows named-pipe backend keeps protocol 7 and the registration fields `instance_id`, `pid`, `protocol`, `socket`, and `token`. Treat `socket` as an opaque -local endpoint: on Windows it will be `\\.\pipe\opencode-pty-`, not a +local endpoint: on Windows it is `\\.\pipe\opencode-pty-`, not a filesystem socket. A random per-instance name, exclusive first pipe instance, current-user access control, and rejection of remote clients protect the endpoint; the private registration file remains the discovery and authentication source. -Named-pipe completion will use explicit bounded, cancellable delivery rather than -pretending to support Unix half-close. This boundary alone does not implement -Windows transport or daemon support. +The pipe is full-duplex and byte-mode, with overlapped reads/writes capped at +64 KiB per kernel operation. Accept polling never waits for a client; connect +retries are bounded to five seconds. Each pending operation is cancelled and its +completion reaped before its buffers or event are freed. +The kernel-retained `OVERLAPPED` allocation uses an owned raw pointer, not a +retained borrow from a movable `Box`; it is reclaimed only after I/O completion. +The allocation helper's move/repeated-access behavior is checked under both +Miri borrow models alongside the callback ownership tests. + +Named pipes have no half-close. Protocol clients close after reading an ordinary +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. ## Architecture @@ -235,8 +248,9 @@ the build tree and runs it without Cargo's DLL search paths, verifying that libghostty is statically linked. It does not publish packages or releases. The current service, ownership, playground, and rows integration suites are -Unix-only. A green Windows job verifies compilation and the enabled parser and -protocol tests, not working Windows transport or ConPTY lifecycle support. +Unix-only. Windows library tests also exercise real named-pipe roundtrips, +multiple connections, namespace ownership, cancellation, and final-frame +completion. They do not yet verify Windows daemon or ConPTY lifecycle support. Once the workflow is on `master`, it can also be run manually against a branch: @@ -274,8 +288,8 @@ uses named pipes. Platform signing will be added later. ## Current Limits -- Persistent transport currently uses Unix sockets; Windows named pipes are not - implemented yet. +- The Windows named-pipe backend is tested independently; persistent daemon + startup and private registration storage are still Unix-only. - 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/client.rs b/src/client.rs index 07ff218..60bf3af 100644 --- a/src/client.rs +++ b/src/client.rs @@ -7,10 +7,11 @@ use anyhow::{Context, Result, anyhow, bail}; use base64::Engine; use crate::daemon::{Registration, read_registration}; +#[cfg(unix)] use crate::protocol::{ - AttachmentRole, Envelope, PROTOCOL_VERSION, Request, Response, SubscriptionEvent, read_frame, - read_subscription_event, write_frame, + AttachmentRole, Envelope, SubscriptionEvent, read_frame, read_subscription_event, write_frame, }; +use crate::protocol::{PROTOCOL_VERSION, Request, Response}; use crate::service::{CreateTerminal, TerminalId, TerminalInfo, TerminalRows}; #[cfg(unix)] use crate::transport::Connection; diff --git a/src/daemon/server.rs b/src/daemon/server.rs index 3b23ae8..6ccb043 100644 --- a/src/daemon/server.rs +++ b/src/daemon/server.rs @@ -55,7 +55,7 @@ pub fn run() -> Result<()> { } } - drop(listener); + listener.stop(); // Unblock partial requests, owner reads, and backpressured subscriptions // before joining. PTY workers still use their existing termination path. for (cancellation, _) in &handlers { @@ -76,6 +76,7 @@ pub fn run() -> Result<()> { } drop(service); platform::cleanup(registration)?; + drop(listener); let _ = cleanup_tx.send(()); let _ = watchdog.join(); drop(runtime); diff --git a/src/lib.rs b/src/lib.rs index 0eba3b7..6b0cb8e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,5 +5,5 @@ mod ghostty; mod ownership; pub mod protocol; pub mod service; -#[cfg(unix)] +#[cfg(any(unix, windows))] mod transport; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 1e165e3..81d0809 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -14,3 +14,11 @@ mod unix; #[cfg(unix)] 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; diff --git a/src/transport/retained.rs b/src/transport/retained.rs new file mode 100644 index 0000000..e1aaa2d --- /dev/null +++ b/src/transport/retained.rs @@ -0,0 +1,26 @@ +use std::ptr::NonNull; + +/// Own a value whose raw pointer is retained by native I/O. Moving this owner or +/// retrieving its pointer must not reborrow the value while native code accesses +/// it. The caller must reap native I/O before dropping this allocation. +pub(crate) struct Retained(NonNull); + +impl Retained { + pub fn new(value: T) -> Self { + Self(NonNull::new(Box::into_raw(Box::new(value))).expect("Box pointer is non-null")) + } + + pub fn as_ptr(&self) -> *mut T { + self.0.as_ptr() + } +} + +impl Drop for Retained { + fn drop(&mut self) { + // SAFETY: this owner uniquely owns the Box allocation. All native access + // must have finished before the caller permits this owner to drop. + unsafe { + drop(Box::from_raw(self.0.as_ptr())); + } + } +} diff --git a/src/transport/unix.rs b/src/transport/unix.rs index b2c5ccd..429edf6 100644 --- a/src/transport/unix.rs +++ b/src/transport/unix.rs @@ -7,18 +7,21 @@ use std::time::Duration; use crossbeam_channel::{Receiver, bounded}; -pub(crate) struct Listener(UnixListener); +pub(crate) struct Listener(Option); impl Listener { pub fn bind(endpoint: &Path) -> io::Result { let listener = UnixListener::bind(endpoint)?; listener.set_nonblocking(true)?; - Ok(Self(listener)) + Ok(Self(Some(listener))) } /// Poll for a connection without blocking the daemon control path. pub fn accept(&mut self) -> io::Result> { - match self.0.accept() { + let Some(listener) = &self.0 else { + return Ok(None); + }; + match listener.accept() { Ok((stream, _)) => { // macOS inherits the listener's nonblocking mode on accepted sockets. stream.set_nonblocking(false)?; @@ -28,6 +31,10 @@ impl Listener { Err(error) => Err(error), } } + + pub fn stop(&mut self) { + self.0.take(); + } } pub(crate) struct Connection(UnixStream); diff --git a/src/transport/windows.rs b/src/transport/windows.rs new file mode 100644 index 0000000..1b177b2 --- /dev/null +++ b/src/transport/windows.rs @@ -0,0 +1,686 @@ +//! Full-duplex, overlapped byte-mode named pipes. Every pending operation owns +//! its OVERLAPPED/event and is cancelled and reaped before its buffer is released. +//! No FlushFileBuffers: it can wait indefinitely for an uncooperative client. + +use std::io::{self, Read, Write}; +use std::os::windows::ffi::OsStrExt; +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; +use std::path::Path; +use std::ptr::{null, null_mut}; +use std::sync::Arc; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use crossbeam_channel::{Receiver, bounded}; +use windows_sys::Win32::Foundation::{ + ERROR_BROKEN_PIPE, ERROR_IO_INCOMPLETE, ERROR_IO_PENDING, ERROR_NO_DATA, ERROR_PIPE_BUSY, + ERROR_PIPE_CONNECTED, ERROR_PIPE_NOT_CONNECTED, GENERIC_READ, GENERIC_WRITE, HANDLE, + INVALID_HANDLE_VALUE, WAIT_OBJECT_0, WAIT_TIMEOUT, +}; +use windows_sys::Win32::Storage::FileSystem::{ + CreateFileW, FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED, OPEN_EXISTING, + PIPE_ACCESS_DUPLEX, ReadFile, SECURITY_IDENTIFICATION, SECURITY_SQOS_PRESENT, WriteFile, +}; +use windows_sys::Win32::System::IO::{CancelIoEx, GetOverlappedResult, OVERLAPPED}; +use windows_sys::Win32::System::Pipes::{ + ConnectNamedPipe, CreateNamedPipeW, PIPE_READMODE_BYTE, PIPE_REJECT_REMOTE_CLIENTS, + PIPE_TYPE_BYTE, PIPE_UNLIMITED_INSTANCES, PIPE_WAIT, WaitNamedPipeW, +}; +use windows_sys::Win32::System::Threading::{ + CreateEventW, INFINITE, SetEvent, WaitForMultipleObjects, WaitForSingleObject, +}; + +pub(crate) mod security; +use super::retained::Retained; +use security::PrivateSecurity; + +const BUFFER_BYTES: u32 = 64 * 1024; +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const COMPLETION_TIMEOUT: Duration = Duration::from_secs(2); + +pub(crate) struct Listener { + endpoint: Vec, + security: PrivateSecurity, + pending: PendingConnection, + stopped: bool, +} + +impl Listener { + pub fn bind(endpoint: &Path) -> io::Result { + let endpoint = endpoint_name(endpoint)?; + let security = PrivateSecurity::new()?; + // Fail closed if anyone already owns this name. Never attach to or + // replace an existing server, even one with the same user SID. + let pending = PendingConnection::new(&endpoint, &security, true)?; + Ok(Self { + endpoint, + security, + pending, + stopped: false, + }) + } + + pub fn accept(&mut self) -> io::Result> { + if self.stopped || !self.pending.ready()? { + return Ok(None); + } + // Create the next instance BEFORE handing off the connected one, so the + // registered name is continuously owned even between short requests. + let next = PendingConnection::new(&self.endpoint, &self.security, false)?; + let previous = std::mem::replace(&mut self.pending, next); + Ok(Some(Connection { + pipe: Arc::clone(&previous.operation.pipe), + })) + } + + /// Stop accepting without releasing the namespace. Keep the listener alive + /// until registration is removed, including throughout daemon cleanup. + pub fn stop(&mut self) { + self.stopped = true; + self.pending.operation.pipe.cancel(); + } +} + +struct PendingConnection { + operation: Operation, + connected: bool, +} + +impl PendingConnection { + fn new(endpoint: &[u16], security: &PrivateSecurity, first: bool) -> io::Result { + let flags = PIPE_ACCESS_DUPLEX + | FILE_FLAG_OVERLAPPED + | if first { + FILE_FLAG_FIRST_PIPE_INSTANCE + } else { + 0 + }; + // SAFETY: endpoint is NUL-terminated; attributes and their descriptor live + // through the call. The returned handle is uniquely adopted below. + let handle = unsafe { + CreateNamedPipeW( + endpoint.as_ptr(), + flags, + PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS, + PIPE_UNLIMITED_INSTANCES, + BUFFER_BYTES, + BUFFER_BYTES, + 0, + &security.attributes(), + ) + }; + let pipe = Arc::new(Pipe::new(owned_handle(handle)?)?); + let mut operation = Operation::new(pipe)?; + let connected = match operation.begin(|handle, overlapped| { + // SAFETY: Operation retains the handle and OVERLAPPED through completion. + unsafe { ConnectNamedPipe(handle, overlapped) } + }) { + Ok(result) => result.is_some(), + // A client can connect (and even close) between CreateNamedPipe and + // ConnectNamedPipe. Hand off that connection; its reader sees EOF. + Err(error) if matches!(error.raw_os_error(), Some(code) if code == ERROR_PIPE_CONNECTED as i32 || code == ERROR_NO_DATA as i32) => { + true + } + Err(error) => return Err(error), + }; + Ok(Self { + operation, + connected, + }) + } + + fn ready(&mut self) -> io::Result { + if !self.connected { + self.connected = self.operation.poll()?.is_some(); + } + Ok(self.connected) + } +} + +struct Pipe { + handle: OwnedHandle, + cancelled: OwnedHandle, +} + +impl Pipe { + fn new(handle: OwnedHandle) -> io::Result { + Ok(Self { + handle, + cancelled: event()?, + }) + } + + fn cancel(&self) { + // SAFETY: this manual-reset event stays valid while any operation or + // cancellation handle holds the Pipe. Signalling it is idempotent. + unsafe { + SetEvent(self.cancelled.as_raw_handle()); + } + } +} + +pub(crate) struct Connection { + pipe: Arc, +} + +impl Connection { + pub fn connect(endpoint: &Path) -> io::Result { + let endpoint = endpoint_name(endpoint)?; + let deadline = Instant::now() + CONNECT_TIMEOUT; + loop { + // SECURITY_IDENTIFICATION prevents a rogue server from impersonating + // this client. Client handles are non-inheritable and overlapped too. + // SAFETY: endpoint is NUL-terminated; unused pointers are null. + let handle = unsafe { + CreateFileW( + endpoint.as_ptr(), + GENERIC_READ | GENERIC_WRITE, + 0, + null(), + OPEN_EXISTING, + FILE_FLAG_OVERLAPPED | SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION, + null_mut(), + ) + }; + match owned_handle(handle) { + Ok(handle) => { + return Ok(Self { + pipe: Arc::new(Pipe::new(handle)?), + }); + } + Err(error) if error.raw_os_error() == Some(ERROR_PIPE_BUSY as i32) => { + if Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "named pipe is busy", + )); + } + // SAFETY: endpoint is valid; each kernel wait is bounded. + unsafe { + WaitNamedPipeW(endpoint.as_ptr(), 10); + } + } + Err(error) => return Err(error), + } + } + } + + pub 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<()> { + match self.read_with_timeout(&mut [0], Some(COMPLETION_TIMEOUT)) { + Ok(0) => Ok(()), + Ok(_) => Err(io::Error::new( + io::ErrorKind::InvalidData, + "unexpected data after request", + )), + Err(error) => { + self.pipe.cancel(); + Err(error) + } + } + } + + pub fn monitor_disconnect(&self) -> io::Result { + let mut reader = Self { + pipe: Arc::clone(&self.pipe), + }; + let cancellation = self.cancellation()?; + let (sender, disconnected) = bounded(1); + let thread = thread::spawn(move || { + let _ = reader.read(&mut [0]); + let _ = sender.send(()); + }); + Ok(DisconnectMonitor { + disconnected, + cancellation, + thread, + }) + } + + fn read_with_timeout( + &mut self, + bytes: &mut [u8], + timeout: Option, + ) -> io::Result { + if bytes.is_empty() { + return Ok(0); + } + let mut operation = Operation::new(Arc::clone(&self.pipe))?; + let result = operation + .begin(|handle, overlapped| { + // SAFETY: bytes and the OVERLAPPED remain live until wait/drop reaps + // the operation. Only one reader is used on each end of a pipe. + unsafe { + ReadFile( + handle, + bytes.as_mut_ptr(), + bytes.len().min(BUFFER_BYTES as usize) as u32, + null_mut(), + overlapped, + ) + } + }) + .and_then(|ready| match ready { + Some(count) => Ok(count), + None => operation.wait(timeout), + }); + match result { + Ok(count) => Ok(count as usize), + Err(error) if matches!(error.raw_os_error(), Some(code) if code == ERROR_BROKEN_PIPE as i32 || code == ERROR_PIPE_NOT_CONNECTED as i32 || code == ERROR_NO_DATA as i32) => { + Ok(0) + } + Err(error) => Err(error), + } + } +} + +impl Read for Connection { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.read_with_timeout(bytes, None) + } +} + +impl Write for Connection { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if bytes.is_empty() { + return Ok(0); + } + let mut operation = Operation::new(Arc::clone(&self.pipe))?; + let ready = operation.begin(|handle, overlapped| { + // SAFETY: bytes and OVERLAPPED remain live through completion; at + // most BUFFER_BYTES are submitted, bounding each kernel I/O buffer. + unsafe { + WriteFile( + handle, + bytes.as_ptr(), + bytes.len().min(BUFFER_BYTES as usize) as u32, + null_mut(), + overlapped, + ) + } + })?; + Ok(match ready { + Some(count) => count, + None => operation.wait(None)?, + } as usize) + } + + fn flush(&mut self) -> io::Result<()> { + // No userspace buffering. FlushFileBuffers is NOT a cancellable stream + // flush and must never be used for framing or shutdown. + Ok(()) + } +} + +pub(crate) struct Cancellation(Arc); + +impl Cancellation { + pub fn cancel(&self) { + self.0.cancel(); + } +} + +pub(crate) struct DisconnectMonitor { + disconnected: Receiver<()>, + cancellation: Cancellation, + thread: JoinHandle<()>, +} + +impl DisconnectMonitor { + pub fn disconnected(&self) -> &Receiver<()> { + &self.disconnected + } + + pub fn finish(self) { + // No half-close exists. Protocol clients close after their final frame; + // retain queued bytes for them, but never wait indefinitely for closure. + if self.disconnected.recv_timeout(COMPLETION_TIMEOUT).is_err() { + self.cancellation.cancel(); + } + let _ = self.thread.join(); + } +} + +struct Operation { + pipe: Arc, + overlapped: Retained, + event: OwnedHandle, + pending: bool, +} + +impl Operation { + fn new(pipe: Arc) -> io::Result { + // SAFETY: the event is owned by pipe and zero timeout only queries it. + if unsafe { WaitForSingleObject(pipe.cancelled.as_raw_handle(), 0) } == WAIT_OBJECT_0 { + return Err(cancelled()); + } + let event = event()?; + let overlapped = Retained::new(OVERLAPPED { + hEvent: event.as_raw_handle(), + ..Default::default() + }); + Ok(Self { + pipe, + overlapped, + event, + pending: false, + }) + } + + fn begin( + &mut self, + submit: impl FnOnce(HANDLE, *mut OVERLAPPED) -> i32, + ) -> io::Result> { + self.pending = true; + if submit(self.pipe.handle.as_raw_handle(), self.overlapped.as_ptr()) != 0 { + return self.poll(); + } + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(ERROR_IO_PENDING as i32) { + Ok(None) + } else { + self.pending = false; + Err(error) + } + } + + fn poll(&mut self) -> io::Result> { + let mut count = 0; + // SAFETY: both handle and OVERLAPPED belong to this live operation. + if unsafe { + GetOverlappedResult( + self.pipe.handle.as_raw_handle(), + self.overlapped.as_ptr(), + &mut count, + 0, + ) + } != 0 + { + self.pending = false; + return Ok(Some(count)); + } + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(ERROR_IO_INCOMPLETE as i32) { + return Ok(None); + } + self.pending = false; + Err(error) + } + + fn wait(&mut self, timeout: Option) -> io::Result { + let handles = [ + self.pipe.cancelled.as_raw_handle(), + self.event.as_raw_handle(), + ]; + let milliseconds = timeout.map_or(INFINITE, |time| { + time.as_millis().min((INFINITE - 1) as u128) as u32 + }); + // SAFETY: both event handles remain live throughout this bounded or + // explicitly cancellable wait. Cancellation wins if both are signalled. + let result = unsafe { WaitForMultipleObjects(2, handles.as_ptr(), 0, milliseconds) }; + if result == WAIT_OBJECT_0 { + return Err(cancelled()); + } + if result == WAIT_TIMEOUT { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "named pipe completion timed out", + )); + } + if result != WAIT_OBJECT_0 + 1 { + return Err(io::Error::last_os_error()); + } + self.poll()? + .ok_or_else(|| io::Error::other("named pipe signalled before I/O completion")) + } +} + +impl Drop for Operation { + fn drop(&mut self) { + if self.pending { + // SAFETY: cancel THIS operation, then reap its completion before + // freeing the OVERLAPPED/event or returning the borrowed I/O buffer. + // CancelIoEx alone does not guarantee the kernel has stopped using it. + unsafe { + CancelIoEx(self.pipe.handle.as_raw_handle(), self.overlapped.as_ptr()); + let mut count = 0; + GetOverlappedResult( + self.pipe.handle.as_raw_handle(), + self.overlapped.as_ptr(), + &mut count, + 1, + ); + } + } + } +} + +fn event() -> io::Result { + // SAFETY: create an unnamed, non-inheritable, initially unset manual-reset event. + owned_handle(unsafe { CreateEventW(null(), 1, 0, null()) }) +} + +pub(crate) fn owned_handle(handle: HANDLE) -> io::Result { + if handle.is_null() || handle == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + // SAFETY: callers pass newly created, uniquely owned handles. + Ok(unsafe { OwnedHandle::from_raw_handle(handle) }) +} + +fn cancelled() -> io::Error { + // Not Interrupted: Read::read_exact retries Interrupted forever. + io::Error::new( + io::ErrorKind::ConnectionAborted, + "named pipe operation cancelled", + ) +} + +fn endpoint_name(endpoint: &Path) -> io::Result> { + let value = endpoint.as_os_str().encode_wide().collect::>(); + let prefix = r"\\.\pipe\opencode-pty-".encode_utf16().collect::>(); + if !value.starts_with(&prefix) + || value.len() == prefix.len() + || value[prefix.len()..] + .iter() + .any(|unit| *unit == 0 || *unit == b'\\' as u16 || *unit == b'/' as u16) + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "expected a local opencode-pty named-pipe endpoint", + )); + } + Ok(value.into_iter().chain([0]).collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::{Response, read_frame, write_frame}; + use std::path::PathBuf; + + fn endpoint() -> PathBuf { + PathBuf::from(format!( + r"\\.\pipe\opencode-pty-{:032x}", + rand::random::() + )) + } + + fn accept(listener: &mut Listener) -> Connection { + let deadline = Instant::now() + Duration::from_secs(3); + loop { + if let Some(connection) = listener.accept().unwrap() { + return connection; + } + assert!(Instant::now() < deadline, "named pipe accept timed out"); + thread::sleep(Duration::from_millis(1)); + } + } + + #[test] + fn multiple_byte_stream_connections_and_namespace_ownership() { + let endpoint = endpoint(); + let mut listener = Listener::bind(&endpoint).unwrap(); + assert!(listener.accept().unwrap().is_none()); + assert!( + Listener::bind(&endpoint).is_err(), + "must not join an existing server name" + ); + let mut first = Connection::connect(&endpoint).unwrap(); + let mut first_server = accept(&mut listener); + let mut second = Connection::connect(&endpoint).unwrap(); + let mut second_server = accept(&mut listener); + // Split writes are a byte stream, not Windows message-mode records. + first.write_all(b"ab").unwrap(); + first.write_all(b"cd").unwrap(); + second.write_all(b"xy").unwrap(); + let mut bytes = [0; 4]; + first_server.read_exact(&mut bytes).unwrap(); + assert_eq!(&bytes, b"abcd"); + second_server.read_exact(&mut bytes[..2]).unwrap(); + assert_eq!(&bytes[..2], b"xy"); + write_frame(&mut first_server, &Response::Owned).unwrap(); + assert!(matches!(read_frame(&mut first).unwrap(), Response::Owned)); + drop((first, first_server, second, second_server)); + assert!(Listener::bind(&endpoint).is_err()); + listener.stop(); + assert!(listener.accept().unwrap().is_none()); + assert!( + Listener::bind(&endpoint).is_err(), + "stop must retain the registered name" + ); + drop(listener); + Listener::bind(&endpoint).unwrap(); + } + + #[test] + fn cancellation_wakes_partial_request_reads() { + let endpoint = endpoint(); + let mut listener = Listener::bind(&endpoint).unwrap(); + let mut client = Connection::connect(&endpoint).unwrap(); + let mut server = accept(&mut listener); + let cancellation = server.cancellation().unwrap(); + client.write_all(&[0, 0]).unwrap(); + let (sender, receiver) = bounded(1); + let reader = thread::spawn(move || { + sender.send(read_frame::(&mut server)).unwrap(); + }); + cancellation.cancel(); + cancellation.cancel(); + assert!( + receiver + .recv_timeout(Duration::from_secs(3)) + .unwrap() + .is_err() + ); + reader.join().unwrap(); + } + + #[test] + fn cancellation_wakes_backpressured_writes() { + let endpoint = endpoint(); + let mut listener = Listener::bind(&endpoint).unwrap(); + let _client = Connection::connect(&endpoint).unwrap(); + let mut server = accept(&mut listener); + let cancellation = server.cancellation().unwrap(); + let (started, ready) = bounded(1); + let (sender, receiver) = bounded(1); + let writer = thread::spawn(move || { + started.send(()).unwrap(); + sender + .send(server.write_all(&[2; 2 * BUFFER_BYTES as usize])) + .unwrap(); + }); + ready.recv_timeout(Duration::from_secs(3)).unwrap(); + assert!( + receiver.recv_timeout(Duration::from_millis(50)).is_err(), + "write should be backpressured" + ); + cancellation.cancel(); + assert!( + receiver + .recv_timeout(Duration::from_secs(3)) + .unwrap() + .is_err() + ); + writer.join().unwrap(); + } + + #[test] + fn final_response_survives_until_client_reads_and_disconnects() { + let endpoint = endpoint(); + let mut listener = Listener::bind(&endpoint).unwrap(); + let mut client = Connection::connect(&endpoint).unwrap(); + let mut server = accept(&mut listener); + let (sender, receiver) = bounded(1); + let writer = thread::spawn(move || { + write_frame( + &mut server, + &Response::Exited { + exit_code: Some(0), + final_offset: 123, + }, + ) + .unwrap(); + sender.send(server.finish_response()).unwrap(); + }); + assert!( + receiver.recv_timeout(Duration::from_millis(50)).is_err(), + "server must retain final response" + ); + assert!(matches!( + read_frame(&mut client).unwrap(), + Response::Exited { + final_offset: 123, + .. + } + )); + drop(client); + receiver + .recv_timeout(Duration::from_secs(3)) + .unwrap() + .unwrap(); + writer.join().unwrap(); + } + + #[test] + fn subscription_completion_is_bounded_and_cancellable() { + let endpoint = endpoint(); + let mut listener = Listener::bind(&endpoint).unwrap(); + let _client = Connection::connect(&endpoint).unwrap(); + let server = accept(&mut listener); + let monitor = server.monitor_disconnect().unwrap(); + let start = Instant::now(); + monitor.finish(); + assert!(start.elapsed() < Duration::from_secs(4)); + + let _client = Connection::connect(&endpoint).unwrap(); + let server = accept(&mut listener); + let monitor = server.monitor_disconnect().unwrap(); + let cancellation = server.cancellation().unwrap(); + cancellation.cancel(); + monitor + .disconnected() + .recv_timeout(Duration::from_secs(3)) + .unwrap(); + monitor.finish(); + } + + #[test] + fn remote_and_non_pipe_endpoints_are_rejected() { + for endpoint in [ + r"\\remote\pipe\opencode-pty-test", + r"C:\tmp\pipe", + r"\\.\pipe\opencode-pty-test\nested", + ] { + assert!(Connection::connect(Path::new(endpoint)).is_err()); + assert!(Listener::bind(Path::new(endpoint)).is_err()); + } + } +} diff --git a/src/transport/windows/security.rs b/src/transport/windows/security.rs new file mode 100644 index 0000000..369c76f --- /dev/null +++ b/src/transport/windows/security.rs @@ -0,0 +1,108 @@ +use std::io; +use std::os::windows::io::AsRawHandle; +use std::ptr::null_mut; + +use windows_sys::Win32::Foundation::LocalFree; +use windows_sys::Win32::Security::Authorization::{ + ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1, +}; +use windows_sys::Win32::Security::{ + GetTokenInformation, SECURITY_ATTRIBUTES, TOKEN_QUERY, TOKEN_USER, TokenUser, +}; +use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; + +use super::owned_handle; + +pub(crate) struct PrivateSecurity(*mut core::ffi::c_void); + +impl PrivateSecurity { + pub fn new() -> io::Result { + let sid = current_user_sid()?; + // Protected DACL: only the current user, no inherited Everyone/network + // grants. Explicit owner also avoids inheriting an Administrators owner. + let sddl = format!("O:{sid}D:P(A;;GA;;;{sid})\0") + .encode_utf16() + .collect::>(); + let mut descriptor = null_mut(); + // SAFETY: valid NUL-terminated SDDL; successful allocation uses LocalFree. + if unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl.as_ptr(), + SDDL_REVISION_1, + &mut descriptor, + null_mut(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(Self(descriptor)) + } + + pub fn attributes(&self) -> SECURITY_ATTRIBUTES { + SECURITY_ATTRIBUTES { + nLength: size_of::() as u32, + lpSecurityDescriptor: self.0, + bInheritHandle: 0, + } + } +} + +impl Drop for PrivateSecurity { + fn drop(&mut self) { + // SAFETY: this is the unique descriptor allocation returned by conversion. + unsafe { + LocalFree(self.0); + } + } +} + +fn current_user_sid() -> io::Result { + let mut token = null_mut(); + // SAFETY: query the current process token; successful handle is adopted below. + if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut token) } == 0 { + return Err(io::Error::last_os_error()); + } + let token = owned_handle(token)?; + let mut length = 0; + // SAFETY: first call only determines the variable-sized token information buffer. + unsafe { + GetTokenInformation(token.as_raw_handle(), TokenUser, null_mut(), 0, &mut length); + } + if length == 0 { + return Err(io::Error::last_os_error()); + } + // usize provides the alignment TOKEN_USER requires, unlike Vec. + let mut buffer = vec![0_usize; (length as usize).div_ceil(size_of::())]; + // SAFETY: buffer is aligned, large enough, and retained through SID conversion. + if unsafe { + GetTokenInformation( + token.as_raw_handle(), + TokenUser, + buffer.as_mut_ptr().cast(), + length, + &mut length, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + // SAFETY: a successful TokenUser query initializes a TOKEN_USER at this address. + let user = unsafe { &*buffer.as_ptr().cast::() }; + let mut text = null_mut(); + // SAFETY: the queried token owns a valid SID; returned string uses LocalFree. + if unsafe { ConvertSidToStringSidW(user.User.Sid, &mut text) } == 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: conversion returns a NUL-terminated wide string allocated by Windows. + let sid = unsafe { + let mut length = 0; + while *text.add(length) != 0 { + length += 1; + } + let sid = String::from_utf16_lossy(std::slice::from_raw_parts(text, length)); + LocalFree(text.cast()); + sid + }; + Ok(sid) +} diff --git a/tests/transport-retained.rs b/tests/transport-retained.rs new file mode 100644 index 0000000..f050d67 --- /dev/null +++ b/tests/transport-retained.rs @@ -0,0 +1,49 @@ +#[path = "../src/transport/retained.rs"] +mod retained; + +use std::cell::Cell; +use std::rc::Rc; + +use retained::Retained; + +struct Completion { + bytes: usize, + drops: Rc>, +} + +impl Drop for Completion { + fn drop(&mut self) { + self.drops.set(self.drops.get() + 1); + } +} + +#[test] +fn retained_native_pointer_survives_owner_moves_and_repeated_access() { + let drops = Rc::new(Cell::new(0)); + let operation = Retained::new(Completion { + bytes: 0, + drops: Rc::clone(&drops), + }); + let native = operation.as_ptr(); + let mut pending = vec![operation]; + pending.reserve(100); // Move the owner while native code retains its pointer. + for count in 1..=10 { + let polled = pending[0].as_ptr(); + assert_eq!(native, polled); + // SAFETY: stand-in for native mutation and a completion query. No Rust + // reference to the allocation is created while its raw pointer is retained. + unsafe { + (*native).bytes = count; + assert_eq!((*polled).bytes, count); + } + } + let completed = pending.pop().unwrap(); + assert_eq!(drops.get(), 0); + // Reclaim only after the native operation has completed (or been cancelled + // and reaped). Miri checks that reclamation and the final access are valid. + unsafe { + assert_eq!((*native).bytes, 10); + } + drop(completed); + assert_eq!(drops.get(), 1); +} diff --git a/tests/transport-retained/.gitignore b/tests/transport-retained/.gitignore new file mode 100644 index 0000000..b83d222 --- /dev/null +++ b/tests/transport-retained/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/tests/transport-retained/Cargo.lock b/tests/transport-retained/Cargo.lock new file mode 100644 index 0000000..e98adc0 --- /dev/null +++ b/tests/transport-retained/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "transport-retained-miri" +version = "0.0.0" diff --git a/tests/transport-retained/Cargo.toml b/tests/transport-retained/Cargo.toml new file mode 100644 index 0000000..a077a41 --- /dev/null +++ b/tests/transport-retained/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "transport-retained-miri" +version = "0.0.0" +edition = "2024" +publish = false + +[lib] +path = "../transport-retained.rs" +doctest = false + +[workspace]