//! Agents that run in their own user namespace.
//!
//! An agent's config sets `services.hyperhive.agent.user.privateUsers`; the
//! swarm builder records it, hive-c0re reads the record and sends the flag with
//! each start. The host's data owner (`dataOwner`, one uid:gid) comes from this
//! unit's environment. A namespaced agent starts with `--private-users=pick`, a
//! root idmapped rather than chowned, and every bind mounted `owneridmap`: the
//! owner of the bind's target inside the container, the agent user, is the data
//! owner on disk.
//!
//! Getting there moves the agent's data once. [`activate`] chowns its `state`,
//! `harness`, `work` and `claude` dirs to the data owner, never touching
//! anything else, and records that in a marker only root can write; the next
//! call sees the marker and does nothing. Every refusal that can be known up
//! front happens before the first chown, and the agent's own ids are written to
//! `<agent>.pending` before it, so a walk that fails midway is repeated from
//! those ids rather than read off a `state` dir that already changed hands.
//! `owneridmap` needs the target
//! directory to be owned by the agent when nspawn mounts, and nspawn creates a
//! missing one as root in a throwaway snapshot, so the mount points are made
//! in the template root too. The agent's socket dir lives on a tmpfs and goes
//! to the data owner at every start instead ([`owner_for_agent`]).

use std::ffi::CStr;
use std::os::fd::{AsFd, BorrowedFd, OwnedFd};
use std::os::unix::fs::lchown;
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{AGENT_STATE_ROOT, BindMount};
use rustix::fs::{AtFlags, Mode, OFlags};
use rustix::io::Errno;

use crate::files::SOCKET_DIR_ROOT;

const DATA_OWNER_ENV: &str = "HIVE_PRIV_USERNS_DATA_OWNER";

/// One `<owner> <uid> <gid>` file per agent: the data owner its data was moved
/// to, and the agent's own ids inside the container. A move that did not finish
/// leaves the same content in `<agent>.pending` beside it.
const MARKER_ROOT: &str = "/var/lib/hive-priv/userns";

/// NixOS gives a normal user, which the agent is, an id from 1000 up.
const MIN_AGENT_UID: u32 = 1000;

/// The store nspawn bind-mounts into every container; a namespaced container
/// mounts it idmapped.
const STORE: &str = "/nix/store";

/// The type of `statfs`'s `f_type`, which differs by target.
type FsType = libc::__fsword_t;

/// `f_type` of the filesystems that support idmapped mounts and that a store
/// can sit on: ext4, btrfs, xfs and erofs (`EROFS_SUPER_MAGIC_V1` in the
/// kernel's `linux/magic.h`; `libc` has no constant for it). Nothing else is
/// accepted. A namespaced agent ran idmapped binds on the VM test's erofs
/// store, so erofs idmaps on our kernel.
const IDMAP_FS: [FsType; 4] = [0xEF53, 0x9123_683E, 0x5846_5342, 0xE0F5_E1E2];

/// The dirs under `<AGENT_STATE_ROOT>/<agent>` the agent writes into. `config`
/// is not one of them: hive-core writes it and the agent only reads it. `state`
/// stays first: [`move_data`] opens it by name and opens the rest from `[1..]`.
const DATA_DIRS: [&str; 4] = ["state", "harness", "work", "claude"];

/// Where one call looks and writes, and what it was told.
pub(crate) struct Env<'a> {
    pub(crate) state_root: &'a Path,
    /// The agent's socket dir, a bind source like the data dirs.
    pub(crate) socket_dir: &'a Path,
    pub(crate) marker_root: &'a Path,
    pub(crate) rootfs: &'a Path,
    /// The data owner when the agent is to be namespaced, else `None`.
    pub(crate) owner: Option<u32>,
    pub(crate) stopped: bool,
    /// Lowest uid taken for an agent's own: below it a dir belongs to root or
    /// to a system user such as hive-core.
    pub(crate) min_uid: u32,
    /// Whether an agent's gid has to equal its uid: the id map copies the uid
    /// map for gids, so a group of another number has no mapping.
    pub(crate) same_gid: bool,
    /// `f_type` of the filesystem under [`STORE`].
    pub(crate) store_fs: &'a dyn Fn() -> std::io::Result<FsType>,
    /// Chown a mount point in the root template.
    pub(crate) chown: &'a dyn Fn(&Path, u32, u32) -> std::io::Result<()>,
    /// Chown the entry `name` of an open directory, following no link; `"."`
    /// is the directory itself. The only way the data walk changes ownership.
    pub(crate) chown_at: &'a ChownAt<'a>,
}

/// The signature of [`Env::chown_at`].
pub(crate) type ChownAt<'a> = dyn Fn(BorrowedFd<'_>, &CStr, u32, u32) -> std::io::Result<()> + 'a;

fn chown_at(dir: BorrowedFd<'_>, name: &CStr, uid: u32, gid: u32) -> std::io::Result<()> {
    Ok(rustix::fs::chownat(
        dir,
        name,
        Some(rustix::fs::Uid::from_raw(uid)),
        Some(rustix::fs::Gid::from_raw(gid)),
        AtFlags::SYMLINK_NOFOLLOW,
    )?)
}

/// `statfs(2)` `f_type` of `path`.
fn fs_type(path: &Path) -> std::io::Result<FsType> {
    use std::os::unix::ffi::OsStrExt as _;
    let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
    // SAFETY: `c_path` is a valid NUL-terminated string that outlives the
    // call; `statfs` only writes into the zero-initialised `buf`.
    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
    if unsafe { libc::statfs(c_path.as_ptr(), &raw mut buf) } != 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(buf.f_type)
}

/// Refuse a store whose filesystem is not ext4, btrfs, xfs or erofs. nspawn clones a
/// namespaced container's `/nix/store` bind with an idmap, which the kernel
/// refuses on other filesystems, so the container would not boot. Others are
/// refused even where a newer kernel could idmap them.
fn check_store(agent: &str, f_type: std::io::Result<FsType>) -> Result<()> {
    let f_type = f_type.with_context(|| format!("{agent}: statfs {STORE}"))?;
    if IDMAP_FS.contains(&f_type) {
        return Ok(());
    }
    bail!(
        "{agent} sets privateUsers, but {STORE} is on a filesystem (f_type {f_type:#x}) other \
         than ext4, btrfs, xfs or erofs, the only ones accepted; move {STORE} to one of them, or leave \
         privateUsers off for this agent"
    )
}

/// The data owner when `agent` is to run in a user namespace, else `None`.
///
/// `requested` is what hive-c0re sent from the closure record. `None` there
/// means it could not say, and the agent stays as it is: namespaced if
/// hive-priv already `moved` its data.
///
/// # Errors
/// The agent is to be namespaced and the data owner is missing or not a plain
/// uid.
pub(crate) fn owner_for(
    requested: Option<bool>,
    moved: bool,
    owner: Option<&str>,
    agent: &str,
) -> Result<Option<u32>> {
    if !requested.unwrap_or(moved) {
        return Ok(None);
    }
    let owner = owner
        .with_context(|| format!("{agent} sets privateUsers, but {DATA_OWNER_ENV} is not set"))?;
    let owner = owner
        .trim()
        .parse()
        .with_context(|| format!("{DATA_OWNER_ENV}={owner:?} is not a uid"))?;
    Ok(Some(owner))
}

/// [`owner_for`] under the unit's environment and the agent's marker.
///
/// # Errors
/// As [`owner_for`], or `agent` is not an agent name.
pub(crate) fn owner_for_agent(agent: &str, requested: Option<bool>) -> Result<Option<u32>> {
    crate::shared::validate_agent_name(agent)?;
    owner_for(
        requested,
        Path::new(MARKER_ROOT).join(agent).exists(),
        std::env::var(DATA_OWNER_ENV).ok().as_deref(),
        agent,
    )
}

/// Make `agent`'s next start of `container` namespaced when its config asks
/// for it, and say whether it is. `binds` are the ones the start will mount.
///
/// A container that is running, and whose data has not been moved yet, keeps
/// its current start: the move waits for a stopped container.
///
/// # Errors
/// The environment is inconsistent, the data owner changed since the data
/// moved, the agent stopped asking for a namespace after it moved, or a chown
/// or mkdir fails.
pub(crate) fn activate(
    container: &str,
    agent: &str,
    binds: &[BindMount],
    requested: Option<bool>,
) -> Result<bool> {
    let owner = owner_for_agent(agent, requested)?;
    let rootfs = crate::container_config::container_rootfs(container);
    let stopped = crate::rootfs::unit_is_stopped(container)?;
    activate_in(
        &Env {
            state_root: Path::new(AGENT_STATE_ROOT),
            socket_dir: &Path::new(SOCKET_DIR_ROOT).join(agent),
            marker_root: Path::new(MARKER_ROOT),
            rootfs: &rootfs,
            owner,
            stopped,
            min_uid: MIN_AGENT_UID,
            same_gid: true,
            store_fs: &|| fs_type(Path::new(STORE)),
            chown: &|path, uid, gid| lchown(path, Some(uid), Some(gid)),
            chown_at: &chown_at,
        },
        agent,
        binds,
    )
}

/// [`activate`] over an explicit [`Env`].
pub(crate) fn activate_in(env: &Env<'_>, agent: &str, binds: &[BindMount]) -> Result<bool> {
    crate::shared::validate_agent_name(agent)?;
    let marker = env.marker_root.join(agent);
    let pending = env.marker_root.join(format!("{agent}.pending"));
    let moved = read_marker(&marker)?;
    let Some(owner) = env.owner else {
        if moved.is_some() {
            bail!(
                "{agent}: its data was moved to the userns data owner and its config no longer \
                 sets privateUsers; set it again, or chown its dirs back and remove {}",
                marker.display()
            );
        }
        return Ok(false);
    };
    check_store(agent, (env.store_fs)())?;
    let ids = match moved {
        Some((was, uid, gid)) if was == owner => (uid, gid),
        Some((was, ..)) => {
            bail!("{agent}: its data is owned by {was}, not the configured data owner {owner}")
        }
        None if !env.stopped => return Ok(false),
        None => {
            let resume = read_marker(&pending)?.map(|(_, uid, gid)| (uid, gid));
            let ids = move_data(env, agent, owner, &pending, resume)?;
            write_marker(&marker, owner, ids)?;
            std::fs::remove_file(&pending)
                .with_context(|| format!("remove {}", pending.display()))?;
            ids
        }
    };
    if env.stopped {
        mount_points(env, agent, binds, ids)?;
    }
    Ok(true)
}

/// `<owner> <uid> <gid>` from the marker or pending file, `None` when there is
/// none.
fn read_marker(path: &Path) -> Result<Option<(u32, u32, u32)>> {
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e).with_context(|| format!("read {}", path.display())),
    };
    let ids: Vec<u32> = text
        .split_whitespace()
        .map(str::parse)
        .collect::<Result<_, _>>()
        .with_context(|| format!("parse {}", path.display()))?;
    let [owner, uid, gid] = ids[..] else {
        bail!("{} does not hold `<owner> <uid> <gid>`", path.display());
    };
    Ok(Some((owner, uid, gid)))
}

fn write_marker(path: &Path, owner: u32, (uid, gid): (u32, u32)) -> Result<()> {
    if let Some(dir) = path.parent() {
        std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
    }
    std::fs::write(path, format!("{owner} {uid} {gid}\n"))
        .with_context(|| format!("write {}", path.display()))
}

/// Chown the agent's data dirs to `owner:owner`; returns the agent's own uid
/// and gid.
///
/// Every data dir is opened and the ids checked before the first chown, so a
/// dir that is a link or a file refuses while nothing has changed. The ids are
/// read from `state` and written to `pending` first; `resume` is the ids a
/// failed earlier walk left there, used as they are because `state` may already
/// belong to `owner`. Walking a dir again chowns it again to the same owner.
///
/// Only the [`DATA_DIRS`] under `<state_root>/<agent>` are touched, each only
/// if it is a real directory. Every step is relative to an open directory fd
/// and follows no link (see [`chown_tree`]), because an agent that holds the
/// dirs, a manager for one, can write them while this runs as root. The
/// container itself is stopped.
fn move_data(
    env: &Env<'_>,
    agent: &str,
    owner: u32,
    pending: &Path,
    resume: Option<(u32, u32)>,
) -> Result<(u32, u32)> {
    let base_path = env.state_root.join(agent);
    let base = open_base(&base_path)?;
    let state = open_dir_at(&base, "state")
        .with_context(|| format!("open {}", base_path.join("state").display()))?
        .with_context(|| format!("{} is not a directory", base_path.join("state").display()))?;
    let mut dirs = vec![state];
    for sub in &DATA_DIRS[1..] {
        match open_dir_at(&base, *sub) {
            Ok(Some(dir)) => dirs.push(dir),
            Ok(None) => bail!("{} is not a directory", base_path.join(sub).display()),
            Err(e) if is_not_found(&e) => {}
            Err(e) => return Err(e.context(format!("open {sub}"))),
        }
    }
    let ids = if let Some(ids) = resume {
        ids
    } else {
        let st = rustix::fs::fstat(&dirs[0]).context("stat state")?;
        let ids = (st.st_uid, st.st_gid);
        check_ids(agent, ids, owner, (env.min_uid, env.same_gid))?;
        write_marker(pending, owner, ids)?;
        ids
    };
    for dir in &dirs {
        chown_tree(env, dir, owner, 0)?;
    }
    Ok(ids)
}

/// Refuse an agent whose `state` dir is not owned by a container user, or whose
/// gid differs from its uid when `same_gid` asks for them to match.
///
/// The ids come from the dir's owner, which the agent's own root can set, so
/// this catches a misconfigured pin, not a hostile agent.
fn check_ids(
    agent: &str,
    (uid, gid): (u32, u32),
    owner: u32,
    (min_uid, same_gid): (u32, bool),
) -> Result<()> {
    if uid < min_uid || uid == owner {
        bail!(
            "{agent}: its state dir is owned by uid {uid}, not a container user; start the \
             agent once without the namespace so its user exists"
        );
    }
    if same_gid && gid != uid {
        bail!(
            "{agent}: its state dir is owned by {uid}:{gid}, and a namespaced agent's gid has \
             to equal its uid; pin services.hyperhive.agent.user.uid and .gid to {uid} in its \
             config and start it once before listing it"
        );
    }
    Ok(())
}

/// The deepest directory [`chown_tree`] enters. It holds one fd per level, and
/// an agent can make a tree as deep as it likes.
const MAX_DEPTH: u32 = 256;

/// Open `path` as a directory, following no link in its last component.
fn open_base(path: &Path) -> Result<OwnedFd> {
    rustix::fs::open(
        path,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        Mode::empty(),
    )
    .with_context(|| format!("open {}", path.display()))
}

/// Open `name` under `dir` as a directory without following a link. `None`
/// when the entry is something else, a link included; an error when it is
/// missing or unreadable.
fn open_dir_at(dir: impl AsFd, name: impl rustix::path::Arg) -> Result<Option<OwnedFd>> {
    match rustix::fs::openat(
        dir,
        name,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        Mode::empty(),
    ) {
        Ok(fd) => Ok(Some(fd)),
        Err(Errno::NOTDIR | Errno::LOOP) => Ok(None),
        Err(e) => Err(std::io::Error::from(e).into()),
    }
}

/// Chown the directory `top` is open on, and everything under it, to
/// `owner:owner`.
///
/// The walk never names a path. Each directory is entered with
/// `openat(parent_fd, name, O_NOFOLLOW|O_DIRECTORY)` and each other entry is
/// chowned through `chown_at` on its parent's fd, which does not follow a link.
/// A directory swapped for a link while this runs is therefore chowned as the
/// link, or skipped, and never followed. A directory on another filesystem than
/// `top` is left alone.
fn chown_tree(env: &Env<'_>, top: &OwnedFd, owner: u32, depth: u32) -> Result<()> {
    if depth > MAX_DEPTH {
        bail!("a data dir is nested more than {MAX_DEPTH} levels deep");
    }
    let dev = rustix::fs::fstat(top)?.st_dev;
    (env.chown_at)(top.as_fd(), c".", owner, owner).context("chown a data dir")?;
    for entry in rustix::fs::Dir::read_from(top)? {
        let entry = entry?;
        let name = entry.file_name();
        if matches!(name.to_bytes(), b"." | b"..") {
            continue;
        }
        match open_dir_at(top, name) {
            Ok(Some(dir)) => {
                if rustix::fs::fstat(&dir)?.st_dev == dev {
                    chown_tree(env, &dir, owner, depth + 1)?;
                }
            }
            Ok(None) => (env.chown_at)(top.as_fd(), name, owner, owner)
                .with_context(|| format!("chown {}", name.to_string_lossy()))?,
            // Removed since it was listed.
            Err(e) if is_not_found(&e) => {}
            Err(e) => return Err(e.context(format!("open {}", name.to_string_lossy()))),
        }
    }
    Ok(())
}

fn is_not_found(e: &anyhow::Error) -> bool {
    e.downcast_ref::<std::io::Error>()
        .is_some_and(|e| e.kind() == std::io::ErrorKind::NotFound)
}

/// Create, in the template root, the mount point of each bind whose source is
/// one of the agent's data dirs or its socket dir, owned by the agent's ids
/// inside the container.
fn mount_points(
    env: &Env<'_>,
    agent: &str,
    binds: &[BindMount],
    (uid, gid): (u32, u32),
) -> Result<()> {
    let sources: Vec<PathBuf> = DATA_DIRS
        .iter()
        .map(|sub| env.state_root.join(agent).join(sub))
        .chain([env.socket_dir.to_path_buf()])
        .collect();
    for bind in binds.iter().filter(|b| {
        sources
            .iter()
            .any(|s| s.as_os_str() == b.host_path.as_str())
    }) {
        let parts: Vec<&str> = bind
            .container_path
            .split('/')
            .filter(|p| !p.is_empty())
            .collect();
        let Some(dir) = crate::rootfs::own_dir(env.rootfs, &parts, uid, gid)? else {
            bail!(
                "{}: a link or file where the mount point {} goes",
                env.rootfs.display(),
                bind.container_path
            );
        };
        (env.chown)(&dir, uid, gid).with_context(|| format!("chown {}", dir.display()))?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{Env, activate_in, check_ids, check_store, owner_for};
    use crate::shared::test_support::scratch;
    use hive_priv_sock::BindMount;
    use std::cell::RefCell;
    use std::os::fd::AsRawFd as _;
    use std::os::unix::fs::MetadataExt as _;
    use std::path::{Path, PathBuf};

    /// Not the uid the tests run as, so a chown is a change.
    const OWNER: u32 = 60000;

    fn write(path: &Path) {
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(path, "x").unwrap();
    }

    /// `<root>/agents/a` with data in each dir, `config` beside them, a link
    /// out of `work`, and `outside/o` next to the tree. Returns the root.
    fn tree() -> PathBuf {
        let root = scratch().canonicalize().unwrap();
        let a = root.join("agents/a");
        for f in ["state/f", "harness/d/g", "work/x", "claude/c", "config/cfg"] {
            write(&a.join(f));
        }
        write(&root.join("outside/o"));
        std::os::unix::fs::symlink(root.join("outside"), a.join("work/link")).unwrap();
        root
    }

    /// Run [`activate_in`] on `root` and return its answer with every path it
    /// asked to chown, relative to `root`. The walk's chowns arrive as an open
    /// directory and a bare name; the name is checked to hold no `/`, and the
    /// path is rebuilt from what the fd points at.
    fn run_with(
        root: &Path,
        owner: Option<u32>,
        stopped: bool,
        agent: &str,
        binds: &[BindMount],
        ids: (u32, bool),
    ) -> (anyhow::Result<bool>, Vec<String>) {
        run_failing(root, owner, stopped, agent, binds, ids, None)
    }

    /// [`run_with`], where the chown of the path `fail` (relative to `root`)
    /// fails after it is recorded.
    fn run_failing(
        root: &Path,
        owner: Option<u32>,
        stopped: bool,
        agent: &str,
        binds: &[BindMount],
        (min_uid, same_gid): (u32, bool),
        fail: Option<&str>,
    ) -> (anyhow::Result<bool>, Vec<String>) {
        let seen = RefCell::new(Vec::new());
        let rel = |p: &Path| p.strip_prefix(root).unwrap().display().to_string();
        let chown = |p: &Path, _: u32, _: u32| {
            seen.borrow_mut().push(rel(p));
            Ok(())
        };
        let chown_at = |fd: std::os::fd::BorrowedFd<'_>, name: &std::ffi::CStr, _: u32, _: u32| {
            assert!(
                !name.to_bytes().contains(&b'/'),
                "{name:?} is not a bare name"
            );
            let dir = std::fs::read_link(format!("/proc/self/fd/{}", fd.as_raw_fd())).unwrap();
            let path = if name == c"." {
                dir
            } else {
                dir.join(name.to_str().unwrap())
            };
            seen.borrow_mut().push(rel(&path));
            if fail == Some(rel(&path).as_str()) {
                return Err(std::io::Error::other("injected"));
            }
            Ok(())
        };
        let (state_root, rootfs) = (root.join("agents"), root.join("rootfs"));
        std::fs::create_dir_all(&rootfs).unwrap();
        let env = Env {
            state_root: &state_root,
            socket_dir: &root.join("run/a"),
            marker_root: &root.join("marker"),
            rootfs: &rootfs,
            owner,
            stopped,
            min_uid,
            same_gid,
            store_fs: &|| Ok(0xEF53),
            chown: &chown,
            chown_at: &chown_at,
        };
        let answer = activate_in(&env, agent, binds);
        (answer, seen.into_inner())
    }

    fn run(
        root: &Path,
        owner: Option<u32>,
        stopped: bool,
        agent: &str,
        binds: &[BindMount],
    ) -> (anyhow::Result<bool>, Vec<String>) {
        run_with(root, owner, stopped, agent, binds, (0, false))
    }

    fn bind(root: &Path, sub: &str, target: &str) -> BindMount {
        BindMount {
            host_path: root.join("agents/a").join(sub).display().to_string(),
            container_path: target.to_owned(),
            read_only: false,
        }
    }

    /// The flag decides; with no word from the record the marker does. The
    /// refusal arms and the `None` arms are each other's controls.
    #[test]
    fn the_requested_flag_decides_and_without_it_the_marker_does() {
        let owner = |requested, moved, env| owner_for(requested, moved, env, "a");
        assert_eq!(
            owner(Some(true), false, Some("60000")).unwrap(),
            Some(60000)
        );
        assert_eq!(owner(Some(false), false, Some("60000")).unwrap(), None);
        assert_eq!(owner(Some(false), true, Some("60000")).unwrap(), None);
        assert_eq!(owner(None, true, Some("60000")).unwrap(), Some(60000));
        assert_eq!(owner(None, false, Some("60000")).unwrap(), None);
        assert!(owner(Some(true), false, None).is_err(), "no data owner");
        assert!(owner(Some(true), false, Some("root")).is_err(), "not a uid");
        assert!(
            owner(Some(false), false, None).is_ok(),
            "not asked, not needed"
        );
    }

    /// The move chowns the four data dirs and what is in them and nothing
    /// else: not `config`, and not what a link in `work` points at, which is
    /// chowned as the link. Every chown of the walk is fd-relative (checked in
    /// `run_with`).
    #[test]
    fn the_move_stays_inside_the_agents_data_dirs() {
        let root = tree();
        let (answer, mut seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        seen.sort();
        let want = [
            "agents/a/claude",
            "agents/a/claude/c",
            "agents/a/harness",
            "agents/a/harness/d",
            "agents/a/harness/d/g",
            "agents/a/state",
            "agents/a/state/f",
            "agents/a/work",
            "agents/a/work/link",
            "agents/a/work/x",
        ];
        assert_eq!(seen, want);
        assert!(root.join("marker/a").exists());
    }

    /// A link to a directory is never entered, whatever the listing said it
    /// was when it was read: nothing under the target is chowned, and the
    /// control is that the same walk chowns the real directories beside it.
    #[test]
    fn a_link_to_a_directory_is_not_entered() {
        let root = tree();
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        assert!(seen.contains(&"agents/a/harness/d/g".to_owned()), "control");
        assert!(seen.iter().all(|p| !p.starts_with("outside")), "{seen:?}");
    }

    /// A tree deeper than the walk's fd budget is refused, and nothing is
    /// recorded as moved.
    #[test]
    fn a_tree_nested_too_deep_is_refused() {
        let root = tree();
        std::fs::create_dir_all(root.join("agents/a/work").join("d/".repeat(300))).unwrap();
        let (answer, _) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(format!("{:#}", answer.unwrap_err()).contains("levels deep"));
        assert!(!root.join("marker/a").exists());
    }

    /// Once the marker is there a second call chowns nothing. The first call
    /// in `the_move_stays_inside_the_agents_data_dirs` is the control that
    /// the same setup does chown.
    #[test]
    fn the_move_runs_once() {
        let root = tree();
        assert!(run(&root, Some(OWNER), true, "a", &[]).0.unwrap());
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        assert!(seen.is_empty(), "{seen:?}");
    }

    /// A running container's data is left alone and the start stays as it is.
    #[test]
    fn a_running_container_is_not_moved() {
        let root = tree();
        let (answer, seen) = run(&root, Some(OWNER), false, "a", &[]);
        assert!(!answer.unwrap());
        assert!(seen.is_empty(), "{seen:?}");
        assert!(!root.join("marker/a").exists());
    }

    #[test]
    fn an_agent_without_the_flag_is_left_alone_and_a_moved_one_is_refused() {
        let root = tree();
        let (answer, seen) = run(&root, None, true, "a", &[]);
        assert!(!answer.unwrap() && seen.is_empty());
        assert!(
            run(&root, Some(OWNER), true, "a", &[]).0.unwrap(),
            "control"
        );
        assert!(run(&root, None, true, "a", &[]).0.is_err(), "left the list");
        assert!(
            run(&root, Some(OWNER + 1), true, "a", &[]).0.is_err(),
            "data owner changed"
        );
    }

    /// An idmap-capable store passes, any other is refused with its type and
    /// the fix before anything moves, and an agent without a namespace never
    /// looks.
    #[test]
    fn the_store_has_to_be_on_a_filesystem_that_can_be_idmapped() {
        for t in [0xEF53, 0x9123_683E, 0x5846_5342, 0xE0F5_E1E2] {
            check_store("a", Ok(t)).unwrap();
        }
        // overlayfs and tmpfs
        for t in [0x794c_7630, 0x0102_1994] {
            let msg = format!("{:#}", check_store("a", Ok(t)).unwrap_err());
            assert!(
                msg.contains(&format!("{t:#x}")) && msg.contains("ext4, btrfs, xfs or erofs"),
                "{msg}"
            );
        }
        assert!(check_store("a", Err(std::io::Error::other("x"))).is_err());

        let root = tree();
        let looked = std::cell::Cell::new(false);
        let tmpfs = || {
            looked.set(true);
            Ok(0x0102_1994)
        };
        let (state_root, rootfs) = (root.join("agents"), root.join("rootfs"));
        std::fs::create_dir_all(&rootfs).unwrap();
        let chown = |_: &Path, _: u32, _: u32| Ok(());
        let chown_at = |_: std::os::fd::BorrowedFd<'_>, _: &std::ffi::CStr, _: u32, _: u32| Ok(());
        let mut env = Env {
            state_root: &state_root,
            socket_dir: &root.join("run/a"),
            marker_root: &root.join("marker"),
            rootfs: &rootfs,
            owner: Some(OWNER),
            stopped: true,
            min_uid: 0,
            same_gid: false,
            store_fs: &tmpfs,
            chown: &chown,
            chown_at: &chown_at,
        };
        assert!(activate_in(&env, "a", &[]).is_err(), "opted in, bad store");
        assert!(!root.join("marker/a").exists(), "refused before the move");
        env.owner = None;
        looked.set(false);
        assert!(!activate_in(&env, "a", &[]).unwrap());
        assert!(!looked.get(), "an unnamespaced agent does not look");
    }

    /// A data dir that is a link is refused before anything past it is touched.
    #[test]
    fn a_linked_data_dir_is_refused() {
        let root = tree();
        let work = root.join("agents/a/work");
        std::fs::remove_dir_all(&work).unwrap();
        std::os::unix::fs::symlink(root.join("outside"), &work).unwrap();
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.is_err());
        assert!(seen.iter().all(|p| !p.starts_with("outside")), "{seen:?}");
        assert!(!root.join("marker/a").exists());
    }

    /// A refusal that is known before the walk leaves no trace: not one chown,
    /// no marker, no pending file. The `state` chown in the control run is what
    /// the refusal used to let through.
    #[test]
    fn a_refusal_before_the_walk_changes_nothing() {
        let root = tree();
        let work = root.join("agents/a/work");
        std::fs::remove_dir_all(&work).unwrap();
        std::os::unix::fs::symlink(root.join("outside"), &work).unwrap();
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.is_err());
        assert!(seen.is_empty(), "{seen:?}");
        assert!(!root.join("marker/a").exists());
        assert!(!root.join("marker/a.pending").exists());

        std::fs::remove_file(&work).unwrap();
        std::fs::create_dir(&work).unwrap();
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        assert!(seen.contains(&"agents/a/state".to_owned()), "control");
    }

    /// A walk that fails midway leaves the ids it read in `<agent>.pending`,
    /// and the next call walks again from them, without reading `state`'s
    /// owner: the minimum uid of the second call is above any real uid, which
    /// refuses a fresh move.
    #[test]
    fn a_walk_that_failed_is_resumed_from_the_pending_ids() {
        let root = tree();
        let uid = std::fs::metadata(root.join("agents/a/state"))
            .unwrap()
            .uid();
        let gid = std::fs::metadata(root.join("agents/a/state"))
            .unwrap()
            .gid();
        let (answer, seen) = run_failing(
            &root,
            Some(OWNER),
            true,
            "a",
            &[],
            (0, false),
            Some("agents/a/harness"),
        );
        assert!(format!("{:#}", answer.unwrap_err()).contains("injected"));
        assert!(seen.contains(&"agents/a/state".to_owned()), "state first");
        assert!(!root.join("marker/a").exists());
        let pending = std::fs::read_to_string(root.join("marker/a.pending")).unwrap();
        assert_eq!(pending, format!("{OWNER} {uid} {gid}\n"));

        let saved = std::fs::read_to_string(root.join("marker/a.pending")).unwrap();
        std::fs::remove_file(root.join("marker/a.pending")).unwrap();
        let (answer, _) = run_with(&root, Some(OWNER), true, "a", &[], (u32::MAX, false));
        assert!(answer.is_err(), "control: without pending, ids are checked");
        std::fs::write(root.join("marker/a.pending"), saved).unwrap();

        let (answer, seen) = run_with(&root, Some(OWNER), true, "a", &[], (u32::MAX, false));
        assert!(answer.unwrap());
        assert!(seen.contains(&"agents/a/work/x".to_owned()), "{seen:?}");
        assert_eq!(
            std::fs::read_to_string(root.join("marker/a")).unwrap(),
            pending
        );
        assert!(!root.join("marker/a.pending").exists());
    }

    /// A pending file that does not hold three ids is refused, named, before
    /// anything is chowned.
    #[test]
    fn a_pending_file_that_does_not_parse_is_refused() {
        let root = tree();
        for garbage in ["not ids\n", "60000 1000\n", ""] {
            std::fs::create_dir_all(root.join("marker")).unwrap();
            std::fs::write(root.join("marker/a.pending"), garbage).unwrap();
            let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
            let msg = format!("{:#}", answer.unwrap_err());
            assert!(msg.contains("a.pending"), "{msg}");
            assert!(seen.is_empty(), "{seen:?}");
            assert!(!root.join("marker/a").exists());
        }
    }

    /// `move_data` opens `state` by name and takes the other dirs from `DATA_DIRS[1..]`.
    #[test]
    fn state_is_the_first_data_dir() {
        assert_eq!(super::DATA_DIRS[0], "state");
    }

    #[test]
    fn a_path_is_not_an_agent_name() {
        let root = tree();
        let (answer, seen) = run(&root, Some(OWNER), true, "../outside", &[]);
        assert!(answer.is_err() && seen.is_empty());
    }

    /// The agent's own uid is read off its `state` dir, and root's or the
    /// data owner's is refused.
    #[test]
    fn the_state_dir_has_to_be_owned_by_a_container_user() {
        let root = tree();
        let euid = std::fs::metadata(root.join("agents/a/state"))
            .unwrap()
            .uid();
        let below = run_with(&root, Some(OWNER), true, "a", &[], (euid + 1, false));
        assert!(below.0.is_err() && below.1.is_empty(), "below the minimum");
        let owner = run_with(&root, Some(euid), true, "a", &[], (euid, false));
        assert!(owner.0.is_err(), "already the owner");
        let ok = run_with(&root, Some(OWNER), true, "a", &[], (euid, false));
        assert!(ok.0.unwrap(), "control");
    }

    /// The production `chown_at` acts on the link itself: a link to nothing is
    /// chowned, where following it would fail with ENOENT.
    #[test]
    fn chown_at_does_not_follow_a_link() {
        let root = scratch();
        std::os::unix::fs::symlink(root.join("missing"), root.join("l")).unwrap();
        let dir = super::open_base(&root).unwrap();
        let me = std::fs::metadata(&root).unwrap();
        super::chown_at(std::os::fd::AsFd::as_fd(&dir), c"l", me.uid(), me.gid()).unwrap();
    }

    /// A gid that differs from the uid is refused with the agent named, when
    /// asked for; the same ids pass without it. Through `activate_in` the
    /// answer follows whether the scratch dir's own ids match.
    #[test]
    fn a_gid_other_than_the_uid_is_refused() {
        let err = check_ids("juno", (1000, 999), OWNER, (1000, true)).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("juno") && msg.contains("1000:999"), "{msg}");
        check_ids("juno", (1000, 999), OWNER, (1000, false)).unwrap();
        check_ids("juno", (1000, 1000), OWNER, (1000, true)).unwrap();
        assert!(check_ids("juno", (0, 0), OWNER, (1000, true)).is_err());
        assert!(check_ids("juno", (OWNER, OWNER), OWNER, (1000, true)).is_err());

        let root = tree();
        let meta = std::fs::metadata(root.join("agents/a/state")).unwrap();
        let (answer, _) = run_with(&root, Some(OWNER), true, "a", &[], (0, true));
        assert_eq!(answer.is_ok(), meta.uid() == meta.gid());
    }

    /// A mount point is made for a bind of the agent's own data dir, and for
    /// no other bind; a link in the way is refused.
    #[test]
    fn mount_points_are_made_for_the_agents_data_binds_only() {
        let root = tree();
        let binds = [
            bind(&root, "state", "/agents/a/state"),
            BindMount {
                host_path: root.join("outside").display().to_string(),
                container_path: "/agents/a/stray".to_owned(),
                read_only: false,
            },
            BindMount {
                host_path: root.join("run/a").display().to_string(),
                container_path: "/run/hive-agent/a".to_owned(),
                read_only: false,
            },
        ];
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &binds);
        assert!(answer.unwrap());
        assert!(root.join("rootfs/agents/a/state").is_dir());
        assert!(root.join("rootfs/run/hive-agent/a").is_dir(), "socket dir");
        assert!(!root.join("rootfs/agents/a/stray").exists());
        assert!(
            seen.contains(&"rootfs/agents/a/state".to_owned()),
            "{seen:?}"
        );

        std::fs::remove_dir_all(root.join("rootfs/agents")).unwrap();
        std::os::unix::fs::symlink(root.join("outside"), root.join("rootfs/agents")).unwrap();
        assert!(run(&root, Some(OWNER), true, "a", &binds).0.is_err());
    }
}