//! The execution gate: fanotify permission events.
//!
//! This is the feature that separates Hound from a cron job wrapped
//! around a scanner. inotify tells you a file appeared *after* it
//! appeared, which is forensics. fanotify's `FAN_OPEN_EXEC_PERM` hands us
//! the open and waits for an answer, so we can refuse to let a binary run
//! at all.
//!
//! It is also the single most dangerous thing in the codebase. Every
//! permission event we fail to answer is a process frozen mid-`execve`,
//! and a mark on the root filesystem means *every* process. Three rules
//! follow from that, and none of them are negotiable:
//!
//! 1. **Fail open, always.** A watchdog thread answers anything the scan
//!    loop has not answered within [`DEADLINE`]. A missed detection is a
//!    bad day; a wedged machine is the end of the product.
//! 2. **Never gate ourselves.** Scanning a file means opening it, which
//!    generates an event, which we would then wait on ourselves to
//!    answer. Events from our own pid are allowed before anything else
//!    happens.
//! 3. **Answer exactly once, and always close the fd.** The kernel hands
//!    us an open descriptor per event. Leak them and the daemon runs out
//!    of file descriptors, which fails us into the same wedge.
//!
//! Everything that decides *what* to do lives in [`policy`] and is pure,
//! so it is tested without root. The syscall layer below it is thin
//! enough to audit by eye.

use std::collections::HashSet;
use std::io;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

// ── ABI constants ───────────────────────────────────────────────────────
// Defined here rather than taken from libc so a libc version bump cannot
// silently change what we ask the kernel for.

const FAN_CLOEXEC: u32 = 0x0000_0001;
const FAN_NONBLOCK: u32 = 0x0000_0002;
const FAN_CLASS_CONTENT: u32 = 0x0000_0004;

const FAN_CLOSE_WRITE: u64 = 0x0000_0008;
const FAN_OPEN_PERM: u64 = 0x0001_0000;
const FAN_OPEN_EXEC_PERM: u64 = 0x0004_0000;

/// The permission bits. An event carrying one of these has a process
/// blocked behind it and MUST be answered; an event carrying none of them
/// is a notification and must NOT be — writing a response for one is a
/// protocol error.
const PERM_MASK: u64 = FAN_OPEN_PERM | FAN_OPEN_EXEC_PERM;

const FAN_MARK_ADD: u32 = 0x0000_0001;
const FAN_MARK_MOUNT: u32 = 0x0000_0010;
const FAN_MARK_FILESYSTEM: u32 = 0x0000_0100;

const FAN_ALLOW: u32 = 0x01;
const FAN_DENY: u32 = 0x02;

const FAN_METADATA_VERSION: u8 = 3;

/// How long a process may be held before we let it through regardless.
///
/// Our scans are single-digit milliseconds, so this is three orders of
/// magnitude of headroom. It exists for the pathological case — a stalled
/// NFS read, a pauseerd disk — not the normal one.
pub const DEADLINE: Duration = Duration::from_millis(500);

/// `struct fanotify_event_metadata` — 24 bytes, stable since Linux 2.6.37.
#[repr(C)]
#[derive(Clone, Copy, Debug)]
struct EventMetadata {
    event_len: u32,
    vers: u8,
    reserved: u8,
    metadata_len: u16,
    mask: u64,
    fd: i32,
    pid: i32,
}

const METADATA_SIZE: usize = std::mem::size_of::<EventMetadata>();

/// `struct fanotify_response`.
#[repr(C)]
struct Response {
    fd: i32,
    response: u32,
}

/// One decoded event.
#[derive(Debug)]
pub struct Event {
    /// Our own identity for this event.
    ///
    /// It exists because **a file descriptor number is not a stable
    /// identity**. The kernel allocates an fd per event and recycles the
    /// number as soon as we close it, so a single write produces an
    /// `FAN_OPEN_PERM` on fd 6 and then an `FAN_CLOSE_WRITE` on fd 6
    /// again. Keying "have I handled this?" on the fd made the second
    /// event look like a duplicate of the first and silently dropped it.
    /// Sequence numbers are never reused.
    pub seq: u64,
    /// Descriptor for the file. Ours to close, exactly once.
    pub fd: RawFd,
    /// The process being held, for permission events.
    pub pid: i32,
    pub mask: u64,
}

impl Event {
    /// True when this open is an `execve`, as opposed to an ordinary read.
    pub fn is_exec(&self) -> bool {
        self.mask & FAN_OPEN_EXEC_PERM != 0
    }

    /// True when a process is blocked waiting for our answer.
    ///
    /// The distinction drives everything downstream: a permission event is
    /// a deadline, a notification is a chore. Answering a notification is
    /// a protocol error, and failing to answer a permission event freezes
    /// a process.
    pub fn needs_response(&self) -> bool {
        self.mask & PERM_MASK != 0
    }

    /// True when this is a completed write — a file just changed on disk.
    /// Nobody is waiting; this is the post-hoc path that replaces what
    /// inotify used to do, with whole-filesystem coverage and no watch
    /// descriptor limit.
    pub fn is_write(&self) -> bool {
        self.mask & FAN_CLOSE_WRITE != 0
    }

    /// Resolve what the descriptor points at, via `/proc/self/fd`.
    ///
    /// This is a readlink, not an open, so it generates no event.
    pub fn path(&self) -> Option<PathBuf> {
        std::fs::read_link(format!("/proc/self/fd/{}", self.fd)).ok()
    }

    /// Full metadata for the descriptor, for cache lookups that must not
    /// read the file.
    pub fn metadata(&self) -> Option<std::fs::Metadata> {
        use std::os::unix::io::{AsRawFd, FromRawFd};
        // SAFETY: borrowed for the duration of the call and immediately
        // forgotten, so the descriptor is not closed twice.
        let f = unsafe { std::fs::File::from_raw_fd(self.fd) };
        let md = f.metadata().ok();
        let _ = f.as_raw_fd();
        std::mem::forget(f);
        md
    }

    /// Size of the file behind the descriptor, via `fstat`.
    pub fn size(&self) -> Option<u64> {
        // SAFETY: zeroed stat is a valid initial value; fd is ours.
        let mut st: libc::stat = unsafe { std::mem::zeroed() };
        let rc = unsafe { libc::fstat(self.fd, &mut st) };
        (rc == 0).then(|| st.st_size as u64)
    }

    /// Read the file's contents **from the descriptor the kernel gave us**.
    ///
    /// This is load-bearing, not an optimisation. Re-opening the path with
    /// `std::fs::read` issues a fresh `open()` on a watched mount, which
    /// queues a permission event behind the one we are currently holding —
    /// and we cannot answer that one until we finish this one. The daemon
    /// deadlocks against itself and takes every process touching the mount
    /// with it. Allowing our own pid in [`policy::decide`] does not save us,
    /// because the thread never gets back to the queue to apply it.
    ///
    /// `pread` also leaves the file offset alone, so the process we are
    /// gating sees exactly the file it opened.
    pub fn content(&self, max: usize) -> Option<Vec<u8>> {
        let size = self.size()? as usize;
        if size > max {
            return None;
        }
        let mut buf = vec![0u8; size];
        let mut read = 0usize;
        while read < size {
            // SAFETY: writing within buf's allocation, bounded by size.
            let n = unsafe {
                libc::pread(
                    self.fd,
                    buf[read..].as_mut_ptr() as *mut libc::c_void,
                    size - read,
                    read as libc::off_t,
                )
            };
            if n <= 0 {
                break;
            }
            read += n as usize;
        }
        buf.truncate(read);
        Some(buf)
    }
}

// ── policy: the pure half ───────────────────────────────────────────────

pub mod policy {
    use super::*;

    /// What to do with an event, decided before any I/O happens.
    #[derive(Debug, Clone, PartialEq, Eq)]
    pub enum Decision {
        /// Let it through without looking. Cheap and by far the common case.
        AllowNow(&'static str),
        /// Worth scanning before answering.
        Scan,
    }

    /// Decide without touching the disk.
    ///
    /// Ordering matters: the self-pid check comes first because getting it
    /// wrong deadlocks the daemon against itself, and no other rule can
    /// save us from that.
    pub fn decide(
        event_pid: i32,
        our_pid: i32,
        path: Option<&Path>,
        size: Option<u64>,
        max_size: u64,
        excludes: &[String],
    ) -> Decision {
        if event_pid == our_pid {
            return Decision::AllowNow("self");
        }
        let Some(path) = path else {
            // A descriptor we cannot resolve is one we cannot scan. Holding
            // the process would gain nothing.
            return Decision::AllowNow("unresolvable");
        };
        if is_excluded(path, excludes) {
            return Decision::AllowNow("excluded");
        }
        if size.is_some_and(|s| s > max_size) {
            return Decision::AllowNow("oversized");
        }
        Decision::Scan
    }

    /// Prefix match on path components, so `/var` never matches `/variable`.
    pub fn is_excluded(path: &Path, excludes: &[String]) -> bool {
        excludes.iter().any(|ex| {
            let ex = ex.trim_end_matches('/');
            if ex.is_empty() {
                return false;
            }
            path.starts_with(ex)
        })
    }
}

// ── the responder: answer once, close always ────────────────────────────

/// Owns the fanotify descriptor and guarantees each event is answered
/// exactly once, whether by the scan loop or by the watchdog.
pub struct Responder {
    fan: OwnedFd,
    /// Events already finished, by sequence number — never by fd, which
    /// the kernel recycles.
    answered: Mutex<HashSet<u64>>,
    allowed: AtomicU64,
    denied: AtomicU64,
    timed_out: AtomicU64,
}

impl Responder {
    fn new(fan: OwnedFd) -> Self {
        Self {
            fan,
            answered: Mutex::new(HashSet::new()),
            allowed: AtomicU64::new(0),
            denied: AtomicU64::new(0),
            timed_out: AtomicU64::new(0),
        }
    }

    /// Answer an event. Idempotent: a second call for the same descriptor
    /// is a no-op, which is what makes the watchdog safe to race with the
    /// scan loop.
    pub fn answer(&self, seq: u64, fd: RawFd, allow: bool, timed_out: bool) {
        self.finish(seq, fd, allow, timed_out, true)
    }

    /// Close a notification event's descriptor without answering it.
    ///
    /// The kernel hands out a descriptor for notifications too, and leaking
    /// them exhausts the fd table just as thoroughly as leaking a
    /// permission event's.
    pub fn close_only(&self, seq: u64, fd: RawFd) {
        self.finish(seq, fd, true, false, false)
    }

    fn finish(&self, seq: u64, fd: RawFd, allow: bool, timed_out: bool, respond: bool) {
        {
            let mut answered = self.answered.lock().expect("responder poisoned");
            if !answered.insert(seq) {
                return;
            }
        }

        if !respond {
            // SAFETY: we own this fd and nothing is waiting on it.
            unsafe { libc::close(fd) };
            return;
        }

        let response = Response {
            fd,
            response: if allow { FAN_ALLOW } else { FAN_DENY },
        };
        // SAFETY: writing a fanotify_response to the fanotify descriptor is
        // the documented way to answer a permission event.
        let written = unsafe {
            libc::write(
                self.fan.as_raw_fd(),
                &response as *const Response as *const libc::c_void,
                std::mem::size_of::<Response>(),
            )
        };
        if written < 0 {
            eprintln!("gate: failed to answer event: {}", io::Error::last_os_error());
        }

        if timed_out {
            self.timed_out.fetch_add(1, Ordering::Relaxed);
        }
        if allow {
            self.allowed.fetch_add(1, Ordering::Relaxed);
        } else {
            self.denied.fetch_add(1, Ordering::Relaxed);
        }

        // The kernel handed us this descriptor; leaking it exhausts the
        // process fd table and wedges us just as surely as not answering.
        // SAFETY: we own this fd and have answered for it.
        unsafe { libc::close(fd) };
    }

    /// `(allowed, denied, timed_out)`.
    pub fn counters(&self) -> (u64, u64, u64) {
        (
            self.allowed.load(Ordering::Relaxed),
            self.denied.load(Ordering::Relaxed),
            self.timed_out.load(Ordering::Relaxed),
        )
    }
}

// ── the syscall layer ───────────────────────────────────────────────────

/// An initialised fanotify group.
pub struct Gate {
    responder: Arc<Responder>,
    /// Events registered but not yet answered: `(seq, fd, deadline)`.
    inflight: Arc<Mutex<Vec<(u64, RawFd, Instant)>>>,
    running: Arc<AtomicBool>,
    /// Source of event identities. Monotonic, never reused.
    seq: Arc<AtomicU64>,
}

impl Gate {
    /// Open a fanotify group in permission mode.
    ///
    /// Fails with `EPERM` without `CAP_SYS_ADMIN`, which is the expected
    /// outcome for an unprivileged run and must be handled by degrading
    /// rather than by dying.
    pub fn init() -> io::Result<Self> {
        // SAFETY: plain syscall with constant arguments.
        let fd = unsafe {
            libc::fanotify_init(
                FAN_CLOEXEC | FAN_NONBLOCK | FAN_CLASS_CONTENT,
                (libc::O_RDONLY | libc::O_LARGEFILE) as u32,
            )
        };
        if fd < 0 {
            return Err(io::Error::last_os_error());
        }
        // SAFETY: fanotify_init returned a valid owned descriptor.
        let fan = unsafe { OwnedFd::from_raw_fd(fd) };
        Ok(Self {
            responder: Arc::new(Responder::new(fan)),
            inflight: Arc::new(Mutex::new(Vec::new())),
            running: Arc::new(AtomicBool::new(false)),
            seq: Arc::new(AtomicU64::new(1)),
        })
    }

    pub fn responder(&self) -> Arc<Responder> {
        Arc::clone(&self.responder)
    }

    /// Watch a mount, by its vfsmount.
    ///
    /// **Only safe when the watcher shares a mount namespace with the
    /// processes it is watching.** A vfsmount is per-namespace, so a
    /// daemon in a private namespace marks its own copy and receives
    /// nothing from anybody else — while still reporting itself armed.
    /// Under systemd that is the default situation, because several
    /// ordinary hardening options force a private mount namespace.
    ///
    /// Kept for the smoke-test example, which runs outside systemd.
    /// Production uses [`Gate::mark_filesystem`].
    pub fn mark_mount(&self, path: &Path) -> io::Result<()> {
        self.mark(path, FAN_MARK_ADD | FAN_MARK_MOUNT)
    }

    /// Watch an entire filesystem, by its superblock.
    ///
    /// A superblock is shared across mount namespaces, so this sees every
    /// process on the machine regardless of where the watcher lives. It
    /// also still scopes correctly: a superblock is exactly one
    /// filesystem, so marking a dedicated mount covers that and nothing
    /// else. This is the production mark, and the reason the watchdog is
    /// not optional.
    pub fn mark_filesystem(&self, path: &Path) -> io::Result<()> {
        self.mark(path, FAN_MARK_ADD | FAN_MARK_FILESYSTEM)
    }

    fn mark(&self, path: &Path, flags: u32) -> io::Result<()> {
        let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path contains NUL"))?;
        // SAFETY: fd is valid, path is a NUL-terminated C string that
        // outlives the call.
        let rc = unsafe {
            libc::fanotify_mark(
                self.responder.fan.as_raw_fd(),
                flags,
                // FAN_OPEN_PERM is deliberately NOT requested.
                //
                // It fires on every open of every file on the filesystem,
                // which is an enormous event volume for no protective
                // value we can actually use: we do not block reads (see
                // the deny-only-on-exec note in `serve`), so each one is a
                // read, a scan and an allow. On a live server that meant
                // re-scanning a multi-megabyte agent transcript on every
                // open, which is what pushed an event past the watchdog
                // deadline within seconds of arming.
                //
                // What remains covers the threat: execution is refused
                // before it happens, and anything malicious written to
                // disk is quarantined when the write completes. An
                // interpreted script is caught as it lands rather than as
                // it is read, which is the same protection one step
                // earlier.
                FAN_OPEN_EXEC_PERM | FAN_CLOSE_WRITE,
                libc::AT_FDCWD,
                c_path.as_ptr(),
            )
        };
        if rc < 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    /// Wait until the group has events, or `timeout_ms` elapses.
    ///
    /// The reader used to poll: read, get `EAGAIN`, sleep a millisecond,
    /// try again. That millisecond lands on the critical path of every
    /// `execve` on a watched filesystem — measured at +2.7 ms per exec on
    /// this machine, nearly all of it waiting for a sleep to finish rather
    /// than doing any work. Blocking on the descriptor means the reader
    /// wakes when the kernel has something to say and not before.
    ///
    /// The timeout exists only so the loop can notice `stop()`.
    pub fn wait_readable(&self, timeout_ms: i32) -> io::Result<bool> {
        let mut fds = libc::pollfd {
            fd: self.responder.fan.as_raw_fd(),
            events: libc::POLLIN,
            revents: 0,
        };
        // SAFETY: one valid descriptor, and the struct outlives the call.
        let rc = unsafe { libc::poll(&mut fds, 1, timeout_ms) };
        if rc < 0 {
            let err = io::Error::last_os_error();
            // A signal interrupting the wait is not an error.
            if err.kind() == io::ErrorKind::Interrupted {
                return Ok(false);
            }
            return Err(err);
        }
        Ok(rc > 0 && fds.revents & libc::POLLIN != 0)
    }

    /// Read whatever events are pending. Returns empty on `EAGAIN`.
    pub fn read_events(&self) -> io::Result<Vec<Event>> {
        let mut buf = [0u8; 8192];
        // SAFETY: reading into a buffer we own, bounded by its length.
        let n = unsafe {
            libc::read(
                self.responder.fan.as_raw_fd(),
                buf.as_mut_ptr() as *mut libc::c_void,
                buf.len(),
            )
        };
        if n < 0 {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::WouldBlock {
                return Ok(Vec::new());
            }
            return Err(err);
        }
        let mut events = decode(&buf[..n as usize]);
        for e in &mut events {
            e.seq = self.seq.fetch_add(1, Ordering::Relaxed);
        }
        if std::env::var_os("HOUNDD_GATE_DEBUG").is_some() {
            for e in &events {
                eprintln!(
                    "gate/debug: fd={} pid={} mask={:#x} exec={} write={} perm={} path={:?}",
                    e.fd,
                    e.pid,
                    e.mask,
                    e.is_exec(),
                    e.is_write(),
                    e.needs_response(),
                    e.path()
                );
            }
        }
        // Register before returning so the watchdog can adopt an event even
        // if the caller panics on the way to scanning it.
        {
            let deadline = Instant::now() + DEADLINE;
            let mut inflight = self.inflight.lock().expect("inflight poisoned");
            for e in &events {
                // Only permission events go to the watchdog. Handing it a
                // notification would make it write a response nobody asked
                // for.
                if e.needs_response() {
                    inflight.push((e.seq, e.fd, deadline));
                }
            }
        }
        Ok(events)
    }

    /// Start the fail-open watchdog.
    ///
    /// This is the safety net the whole design rests on: anything still
    /// unanswered past its deadline is allowed through, and the daemon
    /// records that it happened.
    pub fn start_watchdog(&self) -> std::thread::JoinHandle<()> {
        self.running.store(true, Ordering::SeqCst);
        let responder = Arc::clone(&self.responder);
        let inflight = Arc::clone(&self.inflight);
        let running = Arc::clone(&self.running);
        std::thread::Builder::new()
            .name("hound-gate-watchdog".into())
            .spawn(move || {
                while running.load(Ordering::SeqCst) {
                    std::thread::sleep(Duration::from_millis(50));
                    let now = Instant::now();
                    let mut expired: Vec<(u64, RawFd)> = Vec::new();
                    {
                        let mut list = inflight.lock().expect("inflight poisoned");
                        list.retain(|(seq, fd, deadline)| {
                            if now >= *deadline {
                                expired.push((*seq, *fd));
                                false
                            } else {
                                true
                            }
                        });
                    }
                    for (seq, fd) in expired {
                        // Fail OPEN. Always.
                        responder.answer(seq, fd, true, true);
                    }
                }
            })
            .expect("spawning the watchdog must not fail")
    }

    /// Run the gate: one reader thread, `workers` scanner threads.
    ///
    /// The split is a safety requirement, not a throughput one. The
    /// watchdog can only rescue events it has been told about, and it is
    /// told about them in [`Gate::read_events`]. If a slow scan runs on the
    /// same thread that drains the queue, every event behind it is invisible
    /// to the watchdog and the deadline means nothing. Keeping the reader
    /// free means every event is registered within microseconds of arriving,
    /// whatever the scanners are doing.
    pub fn serve(
        self: &Arc<Self>,
        cfg: GateConfig,
        scan: Arc<dyn Fn(&Path, &[u8]) -> Option<String> + Send + Sync>,
        on_detect: Arc<dyn Fn(&Path, &str, Verdict) + Send + Sync>,
        fast: Arc<dyn Fn(&Event) -> Option<bool> + Send + Sync>,
    ) -> Vec<std::thread::JoinHandle<()>> {
        let (tx, rx) = std::sync::mpsc::channel::<Event>();
        let rx = Arc::new(Mutex::new(rx));
        let mut handles = Vec::new();

        // Reader: drains the kernel queue and never blocks on a scan.
        {
            let gate = Arc::clone(self);
            let running = Arc::clone(&self.running);
            handles.push(
                std::thread::Builder::new()
                    .name("hound-gate-reader".into())
                    .spawn(move || {
                        while running.load(Ordering::SeqCst) {
                            // Block until the kernel has something, so no
                            // exec ever waits on a timer of ours.
                            match gate.wait_readable(100) {
                                Ok(false) => continue,
                                Ok(true) => {}
                                Err(e) => {
                                    eprintln!("gate: poll failed: {e}");
                                    std::thread::sleep(Duration::from_millis(20));
                                    continue;
                                }
                            }
                            match gate.read_events() {
                                Ok(events) if events.is_empty() => {}
                                Ok(events) => {
                                    for e in events {
                                        let (seq, fd) = (e.seq, e.fd);

                                        // Fast path: a verdict already in
                                        // memory is answered here, without
                                        // the channel handoff or a worker
                                        // wakeup. Re-executing a binary we
                                        // have already judged is the
                                        // overwhelmingly common event on a
                                        // busy machine, and it costs an
                                        // fstat and a map lookup — the
                                        // reader still never scans, which
                                        // is what keeps the watchdog
                                        // meaningful.
                                        if e.needs_response() {
                                            if let Some(allow) = fast(&e) {
                                                gate.answer_and_retire(seq, fd, allow);
                                                continue;
                                            }
                                        }

                                        if tx.send(e).is_err() {
                                            // No workers left: fail open.
                                            gate.answer_and_retire(seq, fd, true);
                                        }
                                    }
                                }
                                Err(e) => {
                                    eprintln!("gate: read failed: {e}");
                                    std::thread::sleep(Duration::from_millis(20));
                                }
                            }
                        }
                    })
                    .expect("spawning the gate reader must not fail"),
            );
        }

        let our_pid = std::process::id() as i32;
        for n in 0..cfg.workers.max(1) {
            let gate = Arc::clone(self);
            let rx = Arc::clone(&rx);
            let scan = Arc::clone(&scan);
            let on_detect = Arc::clone(&on_detect);
            let running = Arc::clone(&self.running);
            let excludes = cfg.excludes.clone();
            let max_size = cfg.max_size;
            handles.push(
                std::thread::Builder::new()
                    .name(format!("hound-gate-{n}"))
                    .spawn(move || {
                        while running.load(Ordering::SeqCst) {
                            let event = {
                                let rx = rx.lock().expect("gate channel poisoned");
                                rx.recv_timeout(Duration::from_millis(50))
                            };
                            let Ok(event) = event else { continue };

                            let path = event.path();
                            let decision = policy::decide(
                                event.pid,
                                our_pid,
                                path.as_deref(),
                                event.size(),
                                max_size,
                                &excludes,
                            );

                            let hit = match decision {
                                policy::Decision::AllowNow(_) => None,
                                policy::Decision::Scan => path.as_deref().and_then(|p| {
                                    event.content(max_size as usize).and_then(|b| scan(p, &b))
                                }),
                            };

                            if std::env::var_os("HOUNDD_GATE_DEBUG").is_some() {
                                eprintln!(
                                    "gate/debug: seq={} decision={:?} hit={:?} path={:?} size={:?}",
                                    event.seq, decision, hit, path, event.size()
                                );
                            }

                            if event.needs_response() {
                                // DENY ONLY ON EXECUTION.
                                //
                                // A permission event covers both execve and
                                // an ordinary open. Denying the open means a
                                // matching file cannot be READ by anything —
                                // which on a live server blocked reads of an
                                // AI agent's own transcript and came within a
                                // whisker of breaking the session doing the
                                // testing. Blocking execution is the feature;
                                // blocking every read of a matching file is a
                                // denial of service against the operator.
                                //
                                // A read that matches is still reported, and
                                // the write path still quarantines. The file
                                // simply is not held hostage.
                                let deny = hit.is_some() && event.is_exec();
                                gate.answer_and_retire(event.seq, event.fd, !deny);
                                if let (Some(name), Some(p)) = (&hit, path.as_deref()) {
                                    on_detect(
                                        p,
                                        name,
                                        if deny { Verdict::Blocked } else { Verdict::Seen },
                                    );
                                }
                            } else {
                                // A completed write: nothing is waiting, so
                                // close the descriptor and hand it to the
                                // daemon to quarantine.
                                gate.responder().close_only(event.seq, event.fd);
                                if let (Some(name), Some(p)) = (&hit, path.as_deref()) {
                                    on_detect(p, name, Verdict::Written);
                                }
                            }
                        }
                    })
                    .expect("spawning a gate worker must not fail"),
            );
        }
        handles
    }

    /// Stop the watchdog tracking an event.
    pub fn retire(&self, seq: u64) {
        let mut list = self.inflight.lock().expect("inflight poisoned");
        list.retain(|(s, _, _)| *s != seq);
    }

    pub fn stop(&self) {
        self.running.store(false, Ordering::SeqCst);
    }

    /// Read whatever is pending and answer all of it. Returns
    /// `(handled, denied)`.
    ///
    /// `scan` is handed the file's **bytes**, already read from the event
    /// descriptor, and returns `Some(detection_name)` to block. It must not
    /// open the gated file itself — see [`Event::content`] for why that
    /// deadlocks.
    pub fn serve_once(
        &self,
        max_size: u64,
        excludes: &[String],
        scan: &dyn Fn(&Path, &[u8]) -> Option<String>,
    ) -> io::Result<(usize, usize)> {
        let our_pid = std::process::id() as i32;
        let events = self.read_events()?;
        let mut denied = 0usize;

        for event in &events {
            let path = event.path();
            let size = event.size();

            let decision = policy::decide(
                event.pid,
                our_pid,
                path.as_deref(),
                size,
                max_size,
                excludes,
            );

            let allow = match decision {
                policy::Decision::AllowNow(_) => true,
                policy::Decision::Scan => {
                    let verdict = path.as_deref().and_then(|p| {
                        event
                            .content(max_size as usize)
                            .and_then(|bytes| scan(p, &bytes))
                    });
                    match verdict {
                        Some(_name) => {
                            denied += 1;
                            false
                        }
                        None => true,
                    }
                }
            };

            self.answer_and_retire(event.seq, event.fd, allow);
        }
        Ok((events.len(), denied))
    }

    /// Answer an event and stop the watchdog tracking it.
    pub fn answer_and_retire(&self, seq: u64, fd: RawFd, allow: bool) {
        self.retire(seq);
        self.responder.answer(seq, fd, allow, false);
    }
}

/// How a detection reached us, which decides what the daemon does with it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verdict {
    /// Caught at `execve` and refused. The file never ran.
    Blocked,
    /// Caught after a completed write. It is on disk and should be
    /// quarantined — this is the path that replaces inotify.
    Written,
    /// Matched while being read rather than executed. Reported and
    /// allowed through: see the deny-only-on-exec note in `serve`.
    Seen,
}

/// How the gate should behave once running.
#[derive(Debug, Clone)]
pub struct GateConfig {
    /// Scanner threads. More absorbs bursts; each one is a file read.
    pub workers: usize,
    /// Files above this are allowed through unread.
    pub max_size: u64,
    /// Path prefixes never held for a verdict.
    pub excludes: Vec<String>,
}

/// Files above this are allowed through unread by the gate, whatever the
/// on-demand scan limit is.
///
/// A process is blocked while we decide, so the budget is a deadline
/// rather than a size: reading and matching 100 MB inline cannot finish
/// inside [`DEADLINE`], and every attempt is a watchdog rescue — a
/// process released unscanned, which is worse than never having looked.
pub const GATE_MAX_FILE_BYTES: u64 = 32 * 1024 * 1024;

impl Default for GateConfig {
    fn default() -> Self {
        Self {
            workers: 4,
            max_size: GATE_MAX_FILE_BYTES,
            excludes: vec!["/proc".into(), "/sys".into(), "/dev".into(), "/run".into()],
        }
    }
}

/// Decode a read buffer into events, skipping anything whose ABI version
/// we do not recognise.
fn decode(buf: &[u8]) -> Vec<Event> {
    let mut out = Vec::new();
    let mut offset = 0usize;

    while offset + METADATA_SIZE <= buf.len() {
        // SAFETY: bounds checked above; EventMetadata is repr(C) and POD.
        let meta: EventMetadata =
            unsafe { std::ptr::read_unaligned(buf[offset..].as_ptr() as *const EventMetadata) };

        let len = meta.event_len as usize;
        if len < METADATA_SIZE || offset + len > buf.len() {
            break;
        }
        // A version mismatch means the struct we just read may not mean
        // what we think. Stop rather than guess.
        if meta.vers != FAN_METADATA_VERSION {
            break;
        }
        if meta.fd >= 0 {
            out.push(Event {
                seq: 0, // stamped by read_events, which owns the counter
                fd: meta.fd,
                pid: meta.pid,
                mask: meta.mask,
            });
        }
        offset += len;
    }
    out
}

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

    fn ex(v: &[&str]) -> Vec<String> {
        v.iter().map(|s| s.to_string()).collect()
    }

    // ── the rule that prevents self-deadlock ──

    #[test]
    fn our_own_opens_are_always_allowed_first() {
        // Even with everything else pointing at Scan, our own pid wins.
        let d = decide(4242, 4242, Some(Path::new("/tmp/x")), Some(10), 100, &[]);
        assert_eq!(d, Decision::AllowNow("self"));
    }

    #[test]
    fn self_check_beats_exclusions_and_size() {
        let d = decide(7, 7, Some(Path::new("/usr/bin/ls")), Some(u64::MAX), 1, &ex(&["/usr"]));
        assert_eq!(d, Decision::AllowNow("self"));
    }

    // ── ordinary policy ──

    #[test]
    fn other_processes_get_scanned() {
        let d = decide(100, 7, Some(Path::new("/home/joe/a.bin")), Some(10), 100, &[]);
        assert_eq!(d, Decision::Scan);
    }

    #[test]
    fn unresolvable_paths_are_allowed_not_held() {
        let d = decide(100, 7, None, None, 100, &[]);
        assert_eq!(d, Decision::AllowNow("unresolvable"));
    }

    #[test]
    fn excluded_paths_are_allowed() {
        let d = decide(
            100,
            7,
            Some(Path::new("/var/lib/docker/overlay2/x")),
            Some(10),
            100,
            &ex(&["/var/lib/docker"]),
        );
        assert_eq!(d, Decision::AllowNow("excluded"));
    }

    #[test]
    fn oversized_files_are_allowed_not_held() {
        let d = decide(100, 7, Some(Path::new("/data/big.iso")), Some(5_000), 100, &[]);
        assert_eq!(d, Decision::AllowNow("oversized"));
    }

    #[test]
    fn exclusion_matches_whole_components_only() {
        assert!(is_excluded(Path::new("/var/lib/docker/x"), &ex(&["/var/lib/docker"])));
        assert!(
            !is_excluded(Path::new("/variable/thing"), &ex(&["/var"])),
            "/var must not swallow /variable"
        );
    }

    #[test]
    fn trailing_slashes_and_empties_are_tolerated() {
        assert!(is_excluded(Path::new("/proc/1"), &ex(&["/proc/"])));
        assert!(!is_excluded(Path::new("/proc/1"), &ex(&[""])));
    }

    // ── event decoding ──

    fn encode(vers: u8, mask: u64, fd: i32, pid: i32) -> Vec<u8> {
        let meta = EventMetadata {
            event_len: METADATA_SIZE as u32,
            vers,
            reserved: 0,
            metadata_len: METADATA_SIZE as u16,
            mask,
            fd,
            pid,
        };
        // SAFETY: reading a repr(C) POD struct as bytes.
        unsafe {
            std::slice::from_raw_parts(&meta as *const EventMetadata as *const u8, METADATA_SIZE)
        }
        .to_vec()
    }

    #[test]
    fn decodes_a_single_event() {
        let buf = encode(FAN_METADATA_VERSION, FAN_OPEN_EXEC_PERM, 9, 1234);
        let events = decode(&buf);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].fd, 9);
        assert_eq!(events[0].pid, 1234);
        assert!(events[0].is_exec());
    }

    #[test]
    fn decodes_several_events_in_one_read() {
        let mut buf = encode(FAN_METADATA_VERSION, FAN_OPEN_PERM, 3, 1);
        buf.extend(encode(FAN_METADATA_VERSION, FAN_OPEN_EXEC_PERM, 4, 2));
        let events = decode(&buf);
        assert_eq!(events.len(), 2);
        assert!(!events[0].is_exec(), "a plain open is not an exec");
        assert!(events[1].is_exec());
    }

    #[test]
    fn rejects_an_unknown_abi_version() {
        let buf = encode(99, FAN_OPEN_PERM, 3, 1);
        assert!(
            decode(&buf).is_empty(),
            "a struct we cannot interpret must be skipped, not guessed at"
        );
    }

    #[test]
    fn ignores_a_truncated_trailing_event() {
        let mut buf = encode(FAN_METADATA_VERSION, FAN_OPEN_PERM, 3, 1);
        buf.extend_from_slice(&[0u8; 7]); // not a whole record
        assert_eq!(decode(&buf).len(), 1);
    }

    #[test]
    fn skips_events_without_a_descriptor() {
        // FAN_NOFD (-1) shows up on queue overflow.
        let buf = encode(FAN_METADATA_VERSION, FAN_OPEN_PERM, -1, 1);
        assert!(decode(&buf).is_empty());
    }

    #[test]
    fn empty_read_decodes_to_nothing() {
        assert!(decode(&[]).is_empty());
    }

    // ── init behaviour ──

    #[test]
    fn init_without_privileges_fails_cleanly() {
        // Unprivileged: EPERM. Root: succeeds. Either is a pass — what
        // must never happen is a panic or a hang.
        match Gate::init() {
            Ok(gate) => {
                gate.stop();
                assert_eq!(gate.responder().counters(), (0, 0, 0));
            }
            Err(e) => assert!(
                matches!(e.kind(), io::ErrorKind::PermissionDenied)
                    || e.raw_os_error() == Some(libc::EPERM),
                "expected EPERM without CAP_SYS_ADMIN, got {e}"
            ),
        }
    }

    // ── reading from the descriptor, not the path ──

    fn event_for(path: &Path) -> (Event, std::fs::File) {
        let f = std::fs::File::open(path).unwrap();
        let fd = f.as_raw_fd();
        (Event { seq: 1, fd, pid: 1, mask: FAN_OPEN_PERM }, f)
    }

    #[test]
    fn content_reads_through_the_descriptor() {
        let p = std::env::temp_dir().join(format!("hound-fd-{}", std::process::id()));
        std::fs::write(&p, b"gate content").unwrap();
        let (e, _f) = event_for(&p);
        assert_eq!(e.content(1024).as_deref(), Some(&b"gate content"[..]));
        assert_eq!(e.size(), Some(12));
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn content_refuses_oversized_without_reading() {
        let p = std::env::temp_dir().join(format!("hound-fd-big-{}", std::process::id()));
        std::fs::write(&p, vec![7u8; 4096]).unwrap();
        let (e, _f) = event_for(&p);
        assert!(e.content(1024).is_none(), "must not read past the cap");
        assert_eq!(e.content(8192).map(|b| b.len()), Some(4096));
        let _ = std::fs::remove_file(&p);
    }

    #[test]
    fn content_does_not_disturb_the_file_offset() {
        // The gated process sees the file it opened, at the offset it
        // expects — pread, not read.
        let p = std::env::temp_dir().join(format!("hound-fd-off-{}", std::process::id()));
        std::fs::write(&p, b"abcdefgh").unwrap();
        let (e, f) = event_for(&p);
        let _ = e.content(1024).unwrap();
        let pos = unsafe { libc::lseek(f.as_raw_fd(), 0, libc::SEEK_CUR) };
        assert_eq!(pos, 0, "pread must leave the offset alone");
        let _ = std::fs::remove_file(&p);
    }

    // ── permission events versus notifications ──

    #[test]
    fn permission_events_must_be_answered() {
        for mask in [FAN_OPEN_PERM, FAN_OPEN_EXEC_PERM, FAN_OPEN_PERM | FAN_CLOSE_WRITE] {
            let e = Event { seq: 1, fd: 3, pid: 1, mask };
            assert!(e.needs_response(), "mask {mask:#x} has a process blocked on it");
        }
    }

    #[test]
    fn a_completed_write_must_not_be_answered() {
        let e = Event { seq: 1, fd: 3, pid: 1, mask: FAN_CLOSE_WRITE };
        assert!(!e.needs_response(), "answering a notification is a protocol error");
        assert!(e.is_write());
        assert!(!e.is_exec());
    }

    #[test]
    fn exec_and_write_are_distinguishable() {
        let exec = Event { seq: 1, fd: 3, pid: 1, mask: FAN_OPEN_EXEC_PERM };
        assert!(exec.is_exec() && !exec.is_write());
        let write = Event { seq: 2, fd: 4, pid: 1, mask: FAN_CLOSE_WRITE };
        assert!(write.is_write() && !write.is_exec());
    }

    #[test]
    fn the_watchdog_only_adopts_permission_events() {
        // A notification handed to the watchdog would have it write a
        // response for an event nobody is waiting on.
        let notify = Event { seq: 1, fd: 9, pid: 1, mask: FAN_CLOSE_WRITE };
        let perm = Event { seq: 2, fd: 10, pid: 1, mask: FAN_OPEN_EXEC_PERM };
        let adopted: Vec<RawFd> = [&notify, &perm]
            .iter()
            .filter(|e| e.needs_response())
            .map(|e| e.fd)
            .collect();
        assert_eq!(adopted, vec![10]);
    }

    #[test]
    fn decoding_preserves_the_write_bit() {
        let buf = encode(FAN_METADATA_VERSION, FAN_CLOSE_WRITE, 5, 77);
        let events = decode(&buf);
        assert_eq!(events.len(), 1);
        assert!(events[0].is_write());
        assert!(!events[0].needs_response());
    }

    #[test]
    fn the_same_fd_number_twice_is_two_events() {
        // The kernel recycles an event fd's NUMBER as soon as we close it,
        // so one write arrives as FAN_OPEN_PERM on fd 6 and then
        // FAN_CLOSE_WRITE on fd 6 again. Keying identity on the fd made the
        // second look like a duplicate and dropped it silently — the write
        // path appeared to work and quarantined nothing.
        let first = Event { seq: 1, fd: 6, pid: 99, mask: FAN_OPEN_PERM };
        let second = Event { seq: 2, fd: 6, pid: 99, mask: FAN_CLOSE_WRITE };
        assert_eq!(first.fd, second.fd, "the kernel really does reuse the number");
        assert_ne!(first.seq, second.seq, "identity must not come from the fd");
    }

    #[test]
    fn read_events_stamps_unique_sequence_numbers() {
        let Ok(gate) = Gate::init() else { return }; // needs root; skip otherwise
        let a = gate.seq.fetch_add(1, Ordering::Relaxed);
        let b = gate.seq.fetch_add(1, Ordering::Relaxed);
        assert_ne!(a, b);
        assert!(b > a, "sequence numbers must be monotonic");
        gate.stop();
    }

    #[test]
    fn decode_leaves_seq_for_read_events_to_stamp() {
        let buf = encode(FAN_METADATA_VERSION, FAN_OPEN_PERM, 3, 1);
        assert_eq!(decode(&buf)[0].seq, 0, "decode must not invent identities");
    }

    #[test]
    fn gate_config_defaults_exclude_pseudo_filesystems() {
        let cfg = GateConfig::default();
        assert!(cfg.workers >= 1);
        for d in ["/proc", "/sys", "/dev", "/run"] {
            assert!(
                is_excluded(Path::new(&format!("{d}/thing")), &cfg.excludes),
                "{d} must never be held for a verdict"
            );
        }
    }

    #[test]
    fn deadline_leaves_three_orders_of_magnitude_of_headroom() {
        // Scans measured at ~4ms. If this ever needs raising, the engine
        // regressed and that is the bug.
        assert!(DEADLINE >= Duration::from_millis(100));
        assert!(DEADLINE <= Duration::from_secs(2));
    }
}