//! E1-04 — FD inheritance + ambient-env regression for [`spawn_isolated_workload`].
//!
//! Skeptical-reviewer prompt: "even though the supervisor calls `env_clear()`, can a
//! determined operator still find that the workload child's environment carries a stray
//! variable from the supervisor process — or that the child inherited a numbered FD
//! pointing back at a NATS socket / broker handle / audit log?"
//!
//! This test plants two **ambient** environment variables in the parent (`AMBIENT_LEAK_*`),
//! plants two **declared** keys via [`WorkloadEnv`], then spawns a long-sleeping `/bin/sh`
//! workload via the public [`spawn_isolated_workload`] API. It then reads
//! `/proc/<child>/environ` and asserts byte-for-byte equality with the declared injection set.
//! If any ambient leak is present — or any declared key is missing — the test fails.
//!
//! On non-Linux targets this test is a stub that prints a skip note and returns. The
//! `/proc/<pid>/environ` surface is Linux-only.
//!
//! Conventions follow `break_attempts_post_isolation_residue.rs`: a small `doc_for`
//! fixture builder and a single `#[test]` per residue vector.

use cellos_core::ExecutionCellDocument;

/// Build a minimal `ExecutionCellDocument` with the given id (mirrors the
/// `doc_for` pattern from `break_attempts_post_isolation_residue.rs`).
#[allow(dead_code)]
fn doc_for(id: &str) -> ExecutionCellDocument {
    serde_json::from_value(serde_json::json!({
        "apiVersion": "cellos.io/v1",
        "kind": "ExecutionCell",
        "spec": {
            "id": id,
            "authority": { "secretRefs": ["FD_ISO_TOKEN"] },
            "lifetime": { "ttlSeconds": 60 }
        }
    }))
    .expect("fd_isolation: spec fixture must parse")
}

/// On Linux: spawn a child via the public spawn helper, then read
/// `/proc/<child>/environ` and assert it equals the declared injection set
/// exactly — no ambient leak, no missing declared key.
#[cfg(target_os = "linux")]
#[test]
fn child_environ_equals_declared_injection_set_no_ambient_leak() {
    use std::collections::BTreeMap;

    use cellos_host_cellos::{spawn_isolated_workload, WorkloadEnv};

    // Sanity: the doc fixture parses (mirrors the break-attempts suite).
    let _doc = doc_for("e1-04-fd-isolation");

    // Plant ambient env in the parent. If FD/env hygiene is broken these would surface
    // in /proc/<child>/environ.
    std::env::set_var("AMBIENT_LEAK_A", "supervisor-secret-A");
    std::env::set_var("AMBIENT_LEAK_B", "supervisor-secret-B");

    // Declared injection set — what the workload is *supposed* to see, and nothing else.
    let mut env = WorkloadEnv::new();
    env.push("DECLARED_TOKEN", "declared-value-1").unwrap();
    env.push("DECLARED_REGION", "us-west-2").unwrap();
    env.push("PATH", "/usr/bin:/bin").unwrap();

    // Direct `sleep 30` — long enough that we can read /proc/<pid>/environ.
    // Picking the binary path explicitly avoids relying on PATH lookup at fork-time and
    // sidesteps any shell-internal FDs that would confound the FD-walk test below.
    let sleep_bin = if std::path::Path::new("/usr/bin/sleep").exists() {
        "/usr/bin/sleep"
    } else if std::path::Path::new("/bin/sleep").exists() {
        "/bin/sleep"
    } else {
        eprintln!("[skip] no /usr/bin/sleep or /bin/sleep on this host");
        std::env::remove_var("AMBIENT_LEAK_A");
        std::env::remove_var("AMBIENT_LEAK_B");
        return;
    };
    let argv = vec![sleep_bin.to_string(), "30".to_string()];

    let mut spawned = spawn_isolated_workload(&argv, &env)
        .expect("E1-04: spawn_isolated_workload must succeed on Linux");
    let pid = spawned.pid();
    assert!(pid > 1, "child PID must be a real PID, got {pid}");

    // Read /proc/<pid>/environ. Format: NUL-separated KEY=VALUE pairs.
    // Retry briefly on EAGAIN — kernel may not have the environ vector ready in the
    // microsecond window right after fork+exec.
    let environ_path = format!("/proc/{pid}/environ");
    let mut raw: Vec<u8> = Vec::new();
    let mut last_err: Option<String> = None;
    let mut last_was_empty = false;
    for _attempt in 0..50 {
        match std::fs::read(&environ_path) {
            Ok(bytes) if !bytes.is_empty() => {
                raw = bytes;
                break;
            }
            Ok(_) => {
                last_was_empty = true;
                last_err = None;
            }
            Err(e) => {
                last_was_empty = false;
                last_err = Some(e.to_string());
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
    if raw.is_empty() {
        let _ = spawned.kill();
        let _ = spawned.wait();
        if let Some(e) = last_err {
            panic!("read {environ_path} failed after retries: {e}");
        }
        if last_was_empty {
            panic!("/proc/{pid}/environ was empty after 50 retries");
        }
        panic!("/proc/{pid}/environ never became readable (no error captured)");
    }

    // Parse NUL-separated KEY=VALUE pairs into a map (order doesn't matter for equality).
    let mut child_env: BTreeMap<String, String> = BTreeMap::new();
    for entry in raw.split(|b| *b == 0u8) {
        if entry.is_empty() {
            continue;
        }
        let s = match std::str::from_utf8(entry) {
            Ok(s) => s,
            Err(_) => {
                let _ = spawned.kill();
                let _ = spawned.wait();
                panic!("non-UTF8 entry in /proc/{pid}/environ: {entry:?}");
            }
        };
        if let Some((k, v)) = s.split_once('=') {
            child_env.insert(k.to_string(), v.to_string());
        } else {
            let _ = spawned.kill();
            let _ = spawned.wait();
            panic!("malformed environ entry (no '='): {s:?}");
        }
    }

    // Tear down child before assertions so we never leak even on failure.
    let _ = spawned.kill();
    let _ = spawned.wait();

    // Build the expected declared set.
    let mut expected: BTreeMap<String, String> = BTreeMap::new();
    expected.insert("DECLARED_TOKEN".to_string(), "declared-value-1".to_string());
    expected.insert("DECLARED_REGION".to_string(), "us-west-2".to_string());
    expected.insert("PATH".to_string(), "/usr/bin:/bin".to_string());

    // Hard assertion: child env == declared set, byte-for-byte.
    assert_eq!(
        child_env, expected,
        "child /proc/{pid}/environ must equal the declared injection set exactly\n\
         child_env  = {child_env:?}\n\
         expected   = {expected:?}"
    );

    // Defense in depth: explicitly reject the ambient leaks.
    assert!(
        !child_env.contains_key("AMBIENT_LEAK_A"),
        "AMBIENT_LEAK_A leaked into child env — env_clear() / FD hygiene is broken"
    );
    assert!(
        !child_env.contains_key("AMBIENT_LEAK_B"),
        "AMBIENT_LEAK_B leaked into child env — env_clear() / FD hygiene is broken"
    );

    // Clean up parent ambient env so other tests in the same process see a clean slate.
    std::env::remove_var("AMBIENT_LEAK_A");
    std::env::remove_var("AMBIENT_LEAK_B");
}

/// On Linux: prove the FD-close invariant directly. We open a sentinel pipe in the parent
/// (so its read end has fd > 2), spawn a child that runs `ls /proc/self/fd`, capture by
/// having the child write to the **declared** stdio (which we deliberately drop), and
/// instead probe the child's `/proc/<pid>/fd` directory before it execs anything fancy.
///
/// The simpler, race-free observation: count entries in `/proc/<child>/fd`. After execve
/// of `/bin/sh -c 'sleep 30'`, only fds 0/1/2 should remain. (`/proc/<pid>/fd` itself adds
/// a transient lookup fd while we read it, but that's the *reader's* fd, not the child's.)
#[cfg(target_os = "linux")]
#[test]
fn child_has_no_inherited_fds_beyond_stdio() {
    use cellos_host_cellos::{spawn_isolated_workload, WorkloadEnv};

    // Open a sentinel pipe in the parent. Without our pre_exec CLOEXEC walk, the read and
    // write ends would be inherited by the child as fds > 2.
    let mut pipe_fds: [libc::c_int; 2] = [-1, -1];
    let r = unsafe { libc::pipe(pipe_fds.as_mut_ptr()) };
    assert_eq!(r, 0, "pipe(2) failed: {}", std::io::Error::last_os_error());
    let sentinel_r = pipe_fds[0];
    let sentinel_w = pipe_fds[1];
    assert!(
        sentinel_r > 2,
        "sentinel pipe read fd must be > 2 to be observable"
    );
    assert!(
        sentinel_w > 2,
        "sentinel pipe write fd must be > 2 to be observable"
    );

    let mut env = WorkloadEnv::new();
    env.push("PATH", "/usr/bin:/bin").unwrap();

    // Pick whichever `sleep` binary exists on this distro (path varies between Debian/RHEL).
    let sleep_bin = if std::path::Path::new("/usr/bin/sleep").exists() {
        "/usr/bin/sleep"
    } else if std::path::Path::new("/bin/sleep").exists() {
        "/bin/sleep"
    } else {
        eprintln!("[skip] no /usr/bin/sleep or /bin/sleep on this host");
        unsafe {
            libc::close(sentinel_r);
            libc::close(sentinel_w);
        }
        return;
    };

    // Spawn `sleep 30` directly (no shell) so the only execve is the one we care about and
    // there are no shell-internal FDs to confound the count.
    let argv = vec![sleep_bin.to_string(), "30".to_string()];
    let mut spawned = spawn_isolated_workload(&argv, &env)
        .expect("E1-04 fd-walk: spawn_isolated_workload must succeed on Linux");
    let pid = spawned.pid();

    // Wait briefly for the child to reach `sleep` (post-execve).
    std::thread::sleep(std::time::Duration::from_millis(100));

    // Enumerate /proc/<pid>/fd and collect descriptor numbers.
    let fd_dir = format!("/proc/{pid}/fd");
    let mut child_fds: Vec<i32> = Vec::new();
    match std::fs::read_dir(&fd_dir) {
        Ok(dir) => {
            for entry in dir.flatten() {
                if let Ok(name) = entry.file_name().into_string() {
                    if let Ok(fd) = name.parse::<i32>() {
                        child_fds.push(fd);
                    }
                }
            }
        }
        Err(e) => {
            let _ = spawned.kill();
            let _ = spawned.wait();
            unsafe {
                libc::close(sentinel_r);
                libc::close(sentinel_w);
            }
            panic!("read_dir {fd_dir} failed: {e}");
        }
    }
    child_fds.sort();

    // Tear down child + close sentinel pipe before assertions.
    let _ = spawned.kill();
    let _ = spawned.wait();
    unsafe {
        libc::close(sentinel_r);
        libc::close(sentinel_w);
    }

    // The child must hold *only* the standard stdio fds {0, 1, 2}. Any extra fd is a
    // leak — the sentinel pipe ends, in particular, must not appear.
    let stdio: std::collections::BTreeSet<i32> = [0, 1, 2].into_iter().collect();
    let child_set: std::collections::BTreeSet<i32> = child_fds.iter().copied().collect();

    assert!(
        child_set.is_superset(&stdio),
        "child must keep stdio {{0,1,2}}, got {child_fds:?}"
    );

    let extras: Vec<i32> = child_set.difference(&stdio).copied().collect();
    assert!(
        extras.is_empty(),
        "child inherited fds beyond stdio — FD hygiene is broken. extras = {extras:?}, \
         child_fds = {child_fds:?}, sentinel = ({sentinel_r}, {sentinel_w})"
    );

    // Sentinel cross-check: neither sentinel fd appears in the child.
    assert!(
        !child_set.contains(&sentinel_r),
        "sentinel read fd {sentinel_r} leaked into child"
    );
    assert!(
        !child_set.contains(&sentinel_w),
        "sentinel write fd {sentinel_w} leaked into child"
    );
}

/// Non-Linux stub. `/proc/<pid>/environ` and `/proc/<pid>/fd` are Linux-only surfaces; the
/// E1-04 contract is still meaningful on other Unixes (the production code uses an
/// `RLIMIT_NOFILE`-bounded fallback in `pre_exec`), but this regression *test* needs
/// `/proc` to falsify the contract. Print a skip note instead.
#[cfg(not(target_os = "linux"))]
#[test]
fn child_environ_equals_declared_injection_set_no_ambient_leak() {
    eprintln!(
        "[skip] E1-04 fd_isolation: /proc/<pid>/environ is Linux-only; \
         spawn_isolated_workload contract still applies via the RLIMIT_NOFILE fallback."
    );
}

#[cfg(not(target_os = "linux"))]
#[test]
fn child_has_no_inherited_fds_beyond_stdio() {
    eprintln!(
        "[skip] E1-04 fd_isolation: /proc/<pid>/fd is Linux-only; \
         spawn_isolated_workload contract still applies via the RLIMIT_NOFILE fallback."
    );
}
