syd 3.52.0

//
// Syd: rock-solid application kernel
// src/workers/ipc.rs: `syd_ipc' remote-configuration thread
//
// Copyright (c) 2025, 2026 Ali Polatel <alip@chesswob.org>
//
// SPDX-License-Identifier: GPL-3.0

use std::{
    borrow::Cow,
    hash::{Hash, Hasher},
    io::{IoSlice, IoSliceMut},
    os::{
        fd::{AsFd, AsRawFd, FromRawFd, RawFd},
        unix::net::{UnixListener, UnixStream},
    },
    sync::{atomic::Ordering, Arc, RwLock},
    thread,
};

use libseccomp::{ScmpAction, ScmpFilterContext, ScmpSyscall};
use memchr::{arch::all::is_equal, memchr};
use nix::{
    errno::Errno,
    fcntl::OFlag,
    poll::PollTimeout,
    sched::{unshare, CloneFlags},
    sys::{
        epoll::{Epoll, EpollCreateFlags, EpollEvent, EpollFlags},
        socket::{
            accept4, bind, listen, setsockopt, socket, sockopt::PassCred, Backlog, SockFlag,
            SockaddrStorage, UnixAddr, UnixCredentials,
        },
        stat::{umask, Mode},
    },
    unistd::{write, Gid, Uid},
};
use serde::{ser::SerializeMap, Serialize, Serializer};

use crate::{
    alert,
    compat::{
        epoll_ctl_mod_safe, epoll_ctl_safe, recvmsg, sendmsg, CmsgOwned, CmsgSpace, MsgFlags,
        MsgHdr, SockType,
    },
    config::*,
    confine::{
        confine_scmp_epoll, confine_scmp_fcntl, confine_scmp_madvise, confine_scmp_open_stat,
        confine_scmp_prctl, confine_scmp_setid, confine_scmp_write, confine_scmp_wx_syd,
    },
    err::{err2no, scmp2no, SydJoinHandle, SydResult},
    fd::{closeexcept, peer_creds, SafeOwnedFd},
    hash::SydHashMap,
    info,
    landlock::Errata,
    landlock_policy::LandlockPolicy,
    path::{XPath, XPathBuf},
    retry::retry_on_eintr,
    rng::duprand,
    sandbox::{Options, Sandbox, LINE_MAX},
};

const IPC_ACK: &[u8] = b"{\"err\":0,\"msg\":\"ACK\"}\n";
const IPC_AUTH: &[u8] = b"{\"err\":13,\"msg\":\"AUTH\"}\n";
const IPC_RATE: &[u8] = b"{\"err\":7,\"msg\":\"RATE\"}\n";
const IPC_PINK: &[u8] = b"{\"err\":0,\
\"msg\":\"Change return success. Going and coming without error. Action brings good fortune.\"}\n";
const IPC_PONG: &[u8] = b"{\"err\":0,\"msg\":\"PONG\"}\n";

// Epoll-based, single-threaded IPC server over a UNIX socket.
pub(crate) struct IpcWorker {
    // Path to the UNIX socket (possibly abstract if it starts with '@')
    //
    // This is set to `None` once the bind is successful.
    addr: Option<XPathBuf>,
    // Epoll file descriptor
    pub(crate) epoll: Option<Epoll>,
    // Listener socket
    pub(crate) sock: Option<UnixListener>,
    // Reference to the Sandbox to be configured
    sandbox: Option<Arc<RwLock<Sandbox>>>,
    // Sandbox options specified at startup.
    options: Options,
    // Credentials for authentication.
    creds: (Option<Uid>, Option<Gid>),

    // SafeSetId UID/GID transitions.
    transit_uids: Vec<(Uid, Uid)>,
    transit_gids: Vec<(Gid, Gid)>,
}

// Prompt modes supported (similar to HAProxy)
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
enum PromptMode {
    // Non-interactive: single command, then close.
    NonInteractive,
    // Interactive: accept multiple commands, no prompt.
    Interactive,
    // Prompt: accept multiple commands and send prompt "; " before each.
    Prompt,
}

impl Serialize for PromptMode {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let s = match *self {
            Self::NonInteractive => "non-interactive",
            Self::Interactive => "interactive",
            Self::Prompt => "prompt",
        };
        serializer.serialize_str(s)
    }
}

// Per-connection state: buffering incoming bytes,
// pending outgoing bytes, and tracking mode.
struct Connection {
    // Credentials of the other end.
    creds: UnixCredentials,
    // Underlying UNIX-stream socket.
    stream: UnixStream,
    // Buffered incoming bytes until a full line arrives.
    buf: Vec<u8>,
    // Buffer of bytes to write (response + prompt).
    write_buf: Vec<u8>,
    // Current position in `write_buf`.
    write_pos: usize,
    // Prompt mode for this connection.
    mode: PromptMode,
    // Whether to close after flushing the write buffer.
    should_close: bool,
}

// Type for Connection map.
type ConnectionMap = SydHashMap<RawFd, Connection>;

impl Hash for Connection {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.stream.as_fd().as_raw_fd().hash(state);
    }
}

impl PartialEq for Connection {
    fn eq(&self, other: &Self) -> bool {
        self.stream.as_fd().as_raw_fd() == other.stream.as_fd().as_raw_fd()
    }
}

impl Eq for Connection {}

// Wrapper type over UnixCredentials that implements Serialize.
struct Creds(UnixCredentials);

impl Serialize for Creds {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut m = serializer.serialize_map(Some(3))?;
        m.serialize_entry("uid", &self.0.uid())?;
        m.serialize_entry("gid", &self.0.gid())?;
        m.serialize_entry("pid", &self.0.pid())?;
        m.end()
    }
}

impl Serialize for Connection {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut map = serializer.serialize_map(Some(8))?;

        map.serialize_entry("fd", &self.stream.as_raw_fd())?;
        map.serialize_entry("ilen", &self.buf.len())?;
        map.serialize_entry("icap", &self.buf.capacity())?;
        map.serialize_entry("olen", &self.write_buf.len())?;
        map.serialize_entry("ocap", &self.write_buf.capacity())?;
        map.serialize_entry("mode", &self.mode)?;
        map.serialize_entry("close", &self.should_close)?;
        map.serialize_entry("creds", &Creds(self.creds))?;

        map.end()
    }
}

impl Connection {
    fn new(creds: UnixCredentials, stream: UnixStream) -> Self {
        Self {
            creds,
            stream,
            buf: Vec::with_capacity(1024),
            write_buf: Vec::with_capacity(1024),
            write_pos: 0,
            mode: PromptMode::NonInteractive,
            should_close: false,
        }
    }

    // Change epoll(7) interests.
    fn ctl(&self, epoll: &Epoll, flags: EpollFlags) -> Result<(), Errno> {
        let fd = self.stream.as_raw_fd();

        #[expect(clippy::cast_sign_loss)]
        let event = libc::epoll_event {
            events: flags.bits() as u32,
            u64: fd as u64,
        };

        epoll_ctl_mod_safe(&epoll.0, fd, event)
    }

    // Append data to the write buffer.
    fn enqueue_response(&mut self, data: &[u8]) {
        self.write_buf.extend_from_slice(data);
    }

    // Enqueue the prompt string ("; ") in `write_buf`.
    fn enqueue_prompt(&mut self) {
        self.write_buf.extend_from_slice(b"; ");
    }
}

impl IpcWorker {
    // Create a new IpcWorker for the given path.
    //
    // If `addr` starts with '@', an abstract UNIX socket is used.
    //
    // This does not bind or listen yet; only stores the path.
    // Returns immediately without error.
    pub(crate) fn new(
        addr: &XPath,
        uid: Option<Uid>,
        gid: Option<Gid>,
        options: Options,
        transit_uids: &[(Uid, Uid)],
        transit_gids: &[(Gid, Gid)],
    ) -> Self {
        IpcWorker {
            options,
            creds: (uid, gid),
            addr: Some(addr.to_owned()),
            epoll: None,
            sock: None,
            sandbox: None,
            transit_uids: transit_uids.to_vec(),
            transit_gids: transit_gids.to_vec(),
        }
    }

    // Create the listener socket, bind it, make it non-blocking, listen, and create epoll.
    //
    // This function sets umask(2) and therefore may not be thread-safe!
    pub(crate) fn setup(&mut self) -> Result<(), Errno> {
        // Prepare UNIX socket address, `@' prefix implies abstract socket.
        let addr = self.addr.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        let addr = if matches!(addr.first(), Some(b'@')) {
            UnixAddr::new_abstract(&addr.as_bytes()[1..])?
        } else {
            UnixAddr::new(addr.as_bytes())?
        };
        self.addr = None; // No longer needed.

        // Create UNIX stream socket, set to non-blocking.
        let sock = socket(
            nix::sys::socket::AddressFamily::Unix,
            SockType::Stream.try_into()?,
            SockFlag::SOCK_CLOEXEC | SockFlag::SOCK_NONBLOCK,
            None,
        )?;

        // Randomize to make fd reuse harder.
        let sock_fd = duprand(sock.as_raw_fd(), OFlag::O_CLOEXEC).inspect(|fd| {
            // Used to prevent leaks in sandbox process:
            let mut fd_str = itoa::Buffer::new();
            let fd_str = fd_str.format(fd.as_raw_fd());
            std::env::set_var(ENV_IPC_UNIX_FD, fd_str);
        })?;
        drop(sock);
        let sock = sock_fd;

        // Set SO_PASSCRED for authentication.
        setsockopt(&sock, PassCred, &true)?;

        // Ensure socket is created with sane permissions.
        let umask_orig = umask(Mode::from_bits_truncate(0o077));
        let result = bind(sock.as_raw_fd(), &addr);
        umask(umask_orig);
        result?;

        listen(&sock, Backlog::MAXCONN)?;

        // Create epoll(7) instance.
        let epoll = Epoll::new(EpollCreateFlags::EPOLL_CLOEXEC)?;

        // Randomize the epoll fd to make fd reuse harder.
        let epoll_fd = duprand(epoll.0.as_raw_fd(), OFlag::O_CLOEXEC).inspect(|fd| {
            // Used to prevent leaks in sandbox process:
            let mut fd_str = itoa::Buffer::new();
            let fd_str = fd_str.format(fd.as_raw_fd());
            std::env::set_var(ENV_IPC_POLL_FD, fd_str);
        })?;
        drop(epoll);
        let epoll = Epoll(epoll_fd.into());

        // Register listener_fd for EPOLLIN.
        #[expect(clippy::cast_sign_loss)]
        let event = libc::epoll_event {
            events: EpollFlags::EPOLLIN.bits() as u32,
            u64: sock.as_fd().as_raw_fd() as u64,
        };
        epoll_ctl_safe(&epoll.0, sock.as_fd().as_raw_fd(), Some(event))?;

        self.epoll = Some(epoll);
        #[expect(clippy::disallowed_types)]
        {
            self.sock = Some(UnixListener::from(std::os::fd::OwnedFd::from(sock)));
        }

        Ok(())
    }

    // Set reference to the Sandbox to be configured.
    pub fn set_sandbox(&mut self, sandbox: Arc<RwLock<Sandbox>>) {
        self.sandbox = Some(sandbox);
    }

    // Confine IPC thread.
    pub(crate) fn prepare_confine(
        epoll_fd: RawFd,
        options: Options,
        transit_uids: &[(Uid, Uid)],
        transit_gids: &[(Gid, Gid)],
        dry_run: bool,
    ) -> SydResult<ScmpFilterContext> {
        if !dry_run {
            // Set up a landlock(7) sandbox to disallow all access.
            let abi = crate::landlock::ABI::new_current();
            let errata = crate::landlock::Errata::query();
            let policy = LandlockPolicy {
                scoped_abs: true,
                scoped_sig: errata.contains(Errata::SCOPED_SIGNAL_SAME_TGID),
                ..Default::default()
            };
            let _ = policy.restrict_self(abi);
        }

        // Create seccomp(2) filter with default action.
        let mut ctx = ScmpFilterContext::new(ScmpAction::KillProcess)?;

        // Enforce the NO_NEW_PRIVS functionality before
        // loading the seccomp filter into the kernel.
        ctx.set_ctl_nnp(true)?;

        // DO NOT synchronize filter to all threads.
        // Other threads will self-confine.
        ctx.set_ctl_tsync(false)?;

        // Disable Speculative Store Bypass mitigations
        // with trace/allow_unsafe_exec_speculative:1
        ctx.set_ctl_ssb(options.allow_unsafe_exec_speculative())?;

        // We kill for bad system call and bad arch.
        ctx.set_act_badarch(ScmpAction::KillProcess)?;

        // Use a binary tree sorted by syscall number if possible.
        let _ = ctx.set_ctl_optimize(2);

        // Do NOT add supported architectures to the filter.
        // This ensures Syd can never run a non-native system call,
        // which we do not need at all.
        // seccomp_add_architectures(&mut ctx)?;

        // Allow epoll(7) API to our single epoll fd only.
        confine_scmp_epoll(&mut ctx, epoll_fd)?;

        // Deny open and stat family with ENOSYS rather than KillProcess.
        confine_scmp_open_stat(&mut ctx, true /* openat2 */)?;

        // Allow safe fcntl(2) utility calls.
        confine_scmp_fcntl(&mut ctx, IPC_FCNTL_OPS)?;

        // Allow safe prctl(2) operations.
        confine_scmp_prctl(&mut ctx, IPC_PRCTL_OPS)?;

        // Prevent executable memory.
        confine_scmp_wx_syd(&mut ctx)?;

        // Allow writes to the log-fd.
        // No proc_pid_mem(5) access required here.
        confine_scmp_write(&mut ctx, None, false)?;

        // Allow safe madvise(2) advice.
        confine_scmp_madvise(&mut ctx)?;

        // Allow safe system calls.
        //
        // KCOV_SYSCALLS is empty in case `kcov` feature is disabled.
        for sysname in IPC_SYSCALLS
            .iter()
            .chain(ALLOC_SYSCALLS)
            .chain(FUTEX_SYSCALLS)
            .chain(GETID_SYSCALLS)
            .chain(KCOV_SYSCALLS)
            .chain(VDSO_SYSCALLS)
        {
            match ScmpSyscall::from_name(sysname) {
                Ok(syscall) => {
                    ctx.add_rule(ScmpAction::Allow, syscall)?;
                }
                Err(_) => {
                    info!("ctx": "confine", "op": "allow_ipc_syscall",
                        "msg": format!("invalid or unsupported syscall {sysname}"));
                }
            }
        }

        // Allow UID/GID changing system calls as necessary.
        let safe_setuid = options.allow_safe_setuid();
        let safe_setgid = options.allow_safe_setgid();
        if safe_setuid || safe_setgid {
            confine_scmp_setid(
                "ipc",
                &mut ctx,
                safe_setuid,
                safe_setgid,
                transit_uids,
                transit_gids,
            )?;
        }

        Ok(ctx)
    }

    // Spawn the IPC worker in a new thread named `syd_ipc'.
    pub fn try_spawn(mut self, notif_pipe: (RawFd, RawFd)) -> Result<SydJoinHandle<()>, Errno> {
        thread::Builder::new()
            .name("syd_ipc".into())
            .stack_size(IPC_STACK_SIZE)
            .spawn(move || {
                self.init(notif_pipe)?;
                self.main()
            })
            .map_err(|err| err2no(&err))
    }

    // Initialize IPC worker.
    fn init(&mut self, notif_pipe: (RawFd, RawFd)) -> SydResult<()> {
        // We use exit_group(2) here to bail, because this unsharing is
        // a critical safety feature.
        if let Err(errno) =
            unshare(CloneFlags::CLONE_FS | CloneFlags::CLONE_FILES | CloneFlags::CLONE_SYSVSEM)
        {
            alert!("ctx": "boot", "op": "unshare_ipc_thread",
                "msg": format!("failed to unshare(CLONE_FS|CLONE_FILES|CLONE_SYSVSEM): {errno}"),
                "err": errno as i32);
            std::process::exit(101);
        }

        // SAFETY: notif_pipe points to valid FDs.
        let (pipe_rd, pipe_wr) = unsafe {
            (
                SafeOwnedFd::from_raw_fd(notif_pipe.0),
                SafeOwnedFd::from_raw_fd(notif_pipe.1),
            )
        };
        drop(pipe_rd);
        let buf = [42u8; 1];
        match retry_on_eintr(|| write(&pipe_wr, &buf))? {
            0 => return Err(Errno::EIO.into()),
            1 => {}
            n => unreachable!("BUG: invalid pipe write of size {n}!"),
        }

        // Close the notification pipe.
        drop(pipe_wr);

        // IPC worker needs to inherit only the following FDs:
        // 1. epoll(7) FD.
        // 2. IPC socket FD.
        // 3. Log FD.
        // We have to sort the set as the FDs are randomized.
        let poll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        let sock = self.sock.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        #[expect(clippy::cast_sign_loss)]
        let mut set = vec![
            poll.0.as_raw_fd() as libc::c_uint,
            sock.as_raw_fd() as libc::c_uint,
            crate::log::LOG_FD.load(Ordering::Relaxed) as libc::c_uint,
        ];
        set.sort_unstable();
        closeexcept(&set)?;

        Ok(())
    }

    // Main loop: waits on epoll(7), accepts new connections, reads
    // commands, and writes responses.
    #[expect(clippy::cognitive_complexity)]
    fn main(&self) -> SydResult<()> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        let sock = self.sock.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        let sock_fd = sock.as_fd().as_raw_fd();

        // Confine `syd_ipc' thread.
        // We use exit_group(2) here to bail, because this confinement
        // is a critical safety feature.
        let ctx = match Self::prepare_confine(
            epoll.0.as_raw_fd(),
            self.options,
            &self.transit_uids,
            &self.transit_gids,
            false,
        ) {
            Ok(ctx) => ctx,
            Err(error) => {
                let errno = error.errno().unwrap_or(Errno::ENOSYS);
                alert!("ctx": "boot", "op": "confine_ipc_thread",
                    "msg": format!("failed to confine: {error}"),
                    "err": errno as i32);
                std::process::exit(101);
            }
        };

        // Load seccomp(2) BPF into the kernel.
        // We use exit_group(2) here to bail, because this confinement
        // is a critical safety feature.
        if let Err(error) = ctx.load() {
            let errno = scmp2no(&error).unwrap_or(Errno::ENOSYS);
            alert!("ctx": "boot", "op": "confine_ipc_thread",
                "msg": format!("failed to confine: {error}"),
                "err": errno as i32);
            std::process::exit(101);
        }
        drop(ctx);

        info!("ctx": "confine", "op": "confine_ipc_thread",
            "msg": "IPC thread confined");

        // Set of active connections.
        let mut connections = ConnectionMap::default();

        // Buffer for epoll events.
        // TODO: IPC_EPOLL_MAX_EVENTS=1024 move to config.rs
        let mut events = [EpollEvent::empty(); 1024];

        // Wait for events, block indefinitely.
        loop {
            let nfds = match epoll.wait(&mut events, PollTimeout::NONE) {
                Ok(n) => n,
                Err(Errno::EINTR) => continue,
                Err(errno) => return Err(errno.into()),
            };

            for ev in events.iter().take(nfds) {
                #[expect(clippy::cast_possible_truncation)]
                let fd = ev.data() as RawFd;
                let flags = ev.events();

                if fd == sock_fd {
                    // New incoming connection(s).
                    self.accept_new_connections(&mut connections)?;
                } else {
                    // Existing connection.
                    if flags.contains(EpollFlags::EPOLLIN) {
                        self.handle_readable(&mut connections, fd)?;
                    }
                    if flags.contains(EpollFlags::EPOLLOUT) {
                        self.handle_writable(&mut connections, fd)?;
                    }
                    if flags.intersects(
                        EpollFlags::EPOLLERR | EpollFlags::EPOLLHUP | EpollFlags::EPOLLRDHUP,
                    ) {
                        // Error or hang-up: close connection.
                        self.close_connection(&mut connections, fd)?;
                    }
                }
            }
        }
    }

    // Accept new connections, set them non-blocking, register with epoll.
    fn accept_new_connections(&self, connections: &mut ConnectionMap) -> Result<(), Errno> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
        let sock = self.sock.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        loop {
            #[expect(unreachable_patterns)]
            match accept4(
                sock.as_raw_fd(),
                SockFlag::SOCK_CLOEXEC | SockFlag::SOCK_NONBLOCK,
            )
            .map(|fd| {
                // SAFETY: accept4(2) returns a valid FD on success.
                #[expect(clippy::disallowed_types)]
                UnixStream::from(unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) })
            }) {
                Ok(stream) => {
                    // Register the new socket with epoll for reading.
                    let fd = stream.as_fd().as_raw_fd();
                    #[expect(clippy::cast_sign_loss)]
                    let event = libc::epoll_event {
                        events: EpollFlags::EPOLLIN.bits() as u32,
                        u64: fd as u64,
                    };
                    epoll_ctl_safe(&epoll.0, fd, Some(event))?;

                    // Create Connection state.
                    let creds = peer_creds(&stream)?;
                    let mut conn = Connection::new(creds, stream);

                    // In prompt mode, send initial prompt.
                    if conn.mode == PromptMode::Prompt {
                        conn.enqueue_prompt();
                    }
                    connections.insert(fd, conn);
                }
                Err(Errno::EINTR) => continue,
                Err(Errno::EAGAIN | Errno::EWOULDBLOCK) => {
                    // No more pending connections.
                    break;
                }
                Err(errno) => {
                    // Unexpected error.
                    return Err(errno);
                }
            }
        }

        Ok(())
    }

    // Handle a readable event on connection `fd`.
    #[expect(clippy::cognitive_complexity)]
    fn handle_readable(&self, connections: &mut ConnectionMap, fd: RawFd) -> Result<(), Errno> {
        let conn = if let Some(conn) = connections.get_mut(&fd) {
            conn
        } else {
            return Ok(());
        };

        loop {
            let mut buf = [0u8; LINE_MAX];
            let mut iov = [IoSliceMut::new(&mut buf)];
            let cmsg_siz = libc::ucred::cmsg_space();
            let mut cmsg = vec![0u8; cmsg_siz];

            let mut hdr = MsgHdr::default();
            hdr.set_iov_mut(&mut iov);
            hdr.set_control(&mut cmsg);
            #[expect(clippy::arithmetic_side_effects)]
            #[expect(unreachable_patterns)]
            match recvmsg(&conn.stream, &mut hdr, MsgFlags::MSG_DONTWAIT) {
                Ok(msg) if msg.bytes == 0 => {
                    // EOF on read side: if there's a pending write,
                    // defer closing until after flush.
                    if conn.write_buf.is_empty() {
                        // no response queued => close immediately.
                        self.close_connection(connections, fd)?;
                    } else {
                        // response pending => mark to close after writing.
                        conn.should_close = true;
                    }
                    return Ok(());
                }
                Ok(msg) if conn.buf.len() + msg.bytes >= LINE_MAX => {
                    // Input too large:
                    // 1. Add EPOLLOUT to interests.
                    // 2. Reject with error message.
                    // 3. Close connection after reply.
                    let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
                    conn.ctl(epoll, EpollFlags::EPOLLOUT)?;
                    conn.enqueue_response(IPC_RATE);
                    conn.should_close = true;

                    info!("ctx": "ipc", "op": "reject_ipc_command",
                        "msg": format!("rejected IPC command from pid:{} with uid:{}/gid:{}",
                            conn.creds.pid(),
                            conn.creds.uid(),
                            conn.creds.gid()),
                        "len": conn.buf.len() + msg.bytes,
                        "conn": &conn);
                    return Ok(());
                }
                Ok(msg) => {
                    // Authenticate each and every message.
                    let mut auth = false;
                    if let Ok(cmsgs) = msg.cmsgs() {
                        for cmsg in cmsgs {
                            let creds = if let CmsgOwned::ScmCredentials(creds) = cmsg {
                                creds
                            } else {
                                continue;
                            };

                            if creds.pid() == 0 {
                                // Invalid credentials, skip.
                                continue;
                            }

                            if creds.uid() == 0 && creds.gid() == 0 {
                                // Matched UID and GID for ROOT: Authenticated.
                                auth = true;
                                break;
                            }

                            if let Some(uid) = self.creds.0 {
                                if creds.uid() != uid.as_raw() {
                                    // UID mismatch.
                                    continue;
                                }
                            }

                            if let Some(gid) = self.creds.1 {
                                if creds.gid() != gid.as_raw() {
                                    // GID mismatch.
                                    continue;
                                }
                            }

                            // Matched UID and GID: Authenticated.
                            auth = true;
                            break;
                        }
                    }

                    if !auth {
                        // Authentication failed:
                        // 1. Add EPOLLOUT to interests.
                        // 2. Reject with error message.
                        // 3. Close connection after reply.
                        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
                        conn.ctl(epoll, EpollFlags::EPOLLOUT)?;
                        conn.enqueue_response(IPC_AUTH);
                        conn.should_close = true;

                        return Ok(());
                    }

                    // Extend buffer with the message content.
                    conn.buf.extend_from_slice(&buf[..msg.bytes]);

                    // Process any full lines in buffer.
                    while let Some(pos) = memchr(b'\n', &conn.buf) {
                        // Handle the line.
                        self.process_command(conn, pos)?;

                        // Remove the line and the '\n' from the buffer.
                        conn.buf.drain(..=pos);
                    }
                }
                Err(Errno::EINTR) => continue,
                Err(Errno::EAGAIN | Errno::EWOULDBLOCK) => {
                    // No more data.
                    break;
                }
                Err(_) => {
                    // Other errors -> close connection.
                    self.close_connection(connections, fd)?;
                    return Ok(());
                }
            }
        }

        Ok(())
    }

    // Handle a writable event on connection `fd` (flush pending writes).
    fn handle_writable(&self, connections: &mut ConnectionMap, fd: RawFd) -> Result<(), Errno> {
        let conn = if let Some(conn) = connections.get_mut(&fd) {
            conn
        } else {
            return Ok(());
        };
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        while conn.write_pos < conn.write_buf.len() {
            let iov = [IoSlice::new(&conn.write_buf[conn.write_pos..])];

            #[expect(clippy::arithmetic_side_effects)]
            #[expect(unreachable_patterns)]
            match sendmsg::<_, SockaddrStorage>(
                &conn.stream,
                &iov,
                &[],
                MsgFlags::MSG_DONTWAIT,
                None,
            ) {
                Ok(0) => {
                    // Would block or closed; stop for now.
                    break;
                }
                Ok(n) => {
                    conn.write_pos += n;
                }
                Err(Errno::EINTR) => continue,
                Err(Errno::EAGAIN | Errno::EWOULDBLOCK) => {
                    // Can't write more right now.
                    break;
                }
                Err(_) => {
                    // Fatal write error; close.
                    self.close_connection(connections, fd)?;
                    return Ok(());
                }
            }
        }

        if conn.write_pos >= conn.write_buf.len() {
            // All data written.
            // Clear buffer and disable EPOLLOUT.
            conn.ctl(epoll, EpollFlags::EPOLLIN)?;
            conn.write_buf.clear();
            conn.write_pos = 0;

            // If mode is NonInteractive and last command closed the
            // connection, we close the connection here.
            if conn.should_close {
                self.close_connection(connections, fd)?;
            }
        }

        Ok(())
    }

    // Process at `pos` boundary from connection `conn` on `fd`.
    #[expect(clippy::cognitive_complexity)]
    fn process_command(&self, conn: &mut Connection, pos: usize) -> Result<(), Errno> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        // Convert to UTF-8, errors are fatal.
        let cmd = String::from_utf8(conn.buf[..pos].to_vec()).or(Err(Errno::EINVAL))?;

        // Split on ASCII whitespace.
        let mut parts = cmd.split_ascii_whitespace();

        match parts.next() {
            Some("prompt") => {
                // Change mode.
                match parts.next() {
                    Some("n") => {
                        conn.mode = PromptMode::NonInteractive;
                        conn.should_close = true;
                    }
                    Some("i") => {
                        conn.mode = PromptMode::Interactive;
                    }
                    Some("p") => {
                        conn.mode = PromptMode::Prompt;
                        conn.enqueue_prompt();
                    }
                    None => {
                        // Toggle:
                        // 1. From NonInteractive => Prompt;
                        // 2. From Interactive => Prompt;
                        // 3. From Prompt => Interactive.
                        match conn.mode {
                            PromptMode::NonInteractive => {
                                conn.mode = PromptMode::Prompt;
                                conn.enqueue_prompt();
                            }
                            PromptMode::Interactive => {
                                conn.mode = PromptMode::Prompt;
                                conn.enqueue_prompt();
                            }
                            PromptMode::Prompt => {
                                conn.mode = PromptMode::Interactive;
                            }
                        }
                    }
                    _ => {
                        conn.enqueue_response(b"Unknown prompt mode!\n");
                        if conn.mode == PromptMode::Prompt {
                            conn.enqueue_prompt();
                        }
                    }
                }
            }
            Some("quit") | Some("exit") => conn.should_close = true,
            Some(part) => {
                let response = if part == "ping" {
                    // Alive check, send pong to ping.
                    Cow::Borrowed(IPC_PONG)
                } else if part == "pink" {
                    // Pink Floyd check, reply quote.
                    Cow::Borrowed(IPC_PINK)
                } else if part == "version" {
                    Cow::Owned(format!(
            "{{\"major\":{API_MAJOR_VERSION},\"minor\":{IPC_MINOR_VERSION},\"version\":\"{API_MAJOR_VERSION}.{IPC_MINOR_VERSION}\"}}\n").into_bytes())
                } else {
                    // Regular command: pass to handle_cmd.
                    Cow::Owned(self.handle_cmd(&cmd)?)
                };
                conn.enqueue_response(&response);
                if conn.mode == PromptMode::Prompt {
                    conn.enqueue_prompt();
                } else if conn.mode == PromptMode::NonInteractive {
                    conn.should_close = true;
                }
                info!("ctx": "ipc", "op": "handle_ipc_command",
                    "msg": format!("handled IPC command from pid:{} with uid:{}/gid:{}",
                        conn.creds.pid(),
                        conn.creds.uid(),
                        conn.creds.gid()),
                    "cmd": &cmd,
                    "resp": XPath::from_bytes(&response),
                    "conn": &conn);
            }
            None => {
                // Empty line.
                // Just send prompt if in prompt mode.
                if conn.mode == PromptMode::Prompt {
                    conn.enqueue_prompt();
                }
            }
        }

        // After enqueueing data, ensure EPOLLOUT is enabled.
        if !conn.write_buf.is_empty() {
            conn.ctl(epoll, EpollFlags::EPOLLIN | EpollFlags::EPOLLOUT)?;
        }

        Ok(())
    }

    // Close and clean up connection `fd`.
    fn close_connection(&self, connections: &mut ConnectionMap, fd: RawFd) -> Result<(), Errno> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        // Unregister from epoll(7).
        epoll_ctl_safe(&epoll.0, fd, None)?;

        // Remove from set which drops the stream.
        connections.remove(&fd);

        Ok(())
    }

    // Process a single command string, and return a response buffer.
    fn handle_cmd(&self, cmd: &str) -> Result<Vec<u8>, Errno> {
        // Lock sandbox for write.
        let sandbox = self.sandbox.as_ref().ok_or(Errno::EOWNERDEAD)?;
        let mut sandbox = sandbox.write().unwrap_or_else(|err| err.into_inner());

        // Close connection immediately if sandbox is locked.
        // This also results in thread-exit.
        if sandbox.is_locked() {
            return Err(Errno::EPERM);
        }

        // Intercept display commands and send to socket
        // instead of Syd's standard error:
        // - stat  Prints Sandbox in compact JSON.
        // - stats Prints Sandbox in human-readable format.
        if is_equal(cmd.as_bytes(), b"stat") {
            return Ok(serde_json::to_string(&*sandbox)
                .or(Err(Errno::EINVAL))?
                .into_bytes());
        } else if is_equal(cmd.as_bytes(), b"stats") {
            return Ok(format!("{sandbox}").into_bytes());
        }

        // Use config_unchecked to skip logging, and env init.
        let result = sandbox.config_unchecked(cmd);

        // Close connection immediately if sandbox got locked.
        // This also results in thread-exit.
        if sandbox.is_locked() {
            return Err(Errno::EPERM);
        }

        match result {
            Ok(()) => Ok(IPC_ACK.to_vec()),
            Err(errno) => {
                Ok(format!("{{\"err\":{},\"msg\":\"{errno}\"}}\n", errno as i32).into_bytes())
            }
        }
    }
}