kunobi-daemon 0.10.0

//! Windows named-pipe adapters with peer checks and setup deadlines.
//!
//! Syscall import libs use lowercase names (`kernel32`, `advapi32`). cargo-xwin's
//! `lld-link` on Linux opens `Advapi32.lib` as a file and cannot see the xwin
//! splat's `advapi32.lib`. Native MSVC is case-insensitive, so lowercase is
//! valid on both.

use std::cell::Cell;
use std::ffi::c_void;
use std::io::{self, Read, Write};
use std::os::windows::io::{AsHandle, AsRawHandle};
use std::ptr;
use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};
use std::time::{Duration, Instant};

use interprocess::local_socket::traits::Stream as _;
use interprocess::local_socket::{ConnectOptions, GenericNamespaced, Stream, ToNsName};

use super::{ConnectError, Endpoint};
use crate::local::Duplex;
use crate::transport::WriteHalf;

/// Open or create a lock file, refusing a symlink or junction in its place.
///
/// The refusal is `PermissionDenied`. Access is not checked: a new file
/// inherits the directory's ACL, and an existing one keeps its own.
pub(crate) fn open_private(path: &std::path::Path) -> io::Result<std::fs::File> {
    use std::os::windows::fs::OpenOptionsExt;
    // Opens a reparse point itself instead of what it names.
    const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
    let file = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT)
        .open(path)?;
    if file.metadata()?.file_type().is_symlink() {
        return Err(io::ErrorKind::PermissionDenied.into());
    }
    Ok(file)
}

/// A named-pipe connection with independent read and write ownership.
pub struct WindowsDuplex {
    stream: Stream,
    read_deadline: Cell<Option<Instant>>,
}

/// Whether `pid` runs as this process's token user. `Ok(false)` only after a
/// successful SID comparison; failing to open or query the process is an error.
///
/// The process is looked up by PID, so if the connection's original peer has
/// exited and another process now has its PID, this describes that process.
pub(crate) fn process_runs_as_this_user(pid: crate::ProcessId) -> io::Result<bool> {
    // SAFETY: OpenProcess takes integers and returns a handle or null; query
    // access only.
    let process = unsafe { OpenProcess(0x1000, 0, pid.get()) };
    if process.is_null() {
        return Err(io::Error::last_os_error());
    }
    let process = Event(process);
    let peer = token_user(process.0)?;
    // SAFETY: GetCurrentProcess returns a borrowed pseudo-handle.
    let own = token_user(unsafe { GetCurrentProcess() })?;
    Ok(same_sid(&peer, &own))
}

/// Whether the client of the server-end pipe `pipe` runs as this process's
/// token user, read from the connection's own security context.
///
/// The thread impersonates the client, reads its token's user and goes back to
/// the token it had before, on every path. Under dynamic context tracking
/// Windows records a client's context each time it writes, so before the
/// client's first bytes arrive this fails with `ERROR_CANNOT_IMPERSONATE`; a
/// client that allows only anonymous impersonation gives
/// `ERROR_CANT_OPEN_ANONYMOUS`. `Ok(false)` only after a successful SID
/// comparison.
pub(crate) fn pipe_client_runs_as_this_user(pipe: Handle) -> io::Result<bool> {
    // Read first: as the client, this thread may not be allowed to open this
    // process's token.
    // SAFETY: GetCurrentProcess returns a borrowed pseudo-handle.
    let own = token_user(unsafe { GetCurrentProcess() })?;
    let client = {
        let _client = Impersonating::pipe_client(pipe)?;
        thread_token_user()?
    };
    Ok(same_sid(&client, &own))
}

/// Whether two `TOKEN_USER` buffers name the same SID.
fn same_sid(first: &[usize], second: &[usize]) -> bool {
    // SAFETY: TOKEN_USER starts with a SID_AND_ATTRIBUTES whose first member is
    // the SID pointer, so element 0 of each buffer is that pointer. Both
    // buffers, and the SIDs the kernel wrote inside them, are borrowed for this
    // call. EqualSid only reads them.
    unsafe { EqualSid(first[0] as *const c_void, second[0] as *const c_void) != 0 }
}

/// This thread runs as a named-pipe client until the guard drops, then goes
/// back to the token it had before: none, or the one it was impersonating.
///
/// Dropping restores the token on every path, a panic included. If Windows
/// refuses, the process aborts: a thread left running as the client would run
/// everything after this with the client's identity.
struct Impersonating {
    previous: Option<Event>,
}

impl Impersonating {
    fn pipe_client(pipe: Handle) -> io::Result<Self> {
        let previous = thread_token(TOKEN_QUERY | TOKEN_IMPERSONATE)?;
        // SAFETY: `pipe` is the caller's live server-end pipe handle; the call
        // only changes this thread's token.
        if unsafe { ImpersonateNamedPipeClient(pipe) } == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(Self { previous })
    }
}

impl Drop for Impersonating {
    fn drop(&mut self) {
        let restored = match &self.previous {
            // SAFETY: a null thread pointer names this thread, and `token` is
            // the live token it had before, opened with TOKEN_IMPERSONATE.
            Some(token) => unsafe { SetThreadToken(ptr::null_mut(), token.0) },
            // SAFETY: ends this thread's impersonation; it had no token before.
            None => unsafe { RevertToSelf() },
        };
        if restored == 0 {
            std::process::abort();
        }
    }
}

const TOKEN_IMPERSONATE: u32 = 0x0004;
const TOKEN_QUERY: u32 = 0x0008;
const ERROR_NO_TOKEN: i32 = 1008;

/// The token this thread is impersonating, or `None` when it has none.
///
/// Access is checked against the process's token (`OpenAsSelf`): a client's
/// token at identification level could not open anything.
fn thread_token(access: u32) -> io::Result<Option<Event>> {
    let mut token = ptr::null_mut();
    // SAFETY: GetCurrentThread returns this thread's pseudo-handle and `token`
    // is a valid output slot.
    if unsafe { OpenThreadToken(GetCurrentThread(), access, 1, &mut token) } != 0 {
        return Ok(Some(Event(token)));
    }
    let error = io::Error::last_os_error();
    if error.raw_os_error() == Some(ERROR_NO_TOKEN) {
        Ok(None)
    } else {
        Err(error)
    }
}

/// The user of the token this thread is impersonating.
fn thread_token_user() -> io::Result<Vec<usize>> {
    let token =
        thread_token(TOKEN_QUERY)?.ok_or_else(|| io::Error::from_raw_os_error(ERROR_NO_TOKEN))?;
    query_token_user(&token)
}

/// One-time compatibility fallback for a protocol-v1 peer that cannot drain:
/// terminate it.
///
/// This reaches whichever process has `pid` now. Take the PID from the
/// credentials of a live connection to that peer, and prefer draining: a PID
/// can be reused once its process has exited.
pub fn terminate_legacy_peer(pid: crate::ProcessId) -> io::Result<()> {
    let pid = pid.get();
    use std::ffi::c_void;
    type Handle = *mut c_void;
    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn OpenProcess(access: u32, inherit: i32, pid: u32) -> Handle;
        fn TerminateProcess(process: Handle, exit_code: u32) -> i32;
        fn CloseHandle(object: Handle) -> i32;
    }
    const PROCESS_TERMINATE: u32 = 0x0001;
    // SAFETY: OpenProcess takes integers and returns a handle or null.
    let process = unsafe { OpenProcess(PROCESS_TERMINATE, 0, pid) };
    if process.is_null() {
        return Err(io::Error::last_os_error());
    }
    // SAFETY: `process` is a valid handle returned immediately above.
    let terminated = unsafe { TerminateProcess(process, 0) };
    // SAFETY: this closes exactly the handle opened above.
    let _ = unsafe { CloseHandle(process) };
    if terminated != 0 {
        Ok(())
    } else {
        Err(io::Error::last_os_error())
    }
}

/// Overlapped named-pipe receive half with a setup deadline.
pub struct WindowsReader {
    inner: interprocess::local_socket::RecvHalf,
    read_deadline: Option<Instant>,
    setup: Arc<AtomicBool>,
}
/// Overlapped named-pipe send half with a shared setup deadline.
pub struct WindowsWriter {
    inner: interprocess::local_socket::SendHalf,
    deadline: Option<Instant>,
    setup: Arc<AtomicBool>,
}

impl WindowsReader {
    /// Return to normal blocking named-pipe reads for the steady-state pump.
    /// Restore ordinary unbounded I/O after setup succeeds.
    pub fn clear_read_deadline(&mut self) {
        self.read_deadline = None;
        self.setup.store(false, Ordering::Relaxed);
    }

    fn raw_handle(&self) -> *mut c_void {
        match &self.inner {
            interprocess::local_socket::RecvHalf::NamedPipe(pipe) => {
                pipe.as_handle().as_raw_handle()
            }
        }
    }
}

impl Read for WindowsReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let Some(deadline) = self.read_deadline else {
            return self.inner.read(buf);
        };
        let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
            return Err(io::Error::new(
                io::ErrorKind::TimedOut,
                "named-pipe session establishment timed out",
            ));
        };
        read_overlapped(self.raw_handle(), buf, remaining)
    }
}

impl Write for WindowsWriter {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        if self.setup.load(Ordering::Relaxed) {
            let remaining = self
                .deadline
                .and_then(|deadline| deadline.checked_duration_since(Instant::now()))
                .ok_or_else(|| io::Error::from(io::ErrorKind::TimedOut))?;
            let handle = match &self.inner {
                interprocess::local_socket::SendHalf::NamedPipe(pipe) => {
                    pipe.as_handle().as_raw_handle()
                }
            };
            write_overlapped(handle, buf, remaining)
        } else {
            self.inner.write(buf)
        }
    }
    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

impl WriteHalf for WindowsWriter {
    /// Flush, then report that the peer was not told: a named pipe has no
    /// half-close, and closing this handle would not end the stream while the
    /// reader still holds the pipe. End the session with
    /// [`crate::transport::Outstanding`] instead.
    fn shutdown_write(&mut self) -> io::Result<()> {
        self.flush()?;
        Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "a named pipe has no half-close",
        ))
    }
}

impl super::peer::PeerCredentials for WindowsDuplex {
    fn credentials(&self) -> io::Result<crate::peer::Credentials> {
        super::peer::PeerCredentials::credentials(&self.stream)
    }
}

/// Every instance of the pipe is serving another client. The standard library
/// maps this outside the portable I/O categories.
pub(crate) fn busy_pipe(error: &io::Error) -> bool {
    const ERROR_PIPE_BUSY: i32 = 231;
    error.raw_os_error() == Some(ERROR_PIPE_BUSY)
}

impl Duplex for WindowsDuplex {
    type Reader = WindowsReader;
    type Writer = WindowsWriter;

    fn connect_once_until(endpoint: &Endpoint, deadline: Instant) -> Result<Self, ConnectError> {
        // A malformed name can never connect, so it is not worth retrying.
        let name = endpoint
            .to_ns_name::<GenericNamespaced>()
            .map_err(|error| ConnectError::Failed(error.kind()))?;
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            return Err(ConnectError::ConnectTimeout);
        }
        ConnectOptions::new()
            .name(name)
            .wait_mode(interprocess::ConnectWaitMode::Timeout(Duration::ZERO))
            .connect_sync()
            .map(|stream| Self {
                stream,
                read_deadline: Cell::new(None),
            })
            .map_err(|error| super::connect_error(&error))
    }

    /// Arm an absolute deadline for reads performed during session setup.
    ///
    /// The split reader enforces this with overlapped `ReadFile`, then the
    /// caller clears it before handing the reader to the ordinary byte pump.
    /// Set or clear the absolute setup deadline before splitting.
    fn set_read_deadline(&self, timeout: Option<Duration>) -> io::Result<()> {
        let deadline = timeout
            .map(|duration| {
                Instant::now().checked_add(duration).ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidInput, "read deadline overflow")
                })
            })
            .transpose()?;
        self.read_deadline.set(deadline);
        Ok(())
    }

    fn split(self) -> io::Result<(Self::Reader, Self::Writer)> {
        let deadline = self.read_deadline.get();
        let (read, write) = self.stream.split();
        let setup = Arc::new(AtomicBool::new(deadline.is_some()));
        Ok((
            WindowsReader {
                inner: read,
                read_deadline: deadline,
                setup: Arc::clone(&setup),
            },
            WindowsWriter {
                inner: write,
                deadline,
                setup,
            },
        ))
    }
}

type Handle = *mut c_void;

#[repr(C)]
#[derive(Clone, Copy)]
struct OverlappedOffset {
    offset: u32,
    offset_high: u32,
}

#[repr(C)]
union OverlappedPosition {
    offset: OverlappedOffset,
    pointer: *mut c_void,
}

#[repr(C)]
struct Overlapped {
    internal: usize,
    internal_high: usize,
    position: OverlappedPosition,
    event: Handle,
}

#[link(name = "kernel32")]
unsafe extern "system" {
    fn CreateEventW(
        event_attributes: *const c_void,
        manual_reset: i32,
        initial_state: i32,
        name: *const u16,
    ) -> Handle;
    fn ReadFile(
        file: Handle,
        buffer: *mut c_void,
        bytes_to_read: u32,
        bytes_read: *mut u32,
        overlapped: *mut Overlapped,
    ) -> i32;
    fn WriteFile(
        file: Handle,
        buffer: *const c_void,
        length: u32,
        written: *mut u32,
        overlapped: *mut Overlapped,
    ) -> i32;
    fn OpenProcess(access: u32, inherit: i32, pid: u32) -> Handle;
    fn GetCurrentProcess() -> Handle;
    fn WaitForSingleObject(handle: Handle, milliseconds: u32) -> u32;
    fn CancelIoEx(file: Handle, overlapped: *const Overlapped) -> i32;
    fn GetOverlappedResult(
        file: Handle,
        overlapped: *mut Overlapped,
        bytes_transferred: *mut u32,
        wait: i32,
    ) -> i32;
    fn CloseHandle(object: Handle) -> i32;
    fn SetConsoleCtrlHandler(
        handler: Option<unsafe extern "system" fn(u32) -> i32>,
        add: i32,
    ) -> i32;
    fn GetHandleInformation(object: Handle, flags: *mut u32) -> i32;
    fn SetHandleInformation(object: Handle, mask: u32, flags: u32) -> i32;
    fn LocalFree(memory: *mut c_void) -> *mut c_void;
    fn GetCurrentThread() -> Handle;
}

#[link(name = "advapi32")]
unsafe extern "system" {
    fn OpenProcessToken(process: Handle, access: u32, token: *mut Handle) -> i32;
    fn GetTokenInformation(
        token: Handle,
        class: i32,
        info: *mut c_void,
        length: u32,
        needed: *mut u32,
    ) -> i32;
    fn EqualSid(first: *const c_void, second: *const c_void) -> i32;
    fn ConvertSidToStringSidW(sid: *mut c_void, text: *mut *mut u16) -> i32;
    fn ImpersonateNamedPipeClient(pipe: Handle) -> i32;
    fn OpenThreadToken(thread: Handle, access: u32, open_as_self: i32, token: *mut Handle) -> i32;
    fn SetThreadToken(thread: *mut Handle, token: Handle) -> i32;
    fn RevertToSelf() -> i32;
}

/// Memory the OS allocated with `LocalAlloc`.
struct LocalMemory(*mut c_void);

impl Drop for LocalMemory {
    fn drop(&mut self) {
        // SAFETY: the pointer came from an API that allocates with LocalAlloc
        // and is freed exactly once, here.
        unsafe {
            LocalFree(self.0);
        }
    }
}

/// This process's token user as a SID string, such as `S-1-5-21-…-1001`.
pub(crate) fn current_user_sid() -> io::Result<String> {
    // SAFETY: GetCurrentProcess returns a borrowed pseudo-handle.
    let user = token_user(unsafe { GetCurrentProcess() })?;
    let mut text: *mut u16 = ptr::null_mut();
    // SAFETY: TOKEN_USER starts with the SID pointer, so element 0 of `user`
    // points at the SID inside that live buffer; `text` is a valid output slot.
    if unsafe { ConvertSidToStringSidW(user[0] as *mut c_void, &mut text) } == 0 {
        return Err(io::Error::last_os_error());
    }
    let text = LocalMemory(text.cast());
    let sid = text.0.cast::<u16>();
    let mut length = 0;
    // SAFETY: the string is NUL-terminated, so every index up to and including
    // the NUL is inside it.
    while unsafe { *sid.add(length) } != 0 {
        length += 1;
    }
    // SAFETY: the `length` units before the NUL are initialised and live until
    // `text` is freed at the end of this function.
    let sid = unsafe { std::slice::from_raw_parts(sid, length) };
    String::from_utf16(sid).map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
}

/// The user of `process`'s token.
fn token_user(process: Handle) -> io::Result<Vec<usize>> {
    let mut token = ptr::null_mut();
    // SAFETY: process is live and token is a valid output slot.
    if unsafe { OpenProcessToken(process, TOKEN_QUERY, &mut token) } == 0 {
        return Err(io::Error::last_os_error());
    }
    query_token_user(&Event(token))
}

/// `TOKEN_USER` for `token`, opened with `TOKEN_QUERY`.
fn query_token_user(token: &Event) -> io::Result<Vec<usize>> {
    let mut needed = 0;
    // SAFETY: `token` is a live token handle and TokenUser is class 1. A null
    // buffer with length 0 is the documented size query: the call writes only
    // through `needed` and returns FALSE with ERROR_INSUFFICIENT_BUFFER, which
    // is the 122 checked below.
    let result = unsafe { GetTokenInformation(token.0, 1, ptr::null_mut(), 0, &mut needed) };
    if result != 0
        || io::Error::last_os_error().raw_os_error() != Some(122)
        || needed < std::mem::size_of::<usize>() as u32
        || needed > 65536
    {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "cannot size peer token",
        ));
    }
    // A Vec<usize> gives TOKEN_USER's pointer alignment, which Vec<u8> does not.
    let mut buffer = vec![0usize; (needed as usize).div_ceil(std::mem::size_of::<usize>())];
    // SAFETY: the buffer holds at least `needed` bytes, the size the previous
    // call asked for, and is aligned for the pointer TOKEN_USER starts with.
    // The kernel writes at most `needed` bytes into it.
    if unsafe { GetTokenInformation(token.0, 1, buffer.as_mut_ptr().cast(), needed, &mut needed) }
        == 0
    {
        return Err(io::Error::last_os_error());
    }
    Ok(buffer)
}

const ERROR_IO_PENDING: i32 = 997;
const ERROR_NOT_FOUND: i32 = 1168;
const WAIT_OBJECT_0: u32 = 0;
const WAIT_TIMEOUT: u32 = 258;

#[cfg(target_pointer_width = "64")]
const _: [(); 32] = [(); std::mem::size_of::<Overlapped>()];
#[cfg(target_pointer_width = "32")]
const _: [(); 20] = [(); std::mem::size_of::<Overlapped>()];

struct Event(Handle);

impl Drop for Event {
    fn drop(&mut self) {
        // SAFETY: this closes exactly one owned event, process or token handle.
        let _ = unsafe { CloseHandle(self.0) };
    }
}

fn deadline_millis(timeout: Duration) -> u32 {
    timeout.as_millis().clamp(1, u128::from(u32::MAX)) as u32
}

/// Perform one cancelable named-pipe read without changing steady-state mode.
fn read_overlapped(handle: Handle, buf: &mut [u8], timeout: Duration) -> io::Result<usize> {
    transfer_overlapped(handle, buf.len(), timeout, |operation, len| {
        // SAFETY: the mutable buffer lives until transfer_overlapped reaps this operation.
        unsafe {
            ReadFile(
                handle,
                buf.as_mut_ptr().cast(),
                len,
                ptr::null_mut(),
                operation,
            )
        }
    })
}

fn write_overlapped(handle: Handle, buf: &[u8], timeout: Duration) -> io::Result<usize> {
    transfer_overlapped(handle, buf.len(), timeout, |operation, len| {
        // SAFETY: the immutable buffer lives until completion or cancellation is reaped.
        unsafe { WriteFile(handle, buf.as_ptr().cast(), len, ptr::null_mut(), operation) }
    })
}

fn transfer_overlapped(
    handle: Handle,
    length: usize,
    timeout: Duration,
    start: impl FnOnce(*mut Overlapped, u32) -> i32,
) -> io::Result<usize> {
    if length == 0 {
        return Ok(0);
    }
    // SAFETY: a null security descriptor takes the default, and a null name
    // creates an unnamed event, both documented. The manual-reset and
    // initial-state arguments are plain booleans, so nothing is dereferenced.
    let event = Event(unsafe { CreateEventW(ptr::null(), 1, 0, ptr::null()) });
    if event.0.is_null() {
        return Err(io::Error::last_os_error());
    }
    // SAFETY: OVERLAPPED, its event and the caller's buffer remain alive until reaped.
    let mut operation: Overlapped = unsafe { std::mem::zeroed() };
    operation.event = event.0;
    let mut bytes_read = 0u32;
    let started = start(&mut operation, u32::try_from(length).unwrap_or(u32::MAX));
    if started != 0 {
        // An overlapped operation may complete inline. Ask the kernel for the
        // byte count instead of relying on `lpNumberOfBytesRead`, which must be
        // null for asynchronous handles.
        //
        // SAFETY: `operation` and `bytes_read` are stack locals that outlive
        // this call, and `handle` is the one the operation was started on.
        // `wait = 0` returns immediately rather than leaving the OVERLAPPED
        // pending past this frame.
        let completed = unsafe { GetOverlappedResult(handle, &mut operation, &mut bytes_read, 0) };
        return if completed != 0 {
            Ok(bytes_read as usize)
        } else {
            Err(io::Error::last_os_error())
        };
    }
    let start_error = io::Error::last_os_error();
    if start_error.raw_os_error() != Some(ERROR_IO_PENDING) {
        return Err(start_error);
    }

    // SAFETY: `event` is a live waitable event handle.
    match unsafe { WaitForSingleObject(event.0, deadline_millis(timeout)) } {
        WAIT_OBJECT_0 => {
            // SAFETY: the event signaled completion of this exact operation.
            let completed =
                unsafe { GetOverlappedResult(handle, &mut operation, &mut bytes_read, 0) };
            if completed != 0 {
                Ok(bytes_read as usize)
            } else {
                Err(io::Error::last_os_error())
            }
        }
        WAIT_TIMEOUT => {
            cancel_and_reap(handle, &mut operation, &mut bytes_read)?;
            Err(io::Error::new(
                io::ErrorKind::TimedOut,
                "named-pipe session establishment timed out",
            ))
        }
        _ => {
            let wait_error = io::Error::last_os_error();
            // Even a failed wait does not prove the OVERLAPPED operation is
            // terminal. Cancel and collect it before returning the wait error.
            cancel_and_reap(handle, &mut operation, &mut bytes_read)?;
            Err(wait_error)
        }
    }
}

/// Cancel pending I/O and keep its stack storage alive until the kernel has
/// reached a terminal state.
fn cancel_and_reap(
    handle: Handle,
    operation: &mut Overlapped,
    bytes_read: &mut u32,
) -> io::Result<()> {
    // SAFETY: cancellation targets the exact live operation supplied here.
    let cancelled = unsafe { CancelIoEx(handle, operation) };
    let cancel_error = if cancelled == 0 {
        Some(io::Error::last_os_error())
    } else {
        None
    };

    // SAFETY: `wait = true` does not return while this OVERLAPPED remains
    // pending. This call is mandatory even when cancellation reports an
    // unexpected error: returning earlier could let the kernel write into the
    // caller's released buffer or this released stack structure.
    let _ = unsafe { GetOverlappedResult(handle, operation, bytes_read, 1) };

    match cancel_error {
        // The operation completed between the event wait and cancellation.
        // `GetOverlappedResult` above still establishes terminal completion.
        Some(error) if error.raw_os_error() == Some(ERROR_NOT_FOUND) => Ok(()),
        Some(error) => Err(error),
        None => Ok(()),
    }
}

static SESSION_END: AtomicBool = AtomicBool::new(false);

const CTRL_CLOSE_EVENT: u32 = 2;
const CTRL_LOGOFF_EVENT: u32 = 5;
const CTRL_SHUTDOWN_EVENT: u32 = 6;

unsafe extern "system" fn session_end_handler(event: u32) -> i32 {
    if matches!(
        event,
        CTRL_CLOSE_EVENT | CTRL_LOGOFF_EVENT | CTRL_SHUTDOWN_EVENT
    ) {
        SESSION_END.store(true, Ordering::Release);
        1
    } else {
        0
    }
}

/// Arm CLOSE/LOGOFF/SHUTDOWN so a published pipe is not left after logoff.
pub fn install_session_end_handler() -> io::Result<()> {
    // SAFETY: `session_end_handler` is process-lifetime and only stores an
    // atomic, which is safe on the system-created callback thread.
    if unsafe { SetConsoleCtrlHandler(Some(session_end_handler), 1) } != 0 {
        Ok(())
    } else {
        Err(io::Error::last_os_error())
    }
}

/// True after CLOSE, LOGOFF or SHUTDOWN once [`install_session_end_handler`] ran,
/// or once a `shutdown_signal` future completed on one of them.
pub fn session_end_requested() -> bool {
    SESSION_END.load(Ordering::Acquire)
}

/// Wait for this process to be asked to stop: Ctrl+C, or the console closing,
/// the user logging off or the system shutting down.
///
/// This listens to Tokio's console streams for those four events and creates
/// them before it returns, so a request that arrives afterwards is not lost
/// even if the future has not been polled yet. When it completes on a close,
/// logoff or shutdown it also sets [`session_end_requested`].
///
/// Windows ends the process as soon as a console handler returns for close,
/// logoff or shutdown. Tokio's handler does not return for an event that had
/// a live listener, so the process keeps running until it exits on its own or
/// until Windows' timeout for the event ends it. With no listener left, Tokio
/// passes the event on and Windows ends the process as usual.
///
/// This does not use [`install_session_end_handler`]: Tokio's handler is the
/// only one involved. Tokio registers it once per process and never again, so
/// if a console change (`FreeConsole`, then `AllocConsole` or `AttachConsole`)
/// drops it, these events no longer reach the future.
///
/// Call it inside a Tokio runtime.
#[cfg(feature = "async")]
pub fn shutdown_signal() -> io::Result<impl Future<Output = ()> + Send + 'static> {
    use tokio::signal::windows::{ctrl_c, ctrl_close, ctrl_logoff, ctrl_shutdown};
    let mut interrupt = ctrl_c()?;
    let mut close = ctrl_close()?;
    let mut logoff = ctrl_logoff()?;
    let mut shutdown = ctrl_shutdown()?;
    Ok(async move {
        tokio::select! {
            _ = interrupt.recv() => return,
            _ = close.recv() => {}
            _ = logoff.recv() => {}
            _ = shutdown.recv() => {}
        }
        SESSION_END.store(true, Ordering::Release);
    })
}

const HANDLE_FLAG_INHERIT: u32 = 0x00000001;
const INVALID_HANDLE_VALUE: Handle = -1isize as Handle;

/// Clears inherit on this process's standard handles and restores it on drop.
///
/// Explicit child stdio is unaffected: the standard library marks those
/// handles inheritable. This removes incidental inheritance of the caller's
/// pipes across `Command::spawn`.
pub struct StdioInheritGuard {
    restore: Vec<Handle>,
}

impl StdioInheritGuard {
    /// Suppress inherit for the duration of a spawn. Reading or changing a
    /// live standard handle's flags can fail; partial changes are restored.
    pub fn suppress() -> io::Result<Self> {
        use std::os::windows::io::AsRawHandle;
        Self::for_handles([
            std::io::stdin().as_raw_handle(),
            std::io::stdout().as_raw_handle(),
            std::io::stderr().as_raw_handle(),
        ])
    }

    fn for_handles(handles: [Handle; 3]) -> io::Result<Self> {
        let mut guard = Self {
            restore: Vec::new(),
        };
        for handle in handles {
            if handle.is_null() || handle == INVALID_HANDLE_VALUE {
                continue;
            }
            let mut flags = 0;
            // SAFETY: handle is live for this guard; flags is writable.
            if unsafe { GetHandleInformation(handle, &mut flags) } == 0 {
                return Err(io::Error::last_os_error());
            }
            if flags & HANDLE_FLAG_INHERIT == 0 {
                continue;
            }
            // SAFETY: the handle is live; change only its inherit flag.
            if unsafe { SetHandleInformation(handle, HANDLE_FLAG_INHERIT, 0) } == 0 {
                return Err(io::Error::last_os_error());
            }
            guard.restore.push(handle);
        }
        Ok(guard)
    }

    /// Restore the original flags and report a failure. Drop also attempts
    /// restoration, but cannot return an error.
    pub fn restore(&mut self) -> io::Result<()> {
        let mut failed = None;
        self.restore.retain(|handle| {
            // SAFETY: these live handles were originally inheritable; restore
            // only that flag. Keep failures for another attempt on drop.
            if unsafe { SetHandleInformation(*handle, HANDLE_FLAG_INHERIT, HANDLE_FLAG_INHERIT) }
                != 0
            {
                false
            } else {
                failed.get_or_insert_with(io::Error::last_os_error);
                true
            }
        });
        match failed {
            Some(error) => Err(error),
            None => Ok(()),
        }
    }
}

impl Drop for StdioInheritGuard {
    fn drop(&mut self) {
        let _ = self.restore();
    }
}

#[cfg(test)]
mod stdio_inheritance_tests {
    use super::*;
    use std::os::windows::io::AsRawHandle;

    #[test]
    fn restores_only_the_handles_that_were_inheritable() {
        let (read, write) = std::io::pipe().unwrap();
        let inherited = read.as_raw_handle();
        let private = write.as_raw_handle();
        // SAFETY: both pipe handles are live; change only their inherit flags.
        unsafe {
            assert_ne!(
                SetHandleInformation(inherited, HANDLE_FLAG_INHERIT, HANDLE_FLAG_INHERIT),
                0
            );
            assert_ne!(SetHandleInformation(private, HANDLE_FLAG_INHERIT, 0), 0);
        }
        fn flags(handle: Handle) -> u32 {
            let mut flags = 0;
            // SAFETY: the caller keeps this pipe handle open; flags is writable.
            assert_ne!(unsafe { GetHandleInformation(handle, &mut flags) }, 0);
            flags & HANDLE_FLAG_INHERIT
        }
        {
            let _guard = StdioInheritGuard::for_handles([inherited, private, inherited]).unwrap();
            assert_eq!(flags(inherited), 0);
            assert_eq!(flags(private), 0);
        }
        assert_eq!(flags(inherited), HANDLE_FLAG_INHERIT);
        assert_eq!(flags(private), 0);
    }
}

/// Every readiness pipe's name: this prefix and 32 lowercase hex digits, 128
/// random bits.
pub(crate) const READY_PIPE_PREFIX: &str = r"\\.\pipe\kunobi-daemon-ready-";

/// Whether `name` is a readiness pipe name as the launcher makes them.
fn is_ready_pipe_name(name: &str) -> bool {
    name.strip_prefix(READY_PIPE_PREFIX).is_some_and(|id| {
        id.len() == 32
            && id
                .bytes()
                .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
    })
}

/// Open the daemon's end of the readiness pipe named `name`.
///
/// Only a name of the launcher's own form is opened, so a stale or foreign
/// value cannot make the daemon write into some other file. The daemon
/// connects with identification-level impersonation only, and its handle is
/// not inheritable, so programs it starts do not hold the channel open.
pub(crate) fn open_ready_pipe(name: &str) -> io::Result<std::io::PipeWriter> {
    use std::os::windows::fs::OpenOptionsExt;
    use std::os::windows::io::OwnedHandle;
    const SECURITY_IDENTIFICATION: u32 = 1 << 16;
    if !is_ready_pipe_name(name) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "the readiness channel is not a readiness pipe name",
        ));
    }
    let pipe = std::fs::OpenOptions::new()
        .write(true)
        .security_qos_flags(SECURITY_IDENTIFICATION)
        .open(name)?;
    Ok(std::io::PipeWriter::from(OwnedHandle::from(pipe)))
}

/// The launcher's end of a readiness channel: a named pipe the daemon opens
/// by name, so the daemon inherits no handle for it and no other child can.
#[cfg(feature = "launch")]
pub(crate) mod ready {
    use super::*;
    use std::os::windows::io::{FromRawHandle, OwnedHandle};
    use std::sync::mpsc::Receiver;

    #[repr(C)]
    struct SecurityAttributes {
        length: u32,
        descriptor: *mut c_void,
        inherit: i32,
    }

    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn CreateNamedPipeW(
            name: *const u16,
            open_mode: u32,
            pipe_mode: u32,
            max_instances: u32,
            out_buffer_size: u32,
            in_buffer_size: u32,
            default_timeout: u32,
            security: *const SecurityAttributes,
        ) -> Handle;
        fn ConnectNamedPipe(pipe: Handle, overlapped: *mut Overlapped) -> i32;
        fn PeekNamedPipe(
            pipe: Handle,
            buffer: *mut c_void,
            buffer_size: u32,
            read: *mut u32,
            available: *mut u32,
            left_in_message: *mut u32,
        ) -> i32;
        fn WaitForMultipleObjects(
            count: u32,
            handles: *const Handle,
            wait_all: i32,
            milliseconds: u32,
        ) -> u32;
    }

    #[link(name = "advapi32")]
    unsafe extern "system" {
        fn ConvertStringSecurityDescriptorToSecurityDescriptorW(
            text: *const u16,
            revision: u32,
            descriptor: *mut *mut c_void,
            size: *mut u32,
        ) -> i32;
        #[link_name = "SystemFunction036"]
        fn RtlGenRandom(buffer: *mut c_void, length: u32) -> u8;
    }

    const PIPE_ACCESS_INBOUND: u32 = 0x0000_0001;
    const FILE_FLAG_FIRST_PIPE_INSTANCE: u32 = 0x0008_0000;
    const FILE_FLAG_OVERLAPPED: u32 = 0x4000_0000;
    /// The pipe mode's only flag: byte type, byte reads and blocking are all
    /// zero.
    const PIPE_REJECT_REMOTE_CLIENTS: u32 = 0x0000_0008;
    const SDDL_REVISION_1: u32 = 1;
    const INFINITE: u32 = 0xFFFF_FFFF;
    const ERROR_BROKEN_PIPE: i32 = 109;
    const ERROR_NO_DATA: i32 = 232;
    const ERROR_PIPE_CONNECTED: i32 = 535;
    const ERROR_OPERATION_ABORTED: i32 = 995;

    /// A pipe name with 128 bits from the OS random source.
    fn random_name() -> io::Result<String> {
        let mut bytes = [0u8; 16];
        // SAFETY: the buffer is 16 writable bytes, the length passed.
        if unsafe { RtlGenRandom(bytes.as_mut_ptr().cast(), 16) } == 0 {
            return Err(io::Error::other("the OS random source failed"));
        }
        let id: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
        Ok(format!("{READY_PIPE_PREFIX}{id}"))
    }

    /// A security descriptor whose protected DACL grants the current user,
    /// and nobody else, access.
    fn current_user_only() -> io::Result<LocalMemory> {
        let sddl: Vec<u16> = format!("D:P(A;;GA;;;{})", current_user_sid()?)
            .encode_utf16()
            .chain(std::iter::once(0))
            .collect();
        let mut descriptor = ptr::null_mut();
        // SAFETY: `sddl` is NUL-terminated and `descriptor` is a valid output
        // slot; the size output is optional.
        let converted = unsafe {
            ConvertStringSecurityDescriptorToSecurityDescriptorW(
                sddl.as_ptr(),
                SDDL_REVISION_1,
                &mut descriptor,
                ptr::null_mut(),
            )
        };
        if converted == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(LocalMemory(descriptor))
    }

    /// Create a readiness pipe: its name for the daemon, and the launcher's
    /// end.
    ///
    /// One inbound instance with a random name. `FILE_FLAG_FIRST_PIPE_INSTANCE`
    /// fails the creation instead of joining a pipe someone else made first,
    /// the DACL admits only the current user, remote clients are refused, and
    /// the handle is not inheritable.
    pub(crate) fn create() -> io::Result<(String, OwnedHandle)> {
        let name = random_name()?;
        let wide: Vec<u16> = name.encode_utf16().chain(std::iter::once(0)).collect();
        let descriptor = current_user_only()?;
        let security = SecurityAttributes {
            length: std::mem::size_of::<SecurityAttributes>() as u32,
            descriptor: descriptor.0,
            inherit: 0,
        };
        // SAFETY: `wide` is NUL-terminated and `security`, with the
        // descriptor it points to, lives for the call.
        let pipe = unsafe {
            CreateNamedPipeW(
                wide.as_ptr(),
                PIPE_ACCESS_INBOUND | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_OVERLAPPED,
                PIPE_REJECT_REMOTE_CLIENTS,
                1,
                0,
                4096,
                0,
                &security,
            )
        };
        if pipe == INVALID_HANDLE_VALUE {
            return Err(io::Error::last_os_error());
        }
        // SAFETY: CreateNamedPipeW succeeded, so this is an open handle we own.
        Ok((name, unsafe { OwnedHandle::from_raw_handle(pipe) }))
    }

    /// What one step of reading found.
    enum Next {
        Got(usize),
        /// Nothing more can arrive.
        End,
        /// Try again: connected, or a read found nothing yet.
        Again,
    }

    /// Reads a readiness pipe while watching the daemon's process, so the
    /// daemon's exit ends the channel whether or not it ever connected.
    ///
    /// The process handle arrives once the daemon has started; the channel
    /// is read from before that. When the spawn fails the sender is dropped
    /// and the channel ends.
    pub(crate) struct Reader {
        pipe: OwnedHandle,
        started: Receiver<OwnedHandle>,
        process: Option<OwnedHandle>,
        connected: bool,
        /// The daemon exited: read what it left in the pipe, never wait.
        exited: bool,
        ended: bool,
    }

    impl Reader {
        pub(crate) fn new(pipe: OwnedHandle, started: Receiver<OwnedHandle>) -> Self {
            Self {
                pipe,
                started,
                process: None,
                connected: false,
                exited: false,
                ended: false,
            }
        }

        fn process(&mut self) -> Option<Handle> {
            if self.process.is_none() {
                self.process = self.started.recv().ok();
            }
            self.process.as_ref().map(|process| process.as_raw_handle())
        }

        fn step(&mut self, buf: &mut [u8]) -> io::Result<Next> {
            let Some(process) = self.process() else {
                return Ok(Next::End);
            };
            if !self.connected {
                return self.connect(process);
            }
            if self.exited {
                return self.drain(buf);
            }
            self.receive(buf, Some(process))
        }

        fn connect(&mut self, process: Handle) -> io::Result<Next> {
            let pipe = self.pipe.as_raw_handle();
            let event = new_event()?;
            let mut operation = operation(&event);
            // SAFETY: `operation` and its event outlive the connect: it fails
            // without starting, or `settle` returns only once it is terminal.
            let connected = unsafe { ConnectNamedPipe(pipe, &mut operation) };
            if connected == 0 {
                let error = io::Error::last_os_error();
                match error.raw_os_error() {
                    // The daemon connected, and perhaps already closed its end,
                    // before this call; what it wrote is still in the pipe.
                    Some(ERROR_PIPE_CONNECTED | ERROR_NO_DATA) => {
                        self.connected = true;
                        return Ok(Next::Again);
                    }
                    Some(ERROR_IO_PENDING) => {}
                    _ => return Err(error),
                }
            }
            let (result, exited) = self.settle(&mut operation, Some(process))?;
            self.exited = exited;
            match result {
                Ok(_) => {
                    self.connected = true;
                    Ok(Next::Again)
                }
                // Exited without connecting: nothing can arrive.
                Err(_) if exited => Ok(Next::End),
                Err(error) => Err(error),
            }
        }

        /// The daemon has exited: read only what it left behind.
        fn drain(&mut self, buf: &mut [u8]) -> io::Result<Next> {
            let mut available = 0u32;
            // SAFETY: a null buffer of size zero peeks no data; the call only
            // writes `available`.
            let peeked = unsafe {
                PeekNamedPipe(
                    self.pipe.as_raw_handle(),
                    ptr::null_mut(),
                    0,
                    ptr::null_mut(),
                    &mut available,
                    ptr::null_mut(),
                )
            };
            if peeked == 0 || available == 0 {
                return Ok(Next::End);
            }
            let length = buf.len().min(available as usize);
            self.receive(&mut buf[..length], None)
        }

        /// Read into `buf`, and when `process` is given, stop if the daemon
        /// exits first.
        fn receive(&mut self, buf: &mut [u8], process: Option<Handle>) -> io::Result<Next> {
            let pipe = self.pipe.as_raw_handle();
            let event = new_event()?;
            let mut operation = operation(&event);
            let length = u32::try_from(buf.len()).unwrap_or(u32::MAX);
            // SAFETY: `buf`, `operation` and its event outlive the read: it
            // fails without starting, or `settle` returns only once it is
            // terminal.
            let started = unsafe {
                ReadFile(
                    pipe,
                    buf.as_mut_ptr().cast(),
                    length,
                    ptr::null_mut(),
                    &mut operation,
                )
            };
            if started == 0 {
                let error = io::Error::last_os_error();
                match error.raw_os_error() {
                    Some(ERROR_IO_PENDING) => {}
                    Some(ERROR_BROKEN_PIPE) => return Ok(Next::End),
                    _ => return Err(error),
                }
            }
            let (result, exited) = self.settle(&mut operation, process)?;
            self.exited |= exited;
            match result {
                Ok(0) => Ok(Next::Again),
                Ok(count) => Ok(Next::Got(count as usize)),
                Err(error) => match error.raw_os_error() {
                    Some(ERROR_BROKEN_PIPE) => Ok(Next::End),
                    // Cancelled because the daemon exited: drain next.
                    Some(ERROR_OPERATION_ABORTED) => Ok(Next::Again),
                    _ => Err(error),
                },
            }
        }

        /// Wait for `operation`, or for the daemon's exit when `process` is
        /// given, and return the operation's own result and whether the daemon
        /// exited first. Returns only once the operation is terminal, so the
        /// kernel never writes into it or its buffer afterwards.
        fn settle(
            &self,
            operation: &mut Overlapped,
            process: Option<Handle>,
        ) -> io::Result<(io::Result<u32>, bool)> {
            let pipe = self.pipe.as_raw_handle();
            // The operation's event comes first: when both are signalled the
            // wait reports it, so data sent just before an exit is read.
            let handles = [operation.event, process.unwrap_or(operation.event)];
            let count = if process.is_some() { 2 } else { 1 };
            // SAFETY: both handles are open for the whole wait.
            let woke = unsafe { WaitForMultipleObjects(count, handles.as_ptr(), 0, INFINITE) };
            let exited = process.is_some() && woke == WAIT_OBJECT_0 + 1;
            let wait_error = (woke != WAIT_OBJECT_0 && !exited).then(io::Error::last_os_error);
            if exited || wait_error.is_some() {
                // SAFETY: cancels only this operation, which is still live.
                unsafe {
                    CancelIoEx(pipe, operation);
                }
            }
            let mut transferred = 0u32;
            // SAFETY: `wait = 1` returns only once the operation is terminal.
            let completed = unsafe { GetOverlappedResult(pipe, operation, &mut transferred, 1) };
            let result = if completed != 0 {
                Ok(transferred)
            } else {
                Err(io::Error::last_os_error())
            };
            match wait_error {
                Some(error) => Err(error),
                None => Ok((result, exited)),
            }
        }
    }

    impl Read for Reader {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            while !self.ended && !buf.is_empty() {
                match self.step(buf)? {
                    Next::Got(count) => return Ok(count),
                    Next::End => self.ended = true,
                    Next::Again => {}
                }
            }
            Ok(0)
        }
    }

    fn new_event() -> io::Result<OwnedHandle> {
        // SAFETY: a null security descriptor and a null name create an
        // unnamed manual-reset event, initially not signalled.
        let event = unsafe { CreateEventW(ptr::null(), 1, 0, ptr::null()) };
        if event.is_null() {
            return Err(io::Error::last_os_error());
        }
        // SAFETY: CreateEventW succeeded, so this is an open handle we own.
        Ok(unsafe { OwnedHandle::from_raw_handle(event) })
    }

    fn operation(event: &OwnedHandle) -> Overlapped {
        // SAFETY: an all-zero OVERLAPPED is a valid idle operation.
        let mut operation: Overlapped = unsafe { std::mem::zeroed() };
        operation.event = event.as_raw_handle();
        operation
    }
}

#[cfg(test)]
mod ready_pipe_tests {
    use super::*;

    #[test]
    fn only_a_readiness_pipe_name_is_opened() {
        let id = "0123456789abcdef0123456789abcdef";
        assert!(is_ready_pipe_name(&format!("{READY_PIPE_PREFIX}{id}")));
        for name in [
            String::new(),
            id.to_owned(),
            format!("{READY_PIPE_PREFIX}{}", &id[1..]),
            format!("{READY_PIPE_PREFIX}{id}0"),
            format!("{READY_PIPE_PREFIX}{}", id.to_uppercase()),
            format!(r"\\.\pipe\other-{id}"),
            r"C:\Windows\win.ini".to_owned(),
        ] {
            assert!(!is_ready_pipe_name(&name), "{name}");
            let error = open_ready_pipe(&name).unwrap_err();
            assert_eq!(error.kind(), io::ErrorKind::InvalidInput, "{name}");
        }
    }

    /// A process for the reader to watch, and its handle.
    #[cfg(feature = "launch")]
    fn watched(
        program: &str,
        args: &[&str],
    ) -> (std::process::Child, std::os::windows::io::OwnedHandle) {
        let child = std::process::Command::new(program)
            .args(args)
            .stdout(std::process::Stdio::null())
            .spawn()
            .unwrap();
        let handle = std::os::windows::io::AsHandle::as_handle(&child)
            .try_clone_to_owned()
            .unwrap();
        (child, handle)
    }

    #[cfg(feature = "launch")]
    #[test]
    fn a_readiness_pipe_carries_messages_until_its_writer_closes() {
        let (name, pipe) = ready::create().unwrap();
        assert!(is_ready_pipe_name(&name), "{name}");
        let (mut child, process) = watched("ping", &["-n", "30", "127.0.0.1"]);
        let (started, delivered) = std::sync::mpsc::channel();
        started.send(process).unwrap();
        let mut reader = ready::Reader::new(pipe, delivered);
        let mut daemon = open_ready_pipe(&name).unwrap();
        daemon.write_all(b"progress\nready\n").unwrap();
        drop(daemon);
        let mut received = Vec::new();
        reader.read_to_end(&mut received).unwrap();
        child.kill().unwrap();
        child.wait().unwrap();
        assert_eq!(received, b"progress\nready\n");
    }

    #[cfg(feature = "launch")]
    #[test]
    fn a_daemon_that_exits_without_connecting_ends_its_channel() {
        // Nobody ever opens the pipe, so only the process's exit can end the
        // read; without watching it this would wait forever.
        let (_name, pipe) = ready::create().unwrap();
        let (mut child, process) = watched("cmd", &["/c", "exit", "0"]);
        let (started, delivered) = std::sync::mpsc::channel();
        started.send(process).unwrap();
        let mut received = Vec::new();
        ready::Reader::new(pipe, delivered)
            .read_to_end(&mut received)
            .unwrap();
        child.wait().unwrap();
        assert!(received.is_empty());
    }

    #[cfg(feature = "launch")]
    #[test]
    fn a_failed_spawn_ends_the_channel() {
        let (_name, pipe) = ready::create().unwrap();
        let (started, delivered) = std::sync::mpsc::channel::<std::os::windows::io::OwnedHandle>();
        drop(started);
        let mut received = Vec::new();
        ready::Reader::new(pipe, delivered)
            .read_to_end(&mut received)
            .unwrap();
        assert!(received.is_empty());
    }
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc;

    use interprocess::local_socket::traits::Listener as _;
    use interprocess::local_socket::{GenericNamespaced, ListenerOptions, ToNsName};

    use super::*;
    use crate::local::test_handshake as handshake;

    #[test]
    fn a_named_pipe_reports_that_it_cannot_half_close_and_keeps_its_replies() {
        // shutdown_write used to flush and return Ok, so a relay believed its
        // peer knew the requests had ended, and both ends waited for each other.
        let endpoint = format!("daemon-transport-no-half-close-{}", std::process::id());
        let listener = ListenerOptions::new()
            .name(endpoint.as_str().to_ns_name::<GenericNamespaced>().unwrap())
            .create_sync()
            .unwrap();
        let server = std::thread::spawn(move || {
            let mut peer = listener.accept().unwrap();
            let mut request = [0; 8];
            peer.read_exact(&mut request).unwrap();
            peer.write_all(b"reply\n").unwrap();
            request
        });
        let transport = WindowsDuplex::connect(&endpoint).unwrap();
        let (mut read, mut write) = transport.split().unwrap();
        write.write_all(b"request\n").unwrap();
        let error = write.shutdown_write().unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::Unsupported);
        assert_eq!(
            &server.join().unwrap(),
            b"request\n",
            "the flush still delivers the request"
        );
        let mut reply = [0; 6];
        read.read_exact(&mut reply).unwrap();
        assert_eq!(&reply, b"reply\n");
    }

    #[test]
    fn setup_write_deadline_bounds_a_nonreading_named_pipe() {
        let endpoint = format!("daemon-transport-write-deadline-{}", std::process::id());
        let listener = ListenerOptions::new()
            .name(endpoint.as_str().to_ns_name::<GenericNamespaced>().unwrap())
            .create_sync()
            .unwrap();
        let (release, wait) = mpsc::channel();
        let server = std::thread::spawn(move || {
            let _peer = listener.accept().unwrap();
            let _ = wait.recv_timeout(Duration::from_secs(2));
        });
        let transport = WindowsDuplex::connect(&endpoint).unwrap();
        assert!(
            crate::local::peer::PeerCredentials::credentials(&transport)
                .unwrap()
                .same_user
        );
        transport
            .set_read_deadline(Some(Duration::from_millis(100)))
            .unwrap();
        let (_read, mut write) = transport.split().unwrap();
        let error = write.write_all(&vec![0; 16 << 20]).unwrap_err();
        release.send(()).ok();
        server.join().unwrap();
        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
    }

    #[test]
    fn a_silent_named_pipe_times_out_the_handshake_read() {
        let endpoint = format!("daemon-transport-deadline-silent-{}", std::process::id());
        let name = endpoint.as_str().to_ns_name::<GenericNamespaced>().unwrap();
        let listener = ListenerOptions::new().name(name).create_sync().unwrap();
        let (release_tx, release_rx) = mpsc::channel();
        let server = std::thread::spawn(move || {
            let connection = listener.accept().unwrap();
            release_rx.recv().unwrap();
            drop(connection);
        });

        let transport = WindowsDuplex::connect(&endpoint).unwrap();
        transport
            .set_read_deadline(Some(Duration::from_millis(200)))
            .unwrap();
        let (mut read, mut write) = transport.split().unwrap();
        let started = Instant::now();
        let error = handshake(&mut write, &mut read, "windows-test").unwrap_err();
        assert!(
            error.kind() == io::ErrorKind::TimedOut,
            "silent pipe returned {error:?}"
        );
        assert!(
            started.elapsed() >= Duration::from_millis(150)
                && started.elapsed() < Duration::from_secs(2),
            "handshake deadline was not enforced: {:?}",
            started.elapsed()
        );

        release_tx.send(()).unwrap();
        server.join().unwrap();
    }

    #[test]
    fn clearing_the_deadline_restores_an_unbounded_steady_state_read() {
        let endpoint = format!("daemon-transport-deadline-clear-{}", std::process::id());
        let name = endpoint.as_str().to_ns_name::<GenericNamespaced>().unwrap();
        let listener = ListenerOptions::new().name(name).create_sync().unwrap();
        let (send_tx, send_rx) = mpsc::channel();
        let server = std::thread::spawn(move || {
            let mut connection = listener.accept().unwrap();
            let mut preamble = [0u8; 6];
            connection.read_exact(&mut preamble).unwrap();
            connection.write_all(b"ready\n").unwrap();
            connection.flush().unwrap();
            send_rx.recv().unwrap();
            #[expect(
                clippy::disallowed_methods,
                reason = "The test must receive data after the cleared pipe deadline has passed."
            )]
            std::thread::sleep(Duration::from_millis(650));
            connection.write_all(b"x").unwrap();
        });

        let transport = WindowsDuplex::connect(&endpoint).unwrap();
        transport
            .set_read_deadline(Some(Duration::from_millis(500)))
            .unwrap();
        let (mut read, mut write) = transport.split().unwrap();
        handshake(&mut write, &mut read, "windows-test").unwrap();
        read.clear_read_deadline();
        send_tx.send(()).unwrap();

        // The byte arrives after the old deadline. Receiving it proves the
        // steady-state path reverted to the ordinary blocking reader.
        let mut byte = [0u8; 1];
        read.read_exact(&mut byte).unwrap();
        assert_eq!(byte, *b"x");
        server.join().unwrap();
    }
}

/// Clients for tests of a server end's evidence, including one that writes as
/// another user: the anonymous logon, which any thread can impersonate.
#[cfg(test)]
pub(crate) mod test_clients {
    use super::*;

    #[link(name = "advapi32")]
    unsafe extern "system" {
        fn ImpersonateAnonymousToken(thread: Handle) -> i32;
    }

    /// Open the pipe `name` with dynamic context tracking at identification
    /// level: the server sees the user of whichever thread wrote last.
    pub(crate) fn connect_tracking(name: &str) -> io::Result<std::fs::File> {
        use std::os::windows::fs::OpenOptionsExt;
        const SECURITY_IDENTIFICATION: u32 = 1 << 16;
        const SECURITY_CONTEXT_TRACKING: u32 = 0x0004_0000;
        std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .security_qos_flags(SECURITY_IDENTIFICATION | SECURITY_CONTEXT_TRACKING)
            .open(format!(r"\\.\pipe\{name}"))
    }

    /// Write `bytes` while this thread impersonates the anonymous logon, so a
    /// tracking pipe records another user for this process's connection.
    pub(crate) fn write_as_anonymous(pipe: &mut impl Write, bytes: &[u8]) -> io::Result<()> {
        // SAFETY: GetCurrentThread returns a pseudo-handle for this thread.
        if unsafe { ImpersonateAnonymousToken(GetCurrentThread()) } == 0 {
            return Err(io::Error::last_os_error());
        }
        let written = pipe.write_all(bytes);
        // SAFETY: ends the impersonation this function started on this thread.
        let reverted = unsafe { RevertToSelf() };
        assert_ne!(reverted, 0, "the test client must stop impersonating");
        written
    }
}

#[cfg(test)]
mod server_end_evidence_tests {
    use super::test_clients::{connect_tracking, write_as_anonymous};
    use super::*;
    use crate::{
        ProcessId,
        local::peer::PeerCredentials,
        peer::{Credentials, UserEvidence},
    };
    use interprocess::local_socket::traits::Listener as _;
    use interprocess::local_socket::{ListenerOptions, Stream as Client};
    use std::sync::mpsc;

    const BUDGET: Duration = Duration::from_secs(20);

    /// Accept one connection on a fresh pipe, read the client's first byte
    /// when `read_first`, and run `check` on the server end, on the thread
    /// that accepted it. `client` runs on its own thread with the pipe's name
    /// and should wait for the server to close before it exits.
    fn on_server_end<T: Send + 'static>(
        tag: &str,
        read_first: bool,
        client: impl FnOnce(String) + Send + 'static,
        check: impl FnOnce(&Client) -> T + Send + 'static,
    ) -> T {
        let name = format!("daemon-evidence-{tag}-{}", std::process::id());
        let listener = ListenerOptions::new()
            .name(name.as_str().to_ns_name::<GenericNamespaced>().unwrap())
            .create_sync()
            .unwrap();
        let (sent, received) = mpsc::channel();
        std::thread::spawn(move || {
            let mut server = listener.accept().unwrap();
            if read_first {
                server.read_exact(&mut [0u8; 1]).unwrap();
            }
            let _ = sent.send(check(&server));
        });
        let client = std::thread::spawn(move || client(name));
        let checked = received
            .recv_timeout(BUDGET)
            .expect("the server end was not checked within the budget");
        client.join().unwrap();
        checked
    }

    fn server_evidence(
        tag: &str,
        read_first: bool,
        client: impl FnOnce(String) + Send + 'static,
    ) -> io::Result<Credentials> {
        on_server_end(tag, read_first, client, |server| server.credentials())
    }

    fn writes_as_anonymous(name: String) {
        let mut pipe = connect_tracking(&name).unwrap();
        write_as_anonymous(&mut pipe, b"x").unwrap();
        let _ = pipe.read(&mut [0u8; 1]);
    }

    fn pipe_handle(server: &Client) -> Handle {
        let Client::NamedPipe(pipe) = server;
        pipe.as_handle().as_raw_handle()
    }

    fn thread_has_token() -> bool {
        thread_token(TOKEN_QUERY).unwrap().is_some()
    }

    fn this_process() -> Option<ProcessId> {
        ProcessId::new(std::process::id())
    }

    #[test]
    fn a_server_end_reads_its_clients_user_from_the_connection() {
        // The client is this process, so a lookup by PID would find this
        // process's user. The connection carries the user that wrote.
        let evidence = server_evidence("anonymous", true, writes_as_anonymous);
        let evidence = evidence.unwrap();
        assert_eq!(
            (evidence.pid, evidence.same_user, evidence.user_evidence),
            (this_process(), false, UserEvidence::Connection)
        );
    }

    #[test]
    fn a_panic_while_impersonating_still_restores_the_thread() {
        let outcome = on_server_end("panic", true, writes_as_anonymous, |server| {
            let pipe = pipe_handle(server);
            let mut impersonating = false;
            let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let _client = Impersonating::pipe_client(pipe).unwrap();
                impersonating = thread_has_token();
                panic!("inside the impersonation");
            }));
            (impersonating, panicked.is_err(), thread_has_token())
        });
        assert_eq!(outcome, (true, true, false));
    }

    /// The `TokenId` of `token`: unique to that token, unlike its user.
    fn token_id(token: &Event) -> u64 {
        // TOKEN_STATISTICS is 56 bytes and starts with the TokenId LUID.
        let mut statistics = [0u64; 8];
        let mut needed = 0;
        // SAFETY: `token` is a live token handle opened with TOKEN_QUERY,
        // TokenStatistics is class 10, and the aligned buffer is larger than
        // TOKEN_STATISTICS.
        let read = unsafe {
            GetTokenInformation(
                token.0,
                10,
                statistics.as_mut_ptr().cast(),
                std::mem::size_of_val(&statistics) as u32,
                &mut needed,
            )
        };
        assert_ne!(read, 0, "{}", io::Error::last_os_error());
        statistics[0]
    }

    #[test]
    fn an_impersonating_thread_gets_its_own_token_back() {
        #[link(name = "advapi32")]
        unsafe extern "system" {
            fn ImpersonateSelf(level: i32) -> i32;
        }
        const SECURITY_IMPERSONATION: i32 = 2;
        // The thread already impersonates a copy of this process's token.
        // Reading the client must leave that copy in place, not drop it.
        let outcome = on_server_end("previous", true, writes_as_anonymous, |server| {
            // SAFETY: changes only this thread's token; reverted below.
            assert_ne!(unsafe { ImpersonateSelf(SECURITY_IMPERSONATION) }, 0);
            let before = thread_token(TOKEN_QUERY)
                .unwrap()
                .map(|token| token_id(&token));
            let same_user = server.credentials().map(|evidence| evidence.same_user);
            let token = thread_token(TOKEN_QUERY).unwrap();
            // The very token it had, not a fresh token for the same user.
            let kept = before.is_some() && token.as_ref().map(token_id) == before;
            let user = token.map(|token| query_token_user(&token).unwrap());
            // SAFETY: ends the impersonation ImpersonateSelf started.
            assert_ne!(unsafe { RevertToSelf() }, 0);
            // SAFETY: GetCurrentProcess returns a borrowed pseudo-handle.
            let own = token_user(unsafe { GetCurrentProcess() }).unwrap();
            (same_user.ok(), user.map(|user| same_sid(&user, &own)), kept)
        });
        assert_eq!(outcome, (Some(false), Some(true), true));
    }

    #[test]
    fn a_same_user_client_that_wrote_is_the_same_user() {
        let evidence = server_evidence("same-user", true, |name| {
            let mut pipe =
                Client::connect(name.as_str().to_ns_name::<GenericNamespaced>().unwrap()).unwrap();
            pipe.write_all(b"x").unwrap();
            let _ = pipe.read(&mut [0u8; 1]);
        });
        let evidence = evidence.unwrap();
        assert_eq!(
            (evidence.pid, evidence.same_user, evidence.user_evidence),
            (this_process(), true, UserEvidence::Connection)
        );
    }

    #[test]
    fn a_client_that_allows_only_anonymous_impersonation_falls_back_to_its_pid() {
        let evidence = server_evidence("anonymous-level", true, |name| {
            use std::os::windows::fs::OpenOptionsExt;
            const SECURITY_ANONYMOUS: u32 = 0;
            let mut pipe = std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .security_qos_flags(SECURITY_ANONYMOUS)
                .open(format!(r"\\.\pipe\{name}"))
                .unwrap();
            pipe.write_all(b"x").unwrap();
            let _ = pipe.read(&mut [0u8; 1]);
        });
        let evidence = evidence.unwrap();
        assert_eq!(
            (evidence.pid, evidence.same_user, evidence.user_evidence),
            (this_process(), true, UserEvidence::PidLookup)
        );
    }

    #[test]
    fn a_server_end_whose_client_has_not_written_falls_back_to_its_pid() {
        // The connection carries no user yet, so the evidence is the same as
        // a lookup by PID gives, and says so.
        let evidence = server_evidence("silent", false, |name| {
            let mut pipe = connect_tracking(&name).unwrap();
            let _ = pipe.read(&mut [0u8; 1]);
        });
        let evidence = evidence.unwrap();
        assert_eq!(
            (evidence.pid, evidence.same_user, evidence.user_evidence),
            (this_process(), true, UserEvidence::PidLookup)
        );
    }
}