cellos-supervisor 0.5.1

//! Linux-only: network namespace isolation + nftables policy (`CELLOS_SUBPROCESS_UNSHARE=net`).
//!
//! Tests verify:
//! 1. A cell with `CLONE_NEWNET` (`CELLOS_SUBPROCESS_UNSHARE=net`) runs successfully (lo available).
//! 2. The supervisor emits a `network_policy` CloudEvent to the JSONL sink.
//! 3. A cell with declared egress rules records them in the emitted event.
//! 4. After a net-isolated run, `network_enforcement` records nft apply outcome and command exit.
//! 5. A connection attempt to an external IP (RFC 5737 TEST-NET-1) cannot succeed in private netns.

#![cfg(target_os = "linux")]

mod linux {
    use std::fs::File;
    use std::io::{BufRead, BufReader, Write};
    use std::net::TcpListener;
    use std::path::{Path, PathBuf};
    use std::process::Command;

    /// Returns true when the running process can create a new network namespace.
    ///
    /// On github-hosted runners `unshare(CLONE_NEWNET)` fails with EPERM because
    /// the runner lacks CAP_SYS_ADMIN. Use this guard at the top of every test
    /// that sets CELLOS_SUBPROCESS_UNSHARE so they skip gracefully instead of
    /// failing with "Operation not permitted".
    fn unshare_net_available() -> bool {
        Command::new("unshare")
            .args(["-n", "/bin/true"])
            .status()
            .map(|s| s.success())
            .unwrap_or(false)
    }

    fn supervisor_exe() -> PathBuf {
        if let Some(p) = std::env::var_os("CARGO_BIN_EXE_cellos_supervisor") {
            return PathBuf::from(p);
        }
        let root = Path::new(env!("CARGO_MANIFEST_DIR"))
            .parent()
            .and_then(|p| p.parent())
            .expect("cellos-supervisor crate under workspace root");
        let profile = std::env::var("PROFILE").unwrap_or_else(|_| "debug".into());
        root.join("target").join(profile).join("cellos-supervisor")
    }

    /// Spec with `CLONE_NEWNET` active and no egress rules — verifies loopback is usable (ping lo).
    #[test]
    fn network_namespace_isolation_loopback_works() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        // Use `/bin/sh -c 'ping -c1 -W1 127.0.0.1 >/dev/null 2>&1'` to check loopback reachability.
        // Falls back gracefully if ping is absent: use /usr/bin/true.
        let json = r#"{
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "net-lo-test",
                "authority": {"secretRefs": []},
                "lifetime": {"ttlSeconds": 60},
                "run": {
"secretDelivery": "env","argv": ["/usr/bin/true"]}
            }
        }"#;
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        assert!(
            exe.is_file(),
            "supervisor binary missing at {}",
            exe.display()
        );

        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(
            status.success(),
            "supervisor with CLONE_NEWNET should succeed: {status:?}"
        );
    }

    /// Verify a run in a private network namespace cannot reach a listener bound on the host's
    /// loopback interface, even when it dials `127.0.0.1`.
    #[test]
    fn network_namespace_blocks_host_loopback_listener() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind host loopback listener");
        let port = listener.local_addr().expect("local addr").port();
        let argv_json = serde_json::to_string(&[
            "/usr/bin/python3",
            "-c",
            &format!(
                "import socket, sys\n\
                 s = socket.socket()\n\
                 s.settimeout(1.0)\n\
                 rc = s.connect_ex(('127.0.0.1', {port}))\n\
                 s.close()\n\
                 sys.exit(0 if rc != 0 else 1)\n"
            ),
        ])
        .expect("argv json");
        let json = format!(
            r#"{{
                "apiVersion": "cellos.io/v1",
                "kind": "ExecutionCell",
                "spec": {{
                    "id": "net-host-loopback-blocked",
                    "authority": {{"secretRefs": []}},
                    "lifetime": {{"ttlSeconds": 60}},
                    "run": {
"secretDelivery": "env",{"argv": {argv_json}}}
                }}
            }}"#
        );
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(
            status.success(),
            "run inside private netns should not reach host loopback listener: {status:?}"
        );
    }

    /// Verify the `network_policy` CloudEvent is emitted to the JSONL sink when
    /// `CELLOS_SUBPROCESS_UNSHARE` includes the `net` flag.
    #[test]
    fn network_policy_event_emitted_to_jsonl_sink() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        let jsonl_path = dir.path().join("events.jsonl");
        let json = r#"{
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "net-policy-event",
                "authority": {"secretRefs": []},
                "lifetime": {"ttlSeconds": 60},
                "run": {
"secretDelivery": "env","argv": ["/usr/bin/true"]}
            }
        }"#;
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .env("CELL_OS_JSONL_EVENTS", jsonl_path.as_os_str())
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(status.success(), "supervisor should succeed: {status:?}");
        assert!(jsonl_path.exists(), "JSONL sink file should exist");

        let file = File::open(&jsonl_path).expect("open jsonl");
        let lines: Vec<String> = BufReader::new(file)
            .lines()
            .map(|l| l.expect("read line"))
            .filter(|l| !l.trim().is_empty())
            .collect();

        let network_policy_events: Vec<serde_json::Value> = lines
            .iter()
            .filter_map(|l| serde_json::from_str(l).ok())
            .filter(|v: &serde_json::Value| {
                v["type"] == "dev.cellos.events.cell.observability.v1.network_policy"
            })
            .collect();

        assert!(
            !network_policy_events.is_empty(),
            "expected at least one network_policy event; got events: {:?}",
            lines
                .iter()
                .filter_map(|l| {
                    let v: serde_json::Value = serde_json::from_str(l).ok()?;
                    Some(v["type"].as_str().unwrap_or("?").to_string())
                })
                .collect::<Vec<_>>()
        );

        let ev = &network_policy_events[0];
        assert_eq!(ev["data"]["isolationMode"], "clone_newnet");
        assert_eq!(ev["data"]["cellId"], "net-policy-event");
        assert_eq!(ev["data"]["declaredEgressCount"], 0);
    }

    /// Verify that a connection attempt to an external IP cannot succeed inside a private network
    /// namespace. Uses RFC 5737 TEST-NET-1 (`192.0.2.1`) which is non-routable — the subprocess
    /// confirms `connect_ex` returns a non-zero error code (ENETUNREACH) and then exits 0.
    ///
    /// This is the L2-04 negative egress test: on the hardened path (`CLONE_NEWNET`), disallowed
    /// egress cannot succeed because the child has no routing to external hosts. If the connection
    /// somehow succeeded (rc == 0), the subprocess exits 1, causing the supervisor to exit
    /// non-zero, and the test assertion fails — surfacing the security violation.
    #[test]
    fn private_netns_blocks_external_ip_connection() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        // 192.0.2.1 is in TEST-NET-1 (RFC 5737) — reserved, non-routable documentation range.
        // In a private netns: no routing → connect_ex returns ENETUNREACH (errno != 0).
        // Subprocess exits 0 if blocked (expected), exits 1 if connected (security violation).
        let argv_json = serde_json::to_string(&[
            "/usr/bin/python3",
            "-c",
            "import socket, sys\n\
             s = socket.socket()\n\
             s.settimeout(1.0)\n\
             rc = s.connect_ex(('192.0.2.1', 80))\n\
             s.close()\n\
             sys.exit(0 if rc != 0 else 1)",
        ])
        .expect("argv json");
        let json = format!(
            r#"{{
                "apiVersion": "cellos.io/v1",
                "kind": "ExecutionCell",
                "spec": {{
                    "id": "net-egress-blocked",
                    "authority": {{"secretRefs": []}},
                    "lifetime": {{"ttlSeconds": 60}},
                    "run": {
"secretDelivery": "env",{"argv": {argv_json}}}
                }}
            }}"#
        );
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        assert!(
            exe.is_file(),
            "supervisor binary missing at {} — run `cargo build -p cellos-supervisor`",
            exe.display()
        );

        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(
            status.success(),
            "connection to 192.0.2.1 inside private netns should be blocked (ENETUNREACH); \
             supervisor exited with {status:?} — if this fails, disallowed egress may have succeeded"
        );
    }

    /// With declared egress rules, verify they appear in the `network_policy` event.
    #[test]
    fn network_policy_event_includes_egress_rules() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        let jsonl_path = dir.path().join("events.jsonl");
        let json = r#"{
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "net-egress-rules",
                "authority": {
                    "secretRefs": [],
                    "egressRules": [
                        {"host": "10.0.0.1", "port": 443, "protocol": "tcp"},
                        {"host": "10.0.0.2", "port": 80}
                    ]
                },
                "lifetime": {"ttlSeconds": 60},
                "run": {
"secretDelivery": "env","argv": ["/usr/bin/true"]}
            }
        }"#;
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .env("CELL_OS_JSONL_EVENTS", jsonl_path.as_os_str())
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(status.success(), "supervisor should succeed: {status:?}");

        let file = File::open(&jsonl_path).expect("open jsonl");
        let ev: serde_json::Value = BufReader::new(file)
            .lines()
            .map(|l| l.expect("read line"))
            .filter(|l| !l.trim().is_empty())
            .filter_map(|l| serde_json::from_str(&l).ok())
            .find(|v: &serde_json::Value| {
                v["type"] == "dev.cellos.events.cell.observability.v1.network_policy"
            })
            .expect("network_policy event not found");

        assert_eq!(ev["data"]["isolationMode"], "clone_newnet");
        assert_eq!(ev["data"]["declaredEgressCount"], 2);

        let egress = ev["data"]["declaredEgress"].as_array().expect("array");
        assert_eq!(egress.len(), 2);
        assert_eq!(egress[0]["host"], "10.0.0.1");
        assert_eq!(egress[0]["port"], 443);
        assert_eq!(egress[1]["host"], "10.0.0.2");
        assert_eq!(egress[1]["port"], 80);
    }

    /// Disallowed egress (not in declared allowlist) must fail; JSONL includes `network_enforcement`
    /// with deterministic nft / exit fields when `CLONE_NEWNET` is active (L2-04).
    #[test]
    fn nft_egress_disallowed_destination_fails_with_enforcement_event() {
        if !unshare_net_available() {
            return; // CAP_SYS_ADMIN unavailable (e.g. github-hosted runner)
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let spec_path = dir.path().join("spec.json");
        let jsonl_path = dir.path().join("events.jsonl");
        let argv_json = serde_json::to_string(&[
            "/usr/bin/python3",
            "-c",
            "import socket, sys\n\
             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n\
             s.settimeout(2.0)\n\
             rc = s.connect_ex(('192.0.2.2', 443))\n\
             s.close()\n\
             sys.exit(0 if rc != 0 else 1)\n",
        ])
        .expect("argv json");
        let json = format!(
            r#"{{
                "apiVersion": "cellos.io/v1",
                "kind": "ExecutionCell",
                "spec": {{
                    "id": "net-egress-block-test",
                    "authority": {{
                        "secretRefs": [],
                        "egressRules": [
                            {{"host": "192.0.2.1", "port": 443, "protocol": "tcp"}}
                        ]
                    }},
                    "lifetime": {{"ttlSeconds": 60}},
                    "run": {
"secretDelivery": "env",{"argv": {argv_json}}}
                }}
            }}"#
        );
        let mut f = File::create(&spec_path).expect("create spec");
        f.write_all(json.as_bytes()).expect("write spec");
        drop(f);

        let exe = supervisor_exe();
        let status = Command::new(exe)
            .env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .env("CELLOS_SUBPROCESS_UNSHARE", "net")
            .env("CELL_OS_JSONL_EVENTS", jsonl_path.as_os_str())
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(&spec_path)
            .status()
            .expect("spawn supervisor");

        assert!(
            !status.success(),
            "supervisor should fail when command exits non-zero: {status:?}"
        );

        let file = File::open(&jsonl_path).expect("open jsonl");
        let events: Vec<serde_json::Value> = BufReader::new(file)
            .lines()
            .map(|l| l.expect("read line"))
            .filter(|l| !l.trim().is_empty())
            .filter_map(|l| serde_json::from_str(&l).ok())
            .collect();

        let enforcement = events
            .iter()
            .find(|v| v["type"] == "dev.cellos.events.cell.observability.v1.network_enforcement")
            .expect("network_enforcement event not found");

        assert_eq!(enforcement["data"]["cellId"], "net-egress-block-test");
        assert_eq!(enforcement["data"]["declaredEgressRuleCount"], 1);
        assert_eq!(enforcement["data"]["isolationMode"], "clone_newnet");
        assert_ne!(
            enforcement["data"]["commandExitCode"].as_i64(),
            Some(0),
            "python should exit non-zero when connect is blocked or unreachable"
        );

        let nft_ok = enforcement["data"]["nftRulesApplied"].as_bool() == Some(true);
        if nft_ok {
            assert_eq!(
                enforcement["data"]["supplementaryEgressFilterActive"].as_bool(),
                Some(true)
            );
        }

        let completed = events
            .iter()
            .find(|v| v["type"] == "dev.cellos.events.cell.command.v1.completed")
            .expect("command.v1.completed not found");
        assert_ne!(completed["data"]["exitCode"].as_i64(), Some(0));
    }
}