syd 3.56.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::{BufReader, IoSlice, IoSliceMut, Write},
    os::{
        fd::{AsFd, AsRawFd, FromRawFd, IntoRawFd, RawFd},
        unix::net::{UnixListener, UnixStream},
    },
    sync::{atomic::Ordering, Arc, RwLock},
    thread,
    time::Instant,
};

use dur::Duration;
use libseccomp::{ScmpAction, ScmpFilterContext};
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::{
            bind, listen, setsockopt, socket, sockopt::PassCred, Backlog, SockFlag,
            SockaddrStorage, UnixAddr, UnixCredentials,
        },
        stat::{umask, Mode},
    },
    unistd::{Gid, Uid},
};
use serde::{ser::SerializeMap, Serialize, Serializer};

use crate::{
    alert,
    compat::{
        epoll_ctl_mod_safe, epoll_ctl_safe, CmsgOwned, CmsgSpace, MsgFlags, MsgHdr, SockType,
        SCM_MAX_FD,
    },
    config::*,
    confine::{
        confine_scmp_accept4, confine_scmp_close, confine_scmp_close_range, confine_scmp_epoll,
        confine_scmp_fcntl, confine_scmp_getsockopt, confine_scmp_madvise, confine_scmp_open_stat,
        confine_scmp_prctl, confine_scmp_read, confine_scmp_recvmsg, confine_scmp_sendmsg,
        confine_scmp_setid, confine_scmp_statx, confine_scmp_write, confine_scmp_wx_syd, Sydcall,
    },
    cookie::{safe_accept4, safe_getpeercred, safe_recvmsg, safe_sendmsg, safe_write},
    err::{err2no, scmp2no, SydJoinHandle, SydResult},
    error,
    fd::{close, closeall, closeexcept, is_file, SafeOwnedFd},
    hash::SydHashMap,
    id::SydId,
    info,
    landlock::Errata,
    landlock_policy::LandlockPolicy,
    log::WriteBuf,
    path::{XPath, XPathBuf},
    retry::retry_on_eintr,
    rng::duprand,
    sandbox::{ConfigLoader, Options, Sandbox, LINE_MAX},
    warn, xfmt,
};

const IPC_ACK: &[u8] = b"{\"err\":0,\"msg\":\"ACK\"}\n";
const IPC_AUTH: &[u8] = b"{\"err\":13,\"msg\":\"AUTH\"}\n";
const IPC_NFDS: &[u8] = b"{\"err\":24,\"msg\":\"NFDS\"}\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>),

    // Maximum number of concurrent connections.
    max_connections: usize,
    // Idle timeout for connections.
    idle_timeout: Duration,

    // 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,
    // Monotonic timestamp of last received message.
    last_active: Instant,
}

// 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) -> Result<Self, Errno> {
        let mut buf = Vec::new();
        buf.try_reserve(IPC_CONN_BUF_CAP).or(Err(Errno::ENOMEM))?;
        let mut write_buf = Vec::new();
        write_buf
            .try_reserve(IPC_CONN_BUF_CAP)
            .or(Err(Errno::ENOMEM))?;

        Ok(Self {
            creds,
            stream,
            buf,
            write_buf,
            write_pos: 0,
            mode: PromptMode::NonInteractive,
            should_close: false,
            last_active: Instant::now(),
        })
    }

    // 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]) -> Result<(), Errno> {
        self.write_buf
            .try_reserve(data.len())
            .or(Err(Errno::ENOMEM))?;
        self.write_buf.extend_from_slice(data);
        Ok(())
    }

    // Enqueue the prompt string ("; ") in "write_buf".
    fn enqueue_prompt(&mut self) -> Result<(), Errno> {
        self.enqueue_response(b"; ")
    }

    // Reject connection with given errno and close once reply is flushed.
    fn reject(&mut self, epoll: &Epoll, errno: Errno) -> Result<(), Errno> {
        let msg = match errno {
            Errno::EBADMSG => "EBADMSG",
            Errno::EILSEQ => "EILSEQ",
            Errno::EPROTO => "EPROTO",
            _ => unreachable!(
                "BUG: IpcWorker::reject called with invalid errno: {errno}, report a bug!"
            ),
        };

        let mut buf = WriteBuf::new();
        writeln!(buf, "{{\"err\":{},\"msg\":\"{msg}\"}}", errno as i32).map_err(|e| err2no(&e))?;

        self.ctl(epoll, EpollFlags::EPOLLOUT)?;
        self.enqueue_response(&buf.0)?;
        self.should_close = true;

        Ok(())
    }
}

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.
    #[expect(clippy::too_many_arguments)]
    pub(crate) fn new(
        addr: &XPath,
        uid: Option<Uid>,
        gid: Option<Gid>,
        max_connections: usize,
        idle_timeout: Duration,
        options: Options,
        transit_uids: &[(Uid, Uid)],
        transit_gids: &[(Gid, Gid)],
    ) -> Result<Self, Errno> {
        let addr = XPathBuf::try_from(addr)?;

        let mut tu = Vec::new();
        tu.try_reserve_exact(transit_uids.len())
            .or(Err(Errno::ENOMEM))?;
        tu.extend_from_slice(transit_uids);

        let mut tg = Vec::new();
        tg.try_reserve_exact(transit_gids.len())
            .or(Err(Errno::ENOMEM))?;
        tg.extend_from_slice(transit_gids);

        Ok(IpcWorker {
            creds: (uid, gid),
            addr: Some(addr),
            epoll: None,
            sock: None,
            sandbox: None,
            max_connections,
            idle_timeout,
            options,
            transit_uids: tu,
            transit_gids: tg,
        })
    }

    // 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(0o177));
        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)?;

        // Syscall argument cookies may be disabled
        // at startup with trace/allow_unsafe_nocookie:1.
        let restrict_cookie = !options.allow_unsafe_nocookie();

        // 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.
        // Allow statx(2) with syscall argument cookies to check file type.
        confine_scmp_open_stat(&mut ctx, true /* openat2 */, false /* statx */)?;
        confine_scmp_statx(&mut ctx, restrict_cookie)?;

        // 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 reads to parse configuration file over SCM_RIGHTS.
        confine_scmp_read(&mut ctx, 0x10000, restrict_cookie)?;

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

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

        // close(2) and close_range(2) may be used only with syscall argument cookies.
        confine_scmp_close(&mut ctx, restrict_cookie)?;
        confine_scmp_close_range(&mut ctx, restrict_cookie)?;

        // Allow UNIX networking.
        confine_scmp_accept4(&mut ctx, restrict_cookie)?;
        confine_scmp_recvmsg(&mut ctx, restrict_cookie)?;
        confine_scmp_sendmsg(&mut ctx, restrict_cookie)?;
        confine_scmp_getsockopt(&mut ctx, IPC_GETSOCKOPT_OPS, restrict_cookie)?;

        // 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 Sydcall::from_name(sysname) {
                Ok(syscall) => {
                    ctx.add_rule(ScmpAction::Allow, syscall)?;
                }
                Err(_) => {
                    info!("ctx": "confine", "op": "allow_ipc_syscall",
                        "msg": xfmt!("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> {
        let base = SydId::get_name("syd_ipc");
        let mut name = String::new();
        name.try_reserve(base.len()).or(Err(Errno::ENOMEM))?;
        name.push_str(base);

        thread::Builder::new()
            .name(name)
            .stack_size(IPC_STACK_SIZE)
            .spawn(move || {
                if let Err(error) = self.init(notif_pipe) {
                    let errno = error.errno().unwrap_or(Errno::ENOSYS);

                    error!("ctx": "ipc", "op": "init",
                        "err": errno as i32,
                        "msg": xfmt!("failed to initialize IPC worker: {errno}"));

                    return Err(error);
                }

                match self.main() {
                    Ok(()) => Ok(()),

                    Err(error) => {
                        let errno = error.errno().unwrap_or(Errno::ENOSYS);

                        error!("ctx": "ipc", "op": "main",
                            "err": errno as i32,
                            "msg": xfmt!("exited IPC worker with error: {errno}"));

                        Err(error)
                    }
                }
            })
            .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": xfmt!("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(|| safe_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)?;
        let mut set: Vec<libc::c_uint> = Vec::new();
        set.try_reserve_exact(3).or(Err(Errno::ENOMEM))?;
        #[expect(clippy::cast_sign_loss)]
        {
            set.push(poll.0.as_raw_fd() as libc::c_uint);
            set.push(sock.as_raw_fd() as libc::c_uint);
            set.push(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": xfmt!("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": xfmt!("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.
        let mut events = [EpollEvent::empty(); IPC_EPOLL_MAX_EVENTS];

        // Wait for events, block as necessary.
        loop {
            let timeout = if connections.is_empty() {
                PollTimeout::NONE
            } else {
                PollTimeout::from(IPC_IDLE_POLL_MS)
            };

            let nfds = match epoll.wait(&mut events, timeout) {
                Ok(n) => n,
                Err(Errno::EINTR) => continue,
                Err(errno) => return Err(errno.into()),
            };

            for ev in events.iter().take(nfds) {
                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)?;
                    }
                }
            }

            // Reap idle connections.
            if !connections.is_empty() {
                self.reap_idle_connections(&mut connections)?;
            }
        }
    }

    // Close connections idle longer than the idle timeout.
    fn reap_idle_connections(&self, connections: &mut ConnectionMap) -> Result<(), Errno> {
        let mut expired: Vec<RawFd> = Vec::new();
        expired
            .try_reserve(connections.len())
            .or(Err(Errno::ENOMEM))?;

        for (fd, conn) in connections.iter() {
            if Duration::from(conn.last_active.elapsed()) >= self.idle_timeout {
                expired.push(*fd);
            }
        }

        self.close_connections(connections, &expired)
    }

    // Close a batch of connections in as few close_range(2) calls as possible.
    fn close_connections(
        &self,
        connections: &mut ConnectionMap,
        fds: &[RawFd],
    ) -> Result<(), Errno> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        let mut raw_fds = Vec::new();
        raw_fds.try_reserve(fds.len()).or(Err(Errno::ENOMEM))?;

        for &fd in fds {
            if let Some(conn) = connections.remove(&fd) {
                epoll_ctl_safe(&epoll.0, fd, None)?;

                let raw_fd = conn.stream.into_raw_fd();
                let raw_fd = libc::c_uint::try_from(raw_fd).or(Err(Errno::EBADF))?;

                raw_fds.push(raw_fd);
            }
        }

        raw_fds.sort_unstable();
        closeall(&raw_fds)
    }

    // 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 safe_accept4(
                sock,
                SockFlag::SOCK_CLOEXEC | SockFlag::SOCK_NONBLOCK,
                false, /* want_src_addr */
            )
            .map(|(fd, _)| {
                #[expect(clippy::disallowed_types)]
                UnixStream::from(std::os::fd::OwnedFd::from(fd))
            }) {
                Ok(stream) => {
                    let fd = stream.as_fd().as_raw_fd();

                    // Enforce concurrent connection limit.
                    if connections.len() >= self.max_connections {
                        warn!("ctx": "ipc", "op": "reject",
                            "max": self.max_connections,
                            "msg": xfmt!("rejected IPC connection due to limit `{}'",
                                self.max_connections));
                        let _ = close(stream);
                        continue;
                    }
                    connections.try_reserve(1).or(Err(Errno::ENOMEM))?;

                    // Register the new socket with epoll for reading.
                    #[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 = safe_getpeercred(&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()
                .checked_add(<[RawFd; SCM_MAX_FD]>::cmsg_space())
                .ok_or(Errno::EOVERFLOW)?;

            let mut cmsg = Vec::new();
            cmsg.try_reserve(cmsg_siz).or(Err(Errno::ENOMEM))?;
            cmsg.resize(cmsg_siz, 0u8);

            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 safe_recvmsg(
                &conn.stream,
                &mut hdr,
                MsgFlags::MSG_DONTWAIT | MsgFlags::MSG_CMSG_CLOEXEC,
            ) {
                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": xfmt!("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) => {
                    // Refresh idle deadline on every message.
                    conn.last_active = Instant::now();

                    // Authenticate each and every message.
                    let mut auth = false;
                    let mut nfds: usize = 0;
                    let mut load_fd: Option<SafeOwnedFd> = None;
                    let mut unknown = false;
                    match msg.cmsgs() {
                        Ok(cmsgs) => {
                            for cmsg in cmsgs {
                                let creds = match cmsg {
                                    CmsgOwned::ScmCredentials(creds) => creds,
                                    CmsgOwned::ScmRights(fds) => {
                                        nfds = nfds.saturating_add(fds.len());
                                        for fd in fds {
                                            if load_fd.is_none() {
                                                load_fd = Some(fd);
                                                break;
                                            }
                                        }
                                        continue;
                                    }
                                    _ => {
                                        // Unknown control messages are rejected below.
                                        unknown = true;
                                        continue;
                                    }
                                };

                                if auth {
                                    // Already authenticated.
                                    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;
                                    continue;
                                }

                                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;
                            }
                        }
                        // Unknown control messages are rejected below.
                        Err(_) => unknown = true,
                    }

                    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(());
                    }

                    // Reject unknown or invalid control messages.
                    if unknown {
                        // Unexpected control message:
                        // 1. Reject with EPROTO.
                        // 2. Close connection after reply.
                        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
                        conn.reject(epoll, Errno::EPROTO)?;

                        info!("ctx": "ipc", "op": "reject_ipc_command",
                            "msg": xfmt!("rejected IPC message with invalid control message from pid:{} with uid:{}/gid:{}",
                                conn.creds.pid(),
                                conn.creds.uid(),
                                conn.creds.gid()),
                            "conn": &conn);
                        return Ok(());
                    }

                    // Reject more than one file descriptor per message.
                    if nfds > 1 {
                        // Too many file descriptors:
                        // 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_NFDS)?;
                        conn.should_close = true;

                        info!("ctx": "ipc", "op": "reject_ipc_command",
                            "msg": xfmt!("rejected IPC message with {} file descriptors from pid:{} with uid:{}/gid:{}",
                                nfds,
                                conn.creds.pid(),
                                conn.creds.uid(),
                                conn.creds.gid()),
                            "max": 1,
                            "conn": &conn);
                        return Ok(());
                    }

                    // Load configuration from SCM_RIGHTS as necessary.
                    if let Some(fd) = load_fd {
                        if !conn.buf.is_empty() || !is_equal(&buf[..msg.bytes], b"load\n") {
                            // Ambiguous use of descriptor passing:
                            // 1. Reject with EBADMSG.
                            // 2. Close connection after reply.
                            let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;
                            conn.reject(epoll, Errno::EBADMSG)?;

                            info!("ctx": "ipc", "op": "reject_ipc_command",
                                "msg": xfmt!("rejected ambiguous IPC load from pid:{} with uid:{}/gid:{}",
                                    conn.creds.pid(),
                                    conn.creds.uid(),
                                    conn.creds.gid()),
                                "conn": &conn);
                            return Ok(());
                        }

                        // Handle load request.
                        self.process_load(conn, fd)?;

                        continue;
                    }

                    // Extend buffer with the message content.
                    conn.buf.try_reserve(msg.bytes).or(Err(Errno::ENOMEM))?;
                    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 safe_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 the buffer.
            conn.write_buf.clear();
            conn.write_pos = 0;

            if conn.should_close {
                // NonInteractive mode and the last command closed the
                // connection: close it here.
                self.close_connection(connections, fd)?;
            } else {
                // Disable EPOLLOUT and wait for the next request.
                conn.ctl(epoll, EpollFlags::EPOLLIN)?;
            }
        }

        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)?;

        // Reject interior NUL bytes.
        if memchr(0, &conn.buf[..pos]).is_some() {
            return conn.reject(epoll, Errno::EILSEQ);
        }

        // Allocate buffer gracefully.
        let mut line = Vec::new();
        line.try_reserve(pos).or(Err(Errno::ENOMEM))?;
        line.extend_from_slice(&conn.buf[..pos]);

        // Reject invalid UTF-8.
        let cmd = match String::from_utf8(line) {
            Ok(cmd) => cmd,
            Err(_) => return conn.reject(epoll, Errno::EILSEQ),
        };

        // 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" {
                    // Build Syd version response.
                    let mut buf = WriteBuf::new();
                    writeln!(buf, "{{\"major\":{API_MAJOR_VERSION},\"minor\":{IPC_MINOR_VERSION},\"version\":\"{API_MAJOR_VERSION}.{IPC_MINOR_VERSION}\"}}")
                        .map_err(|e| err2no(&e))?;
                    Cow::Owned(buf.0)
                } else if part == "load" {
                    // Reject load command without SCM_RIGHTS.
                    Cow::Owned(Self::errno_response(Errno::EBADF)?)
                } 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": xfmt!("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)?;

        // Remove from connection map, and close connection.
        if let Some(conn) = connections.remove(&fd) {
            // Unregister from epoll(7).
            epoll_ctl_safe(&epoll.0, fd, None)?;

            let _ = close(conn.stream);
        }

        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") {
            let mut buf = WriteBuf::new();
            serde_json::to_writer(&mut buf, &*sandbox).or(Err(Errno::ENOMEM))?;
            return Ok(buf.0);
        } else if is_equal(cmd.as_bytes(), b"stats") {
            let mut buf = WriteBuf::new();
            write!(buf, "{sandbox}").map_err(|e| err2no(&e))?;
            return Ok(buf.0);
        }

        // 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(()) => Self::ack_response(),
            Err(errno) => Self::errno_response(errno),
        }
    }

    // Process a configuration load request, and return a response buffer.
    fn handle_load(&self, fd: SafeOwnedFd) -> 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);
        }

        let result = (|| {
            // Loads are rejected in lock:drop mode.
            if sandbox.is_drop() {
                return Err(Errno::EPERM);
            }

            // Ensure regular file before parsing.
            if !is_file(&fd)? {
                return Err(Errno::EBADFD);
            }

            // Parse configuration file rejecting include directives.
            let name = XPath::from_bytes(MAGIC_LOAD);
            let file = BufReader::new(fd);
            sandbox.parse_config(file, name, &mut ConfigLoader::sealed())
        })();

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

        match result {
            Ok(()) => Self::ack_response(),
            Err(errno) => Self::errno_response(errno),
        }
    }

    // Process a configuration load request.
    fn process_load(&self, conn: &mut Connection, file: SafeOwnedFd) -> Result<(), Errno> {
        let epoll = self.epoll.as_ref().ok_or(Errno::EADDRNOTAVAIL)?;

        let response = self.handle_load(file)?;
        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_load",
            "msg": xfmt!("handled IPC load from pid:{} with uid:{}/gid:{}",
                conn.creds.pid(),
                conn.creds.uid(),
                conn.creds.gid()),
            "resp": XPath::from_bytes(&response),
            "conn": &conn);

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

        Ok(())
    }

    // Build an acknowledgement response buffer.
    fn ack_response() -> Result<Vec<u8>, Errno> {
        let mut buf = WriteBuf::new();
        buf.try_reserve(IPC_ACK.len()).or(Err(Errno::ENOMEM))?;
        buf.0.extend_from_slice(IPC_ACK);
        Ok(buf.0)
    }

    // Build an error response buffer for the given errno.
    fn errno_response(errno: Errno) -> Result<Vec<u8>, Errno> {
        let mut buf = WriteBuf::new();
        writeln!(buf, "{{\"err\":{},\"msg\":\"{errno}\"}}", errno as i32)
            .map_err(|e| err2no(&e))?;
        Ok(buf.0)
    }
}